diff --git a/workflow/admin_recovery.go b/workflow/admin_recovery.go index 500fc48..7fbe79d 100644 --- a/workflow/admin_recovery.go +++ b/workflow/admin_recovery.go @@ -7,6 +7,7 @@ import ( "errors" "fmt" + "go.temporal.io/sdk/temporal" "go.temporal.io/sdk/workflow" "github.com/OpenNSW/core/shared/maputil" @@ -23,11 +24,26 @@ type terminalAdminError struct { func (e *terminalAdminError) Error() string { return e.err.Error() } func (e *terminalAdminError) Unwrap() error { return e.err } +// terminalAdminErrorType is the Go type name Temporal's default error converter stamps onto +// the ApplicationError it wraps a plain (non-Temporal) error in when that error crosses a +// child workflow boundary — e.g. a PARALLEL_SPLIT or BATCH_SPLIT branch's terminalAdminError, +// returned from the child, re-materializes in the parent as *temporal.ApplicationError with +// this Type(), not as a *terminalAdminError errors.As can find. Must match the unqualified +// type name of terminalAdminError exactly. +const terminalAdminErrorType = "terminalAdminError" + // isTerminalAdminError reports whether err has already been through parkNodeForAdmin and -// was deliberately given up on, meaning it should propagate without being parked again. +// was deliberately given up on, meaning it should propagate without being parked again. This +// checks both forms: a same-workflow error still holding its original Go type, and one that +// crossed a child workflow boundary (a spawned PARALLEL_SPLIT or BATCH_SPLIT branch) and so +// only carries the type name as a string — see terminalAdminErrorType. func isTerminalAdminError(err error) bool { var terminal *terminalAdminError - return errors.As(err, &terminal) + if errors.As(err, &terminal) { + return true + } + var appErr *temporal.ApplicationError + return errors.As(err, &appErr) && appErr.Type() == terminalAdminErrorType } // AdminResolutionAction describes how an admin chooses to resolve a node that is @@ -45,8 +61,7 @@ const ( AdminActionOverride AdminResolutionAction = "OVERRIDE" // AdminActionSkip marks the node completed without setting any variables. AdminActionSkip AdminResolutionAction = "SKIP" - // AdminActionAbort fails the node and the workflow with the original error — - // the same behavior the engine had before the escape hatch existed. + // AdminActionAbort fails the node and the workflow with the original error. AdminActionAbort AdminResolutionAction = "ABORT" ) diff --git a/workflow/admin_recovery_test.go b/workflow/admin_recovery_test.go index 05869fe..adf4e36 100644 --- a/workflow/admin_recovery_test.go +++ b/workflow/admin_recovery_test.go @@ -380,6 +380,12 @@ func TestAdminSkipAndOverrideRejectedForParkedGatewayNode(t *testing.T) { // block a sibling branch running in parallel: the sibling completes while the first branch // is still NodeStatusAwaitingAdmin. Aborting the parked branch afterward still fails the // overall workflow, since parallel join semantics are unchanged. +// TestAdminParkingIsolatesParallelBranches exercises PARALLEL_SPLIT's isolation: each +// matching branch runs as its own child workflow (see parallel_gateway.go), so a node parked +// for admin inside one branch (task_a, here) shows up in that child's own status and signal +// channel, not the parent's (similar to BATCH_SPLIT partitions). +// The child's deterministic ID is FormatBatchChildWorkflowID(parentID, splitNodeID, edgeID); +// parallelWorkflowJSON's split->task_a edge is "e2". func TestAdminParkingIsolatesParallelBranches(t *testing.T) { testSuite := &testsuite.WorkflowTestSuite{} env := testSuite.NewTestWorkflowEnvironment() @@ -396,21 +402,22 @@ func TestAdminParkingIsolatesParallelBranches(t *testing.T) { env.OnActivity("ExecuteTaskActivity", mock.Anything, "TASK_B", mock.Anything). Return(map[string]any{}, nil).Once() + branchWorkflowID := FormatBatchChildWorkflowID("default-test-workflow-id", "split", "e2") + env.RegisterDelayedCallback(func() { - val, err := env.QueryWorkflow("GetStatus") + val, err := env.QueryWorkflowByID(branchWorkflowID, "GetStatus") require.NoError(t, err) var instance WorkflowInstance require.NoError(t, val.Get(&instance)) require.Equal(t, NodeStatusAwaitingAdmin, instance.NodeInfo["task_a"].Status) - require.Equal(t, NodeStatusCompleted, instance.NodeInfo["task_b"].Status) }, time.Second) env.RegisterDelayedCallback(func() { - env.SignalWorkflow(AdminResolutionSignalName, AdminResolutionSignal{ + require.NoError(t, env.SignalWorkflowByID(branchWorkflowID, AdminResolutionSignalName, AdminResolutionSignal{ NodeID: "task_a", Action: AdminActionAbort, - }) + })) }, 2*time.Second) env.ExecuteWorkflow(GraphInterpreterWorkflow, def, map[string]any{}) diff --git a/workflow/batch_gateway_test.go b/workflow/batch_gateway_test.go index adc1d7b..bbd9a34 100644 --- a/workflow/batch_gateway_test.go +++ b/workflow/batch_gateway_test.go @@ -14,6 +14,7 @@ import ( "github.com/stretchr/testify/suite" "go.temporal.io/sdk/activity" "go.temporal.io/sdk/client" + "go.temporal.io/sdk/temporal" "go.temporal.io/sdk/testsuite" "go.temporal.io/sdk/workflow" ) @@ -1332,3 +1333,79 @@ func (s *BatchGatewayTestSuite) TestBatchSplit_ChildReturnsDuplicateItemIDAcross s.Error(err) s.Contains(err.Error(), "duplicate item ID \"item-a\" returned across child partitions") } + +// --- Test 16: Child workflow task aborted by admin propagates to parent without re-parking --- + +func (s *BatchGatewayTestSuite) TestBatchSplit_ChildTaskAdminAbort_PropagatesWithoutReparkingParent() { + env := s.NewTestWorkflowEnvironment() + + acts := &Activities{} + env.RegisterActivityWithOptions(acts.ExecuteTaskActivity, activity.RegisterOptions{Name: "ExecuteTaskActivity"}) + env.RegisterActivityWithOptions(acts.WorkflowCompletedActivity, activity.RegisterOptions{Name: "WorkflowCompletedActivity"}) + + def := WorkflowDefinition{ + ID: "batch_child_abort_test", + Name: "Batch Child Abort Test", + Nodes: []Node{ + {ID: "start", Type: NodeTypeStart}, + {ID: "gw_split", Type: NodeTypeGateway, GatewayType: GatewayTypeBatchSplit, + BatchGateway: &BatchGatewayConfig{}}, // defaults: _items, id + {ID: "process", Type: NodeTypeTask, TaskTemplateID: "PROCESS"}, + {ID: "gw_join", Type: NodeTypeGateway, GatewayType: GatewayTypeBatchJoin, + BatchJoin: &BatchJoinConfig{GatewayNodeID: "gw_split"}}, + {ID: "post_task", Type: NodeTypeTask, TaskTemplateID: "POST_TASK"}, + {ID: "end", Type: NodeTypeEnd}, + }, + Edges: []Edge{ + {ID: "e1", SourceID: "start", TargetID: "gw_split"}, + {ID: "e2", SourceID: "gw_split", TargetID: "process", Condition: `item.needsWork == true`}, + {ID: "e3", SourceID: "process", TargetID: "gw_join"}, + {ID: "e4", SourceID: "gw_join", TargetID: "post_task"}, + {ID: "e5", SourceID: "post_task", TargetID: "end"}, + }, + } + + env.OnActivity("ExecuteTaskActivity", mock.Anything, "PROCESS", mock.Anything). + Return(nil, temporal.NewNonRetryableApplicationError("inspection boom", "TaskFailure", nil)).Once() + + parentWorkflowID := "batch-child-abort-1" + env.RegisterWorkflowWithOptions(GraphInterpreterWorkflow, workflow.RegisterOptions{Name: "GraphInterpreterWorkflow"}) + env.SetStartWorkflowOptions(client.StartWorkflowOptions{ID: parentWorkflowID}) + + childWorkflowID := FormatBatchChildWorkflowID(parentWorkflowID, "gw_split", "e2") + + // 1. Verify that the child's node is parked awaiting admin intervention + env.RegisterDelayedCallback(func() { + val, err := env.QueryWorkflowByID(childWorkflowID, "GetStatus") + s.Require().NoError(err) + var instance WorkflowInstance + s.Require().NoError(val.Get(&instance)) + s.Require().NotNil(instance.NodeInfo["process"], "node 'process' must exist in child NodeInfo") + s.Equal(NodeStatusAwaitingAdmin, instance.NodeInfo["process"].Status) + }, time.Second) + + // 2. Resolve the child node with AdminActionAbort + env.RegisterDelayedCallback(func() { + s.NoError(env.SignalWorkflowByID(childWorkflowID, AdminResolutionSignalName, AdminResolutionSignal{ + NodeID: "process", + Action: AdminActionAbort, + })) + }, 2*time.Second) + + initialVars := map[string]any{ + "_items": []any{ + map[string]any{"id": "item1", "needsWork": true}, + }, + } + + env.ExecuteWorkflow(GraphInterpreterWorkflow, def, initialVars) + + // Workflow completes with failure immediately without re-parking the parent's gw_split + s.True(env.IsWorkflowCompleted()) + err := env.GetWorkflowError() + s.Error(err) + s.Contains(err.Error(), "inspection boom") + + // Verify post_task was never invoked + env.AssertNotCalled(s.T(), "ExecuteTaskActivity", mock.Anything, "POST_TASK", mock.Anything) +} diff --git a/workflow/batch_validate.go b/workflow/batch_validate.go index 52dd67c..8f87563 100644 --- a/workflow/batch_validate.go +++ b/workflow/batch_validate.go @@ -68,7 +68,7 @@ func ValidateBatchGateways(def WorkflowDefinition) error { // 4. Sub-graph topological containment: enforce that every path from BATCH_SPLIT // reaches its paired BATCH_JOIN, and no edges escape or illegally enter the sub-graph region. for splitID, joinID := range pairedJoins { - if err := validateBatchRegion(splitID, joinID, pairedJoins, nodesByID, forwardEdges, reverseEdges); err != nil { + if err := validateGatewayRegion(splitID, joinID, GatewayTypeBatchSplit, "BATCH_SPLIT", "BATCH_JOIN", "batch region", pairedJoins, nodesByID, forwardEdges, reverseEdges); err != nil { return err } } @@ -76,20 +76,28 @@ func ValidateBatchGateways(def WorkflowDefinition) error { return nil } -// validateBatchRegion checks that the sub-graph between splitID and joinID is strictly -// closed: all paths must terminate at joinID, with no dead ends, escaped edges, or illegal entries. -func validateBatchRegion( +// validateGatewayRegion checks that the sub-graph between splitID and joinID is strictly +// closed: all paths must terminate at joinID, with no dead ends, escaped edges, or illegal +// entries. Shared by ValidateBatchGateways and ValidateParallelGateways — nestedType and +// pairedJoins scope the "nested split's join must stay inside this region" check to gateways +// of the same kind as splitID itself (a nested BATCH_SPLIT inside a BATCH_SPLIT region, or a +// nested PARALLEL_SPLIT inside a PARALLEL_SPLIT region); it does not check cross-kind nesting. +// splitLabel/joinLabel/regionLabel format the error messages (e.g. "BATCH_SPLIT"/"BATCH_JOIN"/ +// "batch region" vs "PARALLEL_SPLIT"/"PARALLEL_JOIN"/"parallel region"). +func validateGatewayRegion( splitID, joinID string, + nestedType GatewayType, + splitLabel, joinLabel, regionLabel string, pairedJoins map[string]string, nodesByID map[string]*Node, forwardEdges map[string][]Edge, reverseEdges map[string][]string, ) error { if len(forwardEdges[splitID]) == 0 { - return fmt.Errorf("BATCH_SPLIT node %q has no outgoing edges", splitID) + return fmt.Errorf("%s node %q has no outgoing edges", splitLabel, splitID) } - // 1. Forward reachability: find all nodes in the batch region, stopping at joinID. + // 1. Forward reachability: find all nodes in the region, stopping at joinID. regionNodes := make(map[string]bool) queue := []string{splitID} regionNodes[splitID] = true @@ -132,40 +140,40 @@ func validateBatchRegion( for nodeID := range regionNodes { node, exists := nodesByID[nodeID] if !exists { - return fmt.Errorf("BATCH_SPLIT node %q region references non-existent node %q", splitID, nodeID) + return fmt.Errorf("%s node %q region references non-existent node %q", splitLabel, splitID, nodeID) } if node.Type == NodeTypeEnd { - return fmt.Errorf("BATCH_SPLIT node %q has a path reaching END node %q without passing through paired BATCH_JOIN %q", splitID, nodeID, joinID) + return fmt.Errorf("%s node %q has a path reaching END node %q without passing through paired %s %q", splitLabel, splitID, nodeID, joinLabel, joinID) } if len(forwardEdges[nodeID]) == 0 { - return fmt.Errorf("BATCH_SPLIT node %q region contains dead-end node %q with no outgoing edges", splitID, nodeID) + return fmt.Errorf("%s node %q region contains dead-end node %q with no outgoing edges", splitLabel, splitID, nodeID) } if !canReachJoin[nodeID] { - return fmt.Errorf("BATCH_SPLIT node %q region contains node %q which cannot reach paired BATCH_JOIN %q", splitID, nodeID, joinID) + return fmt.Errorf("%s node %q region contains node %q which cannot reach paired %s %q", splitLabel, splitID, nodeID, joinLabel, joinID) } - // Nested BATCH_SPLIT must have its paired BATCH_JOIN contained within this batch region. - if node.Type == NodeTypeGateway && node.GatewayType == GatewayTypeBatchSplit && nodeID != splitID { + // A nested split of the SAME kind must have its paired join contained within this region. + if node.Type == NodeTypeGateway && node.GatewayType == nestedType && nodeID != splitID { nestedJoinID := pairedJoins[nodeID] if !regionNodes[nestedJoinID] { - return fmt.Errorf("BATCH_SPLIT node %q contains nested BATCH_SPLIT %q whose paired BATCH_JOIN %q is outside the batch region", splitID, nodeID, nestedJoinID) + return fmt.Errorf("%s node %q contains nested %s %q whose paired %s %q is outside the %s", splitLabel, splitID, splitLabel, nodeID, joinLabel, nestedJoinID, regionLabel) } } - // Encapsulation: no edges from outside the batch region may enter intermediate region nodes. + // Encapsulation: no edges from outside the region may enter intermediate region nodes. if nodeID != splitID { for _, prev := range reverseEdges[nodeID] { if !regionNodes[prev] { - return fmt.Errorf("BATCH_SPLIT node %q region contains node %q with incoming edge from outside the batch region (from %q)", splitID, nodeID, prev) + return fmt.Errorf("%s node %q region contains node %q with incoming edge from outside the %s (from %q)", splitLabel, splitID, nodeID, regionLabel, prev) } } } } - // 4. Encapsulation: all incoming edges to the paired BATCH_JOIN must originate within this batch region. + // 4. Encapsulation: all incoming edges to the paired join must originate within this region. for _, prev := range reverseEdges[joinID] { if !regionNodes[prev] { - return fmt.Errorf("BATCH_JOIN node %q has incoming edge from node %q outside its paired BATCH_SPLIT %q region", joinID, prev, splitID) + return fmt.Errorf("%s node %q has incoming edge from node %q outside its paired %s %q region", joinLabel, joinID, prev, splitLabel, splitID) } } diff --git a/workflow/dsl.go b/workflow/dsl.go index 4beca30..8778750 100644 --- a/workflow/dsl.go +++ b/workflow/dsl.go @@ -161,6 +161,7 @@ type Node struct { Signaling *SignalingConfig `json:"signaling,omitempty"` BatchGateway *BatchGatewayConfig `json:"batch_gateway,omitempty"` BatchJoin *BatchJoinConfig `json:"batch_join,omitempty"` + ParallelJoin *ParallelJoinConfig `json:"parallel_join,omitempty"` } // Edge represents a directed connection between two nodes. @@ -219,6 +220,59 @@ type BatchJoinConfig struct { IDField string `json:"id_field,omitempty"` } +// ParallelJoinConfig configures how a PARALLEL_JOIN gateway isolates its branches and +// merges their final workflow variable states back into the parent scope. +// +// WARNING: CONFLICT RESOLUTION AND LAST-WRITE-WINS (LWW) +// When concurrent parallel branches write conflicting data, the engine cannot infer intent +// and applies a deterministic "last-write-wins" policy. Precedence is determined strictly by +// branch edge order (sorted ascending by source edge ID; later branches in that order overwrite +// earlier ones), NOT wall-clock completion time. +// +// Specifically, LAST-WRITE-WINS applies in the following three places: +// 1. Conflicting item fields in MergeByID: If two branches modify the SAME field on the SAME +// item (e.g. branch A sets item["status"]="PASS" and branch B sets item["status"]="FAIL"), +// the later branch's field value overwrites the earlier one. +// 2. Conflicting map keys: If two branches set the SAME leaf key in a map[string]any, the +// later branch's value overwrites the earlier one. +// 3. Scalars and unlisted arrays: Any scalar variable (string, number, boolean) or slice not +// listed in MergeByID that is written by multiple branches will be completely overwritten +// by the later branch. +// +// Recommendation for workflow authors: Ensure concurrent parallel branches write to distinct +// variable paths or distinct field names on shared items if overwriting cannot be tolerated. +// +// Execution and Reconciliation Model: +// Each matching outgoing edge of the paired PARALLEL_SPLIT runs as its own child workflow +// with a deep-copied, isolated set of WorkflowVariables — no branch can see another +// branch's writes while any of them are still running. When all branches complete, +// PARALLEL_JOIN reconciles their (possibly divergent) final states back into one: +// +// - map[string]any values are deep-merged key by key (recursive map merge): +// each branch's mutated sub-fields combine, so two branches changing different +// fields of the same object both survive. +// - Variables listed in MergeByID (each a []map[string]any) are merged element-by-element +// by that id_field across ALL branches — not "one branch's array replaces another's": +// for a given item ID, the fields each branch's copy of that item carries are unioned +// into one item, so lab writing sample_test_result and visual writing visual_result to +// the same underlying item both survive regardless of which branch finishes last. A +// field two branches BOTH carry a (possibly different) value for is a genuine conflict +// this cannot resolve: branches are applied in a fixed, deterministic order (sorted by +// source edge ID) and the last one wins for that field — give each branch its own field +// name if that ambiguity isn't acceptable. +// - Any other variable (scalar, or array not listed in MergeByID) falls back to the same +// fixed deterministic branch order, last one wins. +type ParallelJoinConfig struct { + // GatewayNodeID is the node ID of the paired PARALLEL_SPLIT gateway. + GatewayNodeID string `json:"gateway_node_id"` + + // MergeByID maps a top-level workflow variable name (holding a []map[string]any shared by + // multiple branches) to the field name within each item used as its unique ID. + // Nested dot-paths are not currently supported; see ValidateParallelGateways. + // TODO: Support nested dot-paths (e.g. "order.items") in MergeByID and mergeVariablesInto. + MergeByID map[string]string `json:"merge_by_id,omitempty"` +} + // VarScopePath is the workflow variable key holding the hierarchical scope path string // for batch gateway nesting (e.g. "root/gw_type/lab/gw_result/fail"). const VarScopePath = "_scope_path" diff --git a/workflow/dynamic_split_test.go b/workflow/dynamic_split_test.go index 4afa6ce..b50818c 100644 --- a/workflow/dynamic_split_test.go +++ b/workflow/dynamic_split_test.go @@ -336,10 +336,9 @@ func (s *NSWEngineTestSuite) TestDynamicFanOutWithCollectAllFailures() { }, } - // The child "fail_task" activity error now parks each child's c_task node for admin - // intervention instead of failing the child workflow outright, and the parent's - // m_fanout node parks too once the aggregate failure reaches it. Abort each in turn to - // reproduce today's "multiple branches failed" end-state. + // Activity failures park each child's c_task node for admin intervention, and the + // parent's m_fanout node parks once the aggregate failure reaches it. Abort each node + // to assert the "multiple branches failed" end-state. const parentWorkflowID = "master-collect-all-test" env.SetStartWorkflowOptions(client.StartWorkflowOptions{ID: parentWorkflowID}) @@ -556,11 +555,9 @@ func (s *NSWEngineTestSuite) TestDynamicFanOutWithCrossBranchBroadcast() { // TestConcurrentSplitTasksDoNotCrossTalkBroadcast guards against the broadcast channel being // shared between two SPLIT_TASK nodes that run concurrently in the same workflow execution -// (e.g. both reachable from a PARALLEL_SPLIT gateway). Before the broadcast channel was scoped -// per split node, both groups' monitorChildWorkflows listened on the same fixed signal name, so -// a broadcast emitted by one group's child could be received and relayed by the other group's -// selector instead — delivering the wrong group's payload, or losing it so the intended waiter -// hangs forever. +// (e.g. both reachable from a PARALLEL_SPLIT gateway): the broadcast channel is scoped per +// split node so that broadcasts emitted by one split task group are not cross-talked or +// misrouted to a concurrent group's monitorChildWorkflows. func (s *NSWEngineTestSuite) TestConcurrentSplitTasksDoNotCrossTalkBroadcast() { env := s.NewTestWorkflowEnvironment() @@ -647,7 +644,7 @@ func (s *NSWEngineTestSuite) TestConcurrentSplitTasksDoNotCrossTalkBroadcast() { IterationKey: "custom_iter", }, }, - {ID: "m_pjoin", Type: NodeTypeGateway, GatewayType: GatewayTypeParallelJoin}, + {ID: "m_pjoin", Type: NodeTypeGateway, GatewayType: GatewayTypeParallelJoin, ParallelJoin: &ParallelJoinConfig{GatewayNodeID: "m_psplit"}}, {ID: "m_end", Type: NodeTypeEnd}, }, Edges: []Edge{ @@ -769,8 +766,7 @@ func (s *NSWEngineTestSuite) TestChildBranchEndNodeDoesNotFireCompletionHook() { } env.OnActivity("FetchWorkflowDefinitionActivity", mock.Anything, "child_wf").Return(childDef, nil) - // The test env runs activity mocks on their own goroutines, so if this fix regresses and - // multiple branches fire the hook concurrently, the append would race under -race. Guard it. + // Guard against concurrent appends under -race if multiple branches fire the hook concurrently. var mu sync.Mutex var completedIDs []string env.OnActivity("WorkflowCompletedActivity", mock.Anything, mock.Anything, mock.Anything).Return( @@ -795,12 +791,10 @@ func (s *NSWEngineTestSuite) TestChildBranchEndNodeDoesNotFireCompletionHook() { "completion hook must fire once for the top-level workflow only, not per child branch") } -// TestChildBranchCompletionHandlerErrorDoesNotHang is the direct regression for the reported bug: -// when the host's completion handler errors on IDs it doesn't recognize (child branches carry a -// synthetic ID like master-1--m_fanout--b1-0 that isn't in the host's registry), the overall -// workflow must still complete. Before the fix, each child branch invoked the hook at its END -// node; the error there parked the branch for admin, so it never completed and the parent's -// monitorChildWorkflows blocked forever (ScheduleToClose deadline exceeded). +// TestChildBranchCompletionHandlerErrorDoesNotHang verifies that when the host's completion handler +// errors on synthetic child branch IDs not found in the registry, the overall workflow completes +// successfully: child branches must not invoke the completion hook at their END node, which would +// otherwise park the branch and cause the parent's monitorChildWorkflows to block. func (s *NSWEngineTestSuite) TestChildBranchCompletionHandlerErrorDoesNotHang() { env := s.NewTestWorkflowEnvironment() diff --git a/workflow/engine_test.go b/workflow/engine_test.go index 69489ec..39d9049 100644 --- a/workflow/engine_test.go +++ b/workflow/engine_test.go @@ -78,7 +78,7 @@ const parallelWorkflowJSON = ` { "id": "split", "type": "GATEWAY", "gateway_type": "PARALLEL_SPLIT" }, { "id": "task_a", "type": "TASK", "task_template_id": "TASK_A" }, { "id": "task_b", "type": "TASK", "task_template_id": "TASK_B" }, - { "id": "join", "type": "GATEWAY", "gateway_type": "PARALLEL_JOIN" }, + { "id": "join", "type": "GATEWAY", "gateway_type": "PARALLEL_JOIN", "parallel_join": { "gateway_node_id": "split" } }, { "id": "task_c", "type": "TASK", "task_template_id": "TASK_C" }, { "id": "end", "type": "END" } ] @@ -375,8 +375,7 @@ func TestTaskNodeFailsWhenInputKeyMissing(t *testing.T) { env.RegisterActivityWithOptions(acts.ExecuteTaskActivity, activity.RegisterOptions{Name: "ExecuteTaskActivity"}) env.RegisterActivityWithOptions(acts.WorkflowCompletedActivity, activity.RegisterOptions{Name: "WorkflowCompletedActivity"}) - // The input mapping error now parks the node for admin intervention instead of - // failing the workflow outright. Abort it to reproduce today's end-state. + // The input mapping error parks the node for admin intervention. Abort it so the workflow fails. env.RegisterDelayedCallback(func() { env.SignalWorkflow(AdminResolutionSignalName, AdminResolutionSignal{ NodeID: "task", @@ -553,8 +552,7 @@ func TestTaskNodeFailsWhenRequiredOutputMissing(t *testing.T) { env.OnActivity("ExecuteTaskActivity", mock.Anything, "TASK_MISSING_REQUIRED_OUTPUT", mock.Anything). Return(map[string]any{}, nil).Once() - // The output mapping error now parks the node for admin intervention instead of - // failing the workflow outright. Abort it to reproduce today's end-state. + // The output mapping error parks the node for admin intervention. Abort it so the workflow fails. env.RegisterDelayedCallback(func() { env.SignalWorkflow(AdminResolutionSignalName, AdminResolutionSignal{ NodeID: "task", diff --git a/workflow/parallel_gateway.go b/workflow/parallel_gateway.go new file mode 100644 index 0000000..36da5e9 --- /dev/null +++ b/workflow/parallel_gateway.go @@ -0,0 +1,249 @@ +// SPDX-License-Identifier: Apache-2.0 +// Copyright (c) 2026 Lanka Software Foundation + +package engine + +import ( + "fmt" + "sort" + + "go.temporal.io/sdk/workflow" + + "github.com/OpenNSW/core/shared/deepcopy" + "github.com/OpenNSW/core/shared/maputil" +) + +// parallelBranch holds the future of a spawned child workflow for one PARALLEL_SPLIT branch. +type parallelBranch struct { + EdgeID string + Future workflow.ChildWorkflowFuture +} + +// handleParallelSplitGateway runs each matching outgoing edge as its own isolated child +// workflow — a deep copy of the current WorkflowVariables, invisible to sibling branches +// until they all complete — and reconciles their final states back into the parent via the +// paired PARALLEL_JOIN's merge configuration. This mirrors handleBatchSplitGateway's +// spawn-children-then-merge shape, but branches see the FULL variable set (not a partitioned +// subset), so — unlike batch partitions, which are disjoint by construction — more than one +// branch can legitimately touch the same variable, which is exactly what the merge step +// exists to reconcile. +func (g *graphInterpreter) handleParallelSplitGateway(ctx workflow.Context, nodeInfo *NodeInfo, node *Node, outEdges []Edge) error { + joinNodeID := findPairedParallelJoin(g.def, node.ID) + if joinNodeID == "" { + return fmt.Errorf("PARALLEL_SPLIT node %s: no paired PARALLEL_JOIN found", node.ID) + } + joinNode := g.nodes[joinNodeID] + var mergeByID map[string]string + if joinNode != nil && joinNode.ParallelJoin != nil { + mergeByID = joinNode.ParallelJoin.MergeByID + } + + // 1. Evaluate conditions up front against the pre-split state. + var matchedEdgeIDs []string + edgeByID := make(map[string]Edge, len(outEdges)) + for _, e := range outEdges { + edgeByID[e.ID] = e + match, err := EvaluateCondition(e.Condition, g.instance.WorkflowVariables) + if err != nil { + return err + } + if match { + matchedEdgeIDs = append(matchedEdgeIDs, e.ID) + } + } + if len(matchedEdgeIDs) == 0 { + nodeInfo.Status = NodeStatusCompleted + nodeInfo.UpdatedAt = workflow.Now(ctx) + return g.skipToJoinOutEdge(ctx, joinNodeID) + } + // Deterministic branch order: both spawn order (replay-safety) and merge order (so a + // genuine same-field conflict between branches resolves the same way every run). + sort.Strings(matchedEdgeIDs) + + // 2. Spawn one isolated child workflow per branch, each a deep copy of the current state. + baseVars := deepcopy.Map(g.instance.WorkflowVariables) + parentInfo := workflow.GetInfo(ctx) + branches := make([]parallelBranch, 0, len(matchedEdgeIDs)) + for _, edgeID := range matchedEdgeIDs { + e := edgeByID[edgeID] + subDef := extractSubGraph(g.def, e.TargetID, joinNodeID) + childVars := deepcopy.Map(baseVars) + + childWorkflowID := FormatBatchChildWorkflowID(parentInfo.WorkflowExecution.ID, node.ID, edgeID) + childCtx := workflow.WithChildOptions(ctx, workflow.ChildWorkflowOptions{ + WorkflowID: childWorkflowID, + }) + future := workflow.ExecuteChildWorkflow(childCtx, "GraphInterpreterWorkflow", subDef, childVars) + branches = append(branches, parallelBranch{EdgeID: edgeID, Future: future}) + } + + // 3. Wait for every branch, then reconcile their final states into one. + branchVars := make([]map[string]any, 0, len(branches)) + for _, b := range branches { + var childOutput *WorkflowInstance + if err := b.Future.Get(ctx, &childOutput); err != nil { + return fmt.Errorf("PARALLEL_SPLIT node %s: branch %q failed: %w", node.ID, b.EdgeID, err) + } + if childOutput == nil { + return fmt.Errorf("PARALLEL_SPLIT node %s: branch %q returned nil output", node.ID, b.EdgeID) + } + branchVars = append(branchVars, childOutput.WorkflowVariables) + } + + mergedVars, err := mergeParallelBranches(baseVars, branchVars, mergeByID) + if err != nil { + return fmt.Errorf("PARALLEL_SPLIT node %s: merge failed: %w", node.ID, err) + } + g.instance.WorkflowVariables = mergedVars + + g.instance.AuditTrail = append(g.instance.AuditTrail, + fmt.Sprintf("PARALLEL_SPLIT %s ran %d branch(es) and merged results", node.ID, len(branches))) + + nodeInfo.Status = NodeStatusCompleted + nodeInfo.UpdatedAt = workflow.Now(ctx) + return g.skipToJoinOutEdge(ctx, joinNodeID) +} + +// handleParallelJoinGateway is a structural passthrough: the paired PARALLEL_SPLIT gateway +// already performed the wait + merge, so the join simply marks itself complete and moves on — +// same shape as handleBatchJoinGateway. +func (g *graphInterpreter) handleParallelJoinGateway(ctx workflow.Context, nodeInfo *NodeInfo, node *Node, outEdges []Edge) error { + if node.ParallelJoin == nil || node.ParallelJoin.GatewayNodeID == "" { + return fmt.Errorf("PARALLEL_JOIN node %s: parallel_join.gateway_node_id is required", node.ID) + } + + nodeInfo.Status = NodeStatusCompleted + nodeInfo.UpdatedAt = workflow.Now(ctx) + + g.instance.AuditTrail = append(g.instance.AuditTrail, + fmt.Sprintf("PARALLEL_JOIN %s completed (paired with %s)", node.ID, node.ParallelJoin.GatewayNodeID)) + + if len(outEdges) > 0 { + return g.transitionTo(ctx, outEdges[0]) + } + return nil +} + +// findPairedParallelJoin searches the workflow definition for a PARALLEL_JOIN gateway node +// whose GatewayNodeID matches the given PARALLEL_SPLIT node ID. +func findPairedParallelJoin(def WorkflowDefinition, splitNodeID string) string { + for _, node := range def.Nodes { + if node.Type == NodeTypeGateway && node.GatewayType == GatewayTypeParallelJoin && + node.ParallelJoin != nil && node.ParallelJoin.GatewayNodeID == splitNodeID { + return node.ID + } + } + return "" +} + +// mergeParallelBranches reconciles N branches' final WorkflowVariables (each a full, isolated +// copy that started as `base`) back into one map. Branches are applied in the order given by +// the caller — expected to be a fixed, deterministic order — so that a genuine same-field +// conflict resolves the same way on every run. +// +// mergeByID-listed variables (top-level variable names only — see ValidateParallelGateways) are +// merged item-by-item, by ID, across every branch: for each item ID, the fields each branch's +// copy of that item carries are unioned into one item. Everything else is merged generically: +// map[string]any values merge key by key (recursively); anything else is last-branch-wins. +func mergeParallelBranches(base map[string]any, branchVars []map[string]any, mergeByID map[string]string) (map[string]any, error) { + result := deepcopy.Map(base) + + mergeByIDKeys := make(map[string]bool, len(mergeByID)) + for varPath := range mergeByID { + mergeByIDKeys[varPath] = true + } + + for varPath, idField := range mergeByID { + merged, err := mergeItemsByID(result, branchVars, varPath, idField) + if err != nil { + return nil, err + } + maputil.SetNestedKey(result, varPath, merged) + } + + for _, bv := range branchVars { + mergeVariablesInto(result, bv, mergeByIDKeys) + } + + return result, nil +} + +// mergeItemsByID merges a []map[string]any variable (found at varPath) across the base state +// and every branch's copy of it, keyed by idField. For a given item ID, fields present in a +// later branch's copy overwrite the same field from an earlier one (or from base); fields a +// branch's copy simply doesn't have are left untouched. Items are returned in first-seen order +// (base's order, then any items a branch introduced that base didn't have). +func mergeItemsByID(base map[string]any, branchVars []map[string]any, varPath, idField string) ([]any, error) { + merged := make(map[string]map[string]any) + var order []string + + absorb := func(raw any, origin string) error { + if raw == nil { + return nil + } + items, err := toItemSlice(raw) + if err != nil { + return fmt.Errorf("variable %q from %s is invalid items: %w", varPath, origin, err) + } + for i, item := range items { + idVal := getItemID(item, idField) + idStr := fmt.Sprintf("%v", idVal) + if idVal == nil || idVal == "" || idStr == "" || idStr == "" { + return fmt.Errorf("variable %q from %s has item at index %d missing required ID field %q", + varPath, origin, i, idField) + } + existing, ok := merged[idStr] + if !ok { + merged[idStr] = deepcopy.Map(item) + order = append(order, idStr) + continue + } + for k, v := range item { + existing[k] = deepcopy.Value(v) + } + } + return nil + } + + if baseRaw, ok := maputil.GetNestedKey(base, varPath); ok { + if err := absorb(baseRaw, "base state"); err != nil { + return nil, err + } + } + for i, bv := range branchVars { + if raw, ok := maputil.GetNestedKey(bv, varPath); ok { + if err := absorb(raw, fmt.Sprintf("branch %d", i)); err != nil { + return nil, err + } + } + } + + out := make([]any, 0, len(order)) + for _, id := range order { + out = append(out, merged[id]) + } + return out, nil +} + +// mergeVariablesInto merges src into dst: map[string]any values merge key by key +// (recursively); anything else (scalars, arrays not handled by mergeItemsByID) is a plain +// overwrite. Top-level keys in skipTopKeys (the mergeByID-configured variables, already +// merged separately) are skipped. +func mergeVariablesInto(dst, src map[string]any, skipTopKeys map[string]bool) { + for k, v := range src { + if skipTopKeys[k] { + continue + } + if srcMap, ok := v.(map[string]any); ok { + if existing, exists := dst[k]; exists { + if dstMap, ok := existing.(map[string]any); ok { + mergeVariablesInto(dstMap, srcMap, nil) + continue + } + } + dst[k] = deepcopy.Map(srcMap) + continue + } + dst[k] = deepcopy.Value(v) + } +} diff --git a/workflow/parallel_gateway_test.go b/workflow/parallel_gateway_test.go new file mode 100644 index 0000000..b05dac6 --- /dev/null +++ b/workflow/parallel_gateway_test.go @@ -0,0 +1,358 @@ +// SPDX-License-Identifier: Apache-2.0 +// Copyright (c) 2026 Lanka Software Foundation + +package engine + +import ( + "context" + "fmt" + "strings" + "testing" + "time" + + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/suite" + "go.temporal.io/sdk/activity" + "go.temporal.io/sdk/client" + "go.temporal.io/sdk/testsuite" + "go.temporal.io/sdk/workflow" +) + +type ParallelGatewayTestSuite struct { + suite.Suite + testsuite.WorkflowTestSuite +} + +func TestParallelGatewayTestSuite(t *testing.T) { + suite.Run(t, new(ParallelGatewayTestSuite)) +} + +// buildParallelMergeWorkflow: start -> load_items -> PARALLEL_SPLIT -> (lab_task | visual_task) +// -> PARALLEL_JOIN -> end. Both branches read and independently annotate the same shared +// `commodities` array. +func buildParallelMergeWorkflow(mergeByID map[string]string) WorkflowDefinition { + return WorkflowDefinition{ + ID: "parallel_merge_test", + Name: "Parallel Merge Test", + Nodes: []Node{ + {ID: "start", Type: NodeTypeStart}, + {ID: "load_items", Type: NodeTypeTask, TaskTemplateID: "LOAD_ITEMS", + OutputMapping: map[string]string{"items": "commodities"}}, + {ID: "psplit", Type: NodeTypeGateway, GatewayType: GatewayTypeParallelSplit}, + {ID: "lab_task", Type: NodeTypeTask, TaskTemplateID: "LAB_TASK", + InputMapping: map[string]string{"commodities": "commodities"}, + OutputMapping: map[string]string{"commodities": "commodities"}}, + {ID: "visual_task", Type: NodeTypeTask, TaskTemplateID: "VISUAL_TASK", + InputMapping: map[string]string{"commodities": "commodities"}, + OutputMapping: map[string]string{"commodities": "commodities"}}, + {ID: "pjoin", Type: NodeTypeGateway, GatewayType: GatewayTypeParallelJoin, + ParallelJoin: &ParallelJoinConfig{GatewayNodeID: "psplit", MergeByID: mergeByID}}, + {ID: "end", Type: NodeTypeEnd}, + }, + Edges: []Edge{ + {ID: "e1", SourceID: "start", TargetID: "load_items"}, + {ID: "e2", SourceID: "load_items", TargetID: "psplit"}, + {ID: "e3", SourceID: "psplit", TargetID: "lab_task"}, + {ID: "e4", SourceID: "psplit", TargetID: "visual_task"}, + {ID: "e5", SourceID: "lab_task", TargetID: "pjoin"}, + {ID: "e6", SourceID: "visual_task", TargetID: "pjoin"}, + {ID: "e7", SourceID: "pjoin", TargetID: "end"}, + }, + } +} + +func mockWorkflowCompletedIgnoringChildren(env *testsuite.TestWorkflowEnvironment) { + env.OnActivity("WorkflowCompletedActivity", mock.Anything, mock.Anything, mock.Anything).Return( + func(_ context.Context, workflowID string, _ map[string]any) error { + if strings.Contains(workflowID, "--") { + return fmt.Errorf("workflow %s not found in host registry", workflowID) + } + return nil + }) +} + +// TestParallelSplit_ItemFieldMerge_BothBranchesFieldsSurvive verifies that when two branches +// independently annotate the same items in a shared array (lab_task adds lab_result and +// visual_task adds visual_result to every item), both fields are preserved after merge, +// regardless of which branch's child workflow finishes first. +func (s *ParallelGatewayTestSuite) TestParallelSplit_ItemFieldMerge_BothBranchesFieldsSurvive() { + env := s.NewTestWorkflowEnvironment() + + acts := &Activities{} + env.RegisterActivityWithOptions(acts.ExecuteTaskActivity, activity.RegisterOptions{Name: "ExecuteTaskActivity"}) + env.RegisterActivityWithOptions(acts.WorkflowCompletedActivity, activity.RegisterOptions{Name: "WorkflowCompletedActivity"}) + mockWorkflowCompletedIgnoringChildren(env) + + def := buildParallelMergeWorkflow(map[string]string{"commodities": "id"}) + + env.OnActivity("ExecuteTaskActivity", mock.Anything, "LOAD_ITEMS", mock.Anything). + Return(map[string]any{ + "items": []any{ + map[string]any{"id": "item-1", "name": "Cut Flowers"}, + map[string]any{"id": "item-2", "name": "Timber"}, + }, + }, nil).Once() + + // lab_task: annotates every item with lab_result, echoing the identity fields it saw. + env.OnActivity("ExecuteTaskActivity", mock.Anything, "LAB_TASK", mock.Anything). + Return(func(_ context.Context, _ string, inputs map[string]any) (map[string]any, error) { + in, _ := inputs["commodities"].([]any) + out := make([]any, 0, len(in)) + for _, raw := range in { + item, _ := raw.(map[string]any) + updated := map[string]any{"id": item["id"], "name": item["name"], "lab_result": "pass"} + out = append(out, updated) + } + return map[string]any{"commodities": out}, nil + }).Once() + + // visual_task: annotates every item with visual_result, independently of lab_task. + env.OnActivity("ExecuteTaskActivity", mock.Anything, "VISUAL_TASK", mock.Anything). + Return(func(_ context.Context, _ string, inputs map[string]any) (map[string]any, error) { + in, _ := inputs["commodities"].([]any) + out := make([]any, 0, len(in)) + for _, raw := range in { + item, _ := raw.(map[string]any) + updated := map[string]any{"id": item["id"], "name": item["name"], "visual_result": "clean"} + out = append(out, updated) + } + return map[string]any{"commodities": out}, nil + }).Once() + + env.RegisterWorkflowWithOptions(GraphInterpreterWorkflow, workflow.RegisterOptions{Name: "GraphInterpreterWorkflow"}) + env.SetStartWorkflowOptions(client.StartWorkflowOptions{ID: "parallel-merge-test-1"}) + + env.ExecuteWorkflow(GraphInterpreterWorkflow, def, map[string]any{}) + + s.True(env.IsWorkflowCompleted()) + s.NoError(env.GetWorkflowError()) + + var result WorkflowInstance + s.NoError(env.GetWorkflowResult(&result)) + + items, ok := result.WorkflowVariables["commodities"].([]any) + s.Require().True(ok, "commodities should be a []any after merge") + s.Len(items, 2) + + byID := make(map[string]map[string]any, len(items)) + for _, raw := range items { + item, _ := raw.(map[string]any) + byID[item["id"].(string)] = item + } + + s.Equal("pass", byID["item-1"]["lab_result"], "item-1 must keep lab's contribution") + s.Equal("clean", byID["item-1"]["visual_result"], "item-1 must keep visual's contribution") + s.Equal("pass", byID["item-2"]["lab_result"], "item-2 must keep lab's contribution") + s.Equal("clean", byID["item-2"]["visual_result"], "item-2 must keep visual's contribution") +} + +// TestParallelSplit_GenericMapMerge_DisjointFieldsSurvive covers a shared variable that is a +// map (not an array), with each branch writing a different sub-field — verifying that map +// sub-fields merge properly across isolated child workflows. +func (s *ParallelGatewayTestSuite) TestParallelSplit_GenericMapMerge_DisjointFieldsSurvive() { + env := s.NewTestWorkflowEnvironment() + + acts := &Activities{} + env.RegisterActivityWithOptions(acts.ExecuteTaskActivity, activity.RegisterOptions{Name: "ExecuteTaskActivity"}) + env.RegisterActivityWithOptions(acts.WorkflowCompletedActivity, activity.RegisterOptions{Name: "WorkflowCompletedActivity"}) + mockWorkflowCompletedIgnoringChildren(env) + + def := WorkflowDefinition{ + ID: "parallel_map_merge_test", + Name: "Parallel Map Merge Test", + Nodes: []Node{ + {ID: "start", Type: NodeTypeStart}, + {ID: "psplit", Type: NodeTypeGateway, GatewayType: GatewayTypeParallelSplit}, + {ID: "track_a", Type: NodeTypeTask, TaskTemplateID: "TRACK_A", + OutputMapping: map[string]string{"value": "status.a_done"}}, + {ID: "track_b", Type: NodeTypeTask, TaskTemplateID: "TRACK_B", + OutputMapping: map[string]string{"value": "status.b_done"}}, + {ID: "pjoin", Type: NodeTypeGateway, GatewayType: GatewayTypeParallelJoin, + ParallelJoin: &ParallelJoinConfig{GatewayNodeID: "psplit"}}, + {ID: "end", Type: NodeTypeEnd}, + }, + Edges: []Edge{ + {ID: "e1", SourceID: "start", TargetID: "psplit"}, + {ID: "e2", SourceID: "psplit", TargetID: "track_a"}, + {ID: "e3", SourceID: "psplit", TargetID: "track_b"}, + {ID: "e4", SourceID: "track_a", TargetID: "pjoin"}, + {ID: "e5", SourceID: "track_b", TargetID: "pjoin"}, + {ID: "e6", SourceID: "pjoin", TargetID: "end"}, + }, + } + + env.OnActivity("ExecuteTaskActivity", mock.Anything, "TRACK_A", mock.Anything). + Return(map[string]any{"value": true}, nil).Once() + env.OnActivity("ExecuteTaskActivity", mock.Anything, "TRACK_B", mock.Anything). + Return(map[string]any{"value": true}, nil).Once() + + env.RegisterWorkflowWithOptions(GraphInterpreterWorkflow, workflow.RegisterOptions{Name: "GraphInterpreterWorkflow"}) + env.SetStartWorkflowOptions(client.StartWorkflowOptions{ID: "parallel-map-merge-test-1"}) + + env.ExecuteWorkflow(GraphInterpreterWorkflow, def, map[string]any{}) + + s.True(env.IsWorkflowCompleted()) + s.NoError(env.GetWorkflowError()) + + var result WorkflowInstance + s.NoError(env.GetWorkflowResult(&result)) + + status, ok := result.WorkflowVariables["status"].(map[string]any) + s.Require().True(ok, "status should be a map after merge") + s.Equal(true, status["a_done"], "track_a's field must survive") + s.Equal(true, status["b_done"], "track_b's field must ALSO survive") +} + +// TestParallelSplit_SameFieldConflict_LastBranchWinsDeterministically documents the one case +// the merge cannot resolve on its own: two branches writing DIFFERENT values to the exact same +// field of the exact same item. There is no data-driven "correct" answer here — this just +// pins down that the outcome is deterministic (branches merge in sorted edge-ID order, last one +// wins for a genuinely conflicting field) rather than flaky, so a workflow author who hits this +// knows to give each branch its own field name instead of relying on the resolution order. +func (s *ParallelGatewayTestSuite) TestParallelSplit_SameFieldConflict_LastBranchWinsDeterministically() { + env := s.NewTestWorkflowEnvironment() + + acts := &Activities{} + env.RegisterActivityWithOptions(acts.ExecuteTaskActivity, activity.RegisterOptions{Name: "ExecuteTaskActivity"}) + env.RegisterActivityWithOptions(acts.WorkflowCompletedActivity, activity.RegisterOptions{Name: "WorkflowCompletedActivity"}) + mockWorkflowCompletedIgnoringChildren(env) + + def := buildParallelMergeWorkflow(map[string]string{"commodities": "id"}) + + env.OnActivity("ExecuteTaskActivity", mock.Anything, "LOAD_ITEMS", mock.Anything). + Return(map[string]any{ + "items": []any{map[string]any{"id": "item-1"}}, + }, nil).Once() + + // Both branches write the SAME field ("failure_reason") with DIFFERENT values. + env.OnActivity("ExecuteTaskActivity", mock.Anything, "LAB_TASK", mock.Anything). + Return(map[string]any{ + "commodities": []any{map[string]any{"id": "item-1", "failure_reason": "lab reason"}}, + }, nil).Once() + env.OnActivity("ExecuteTaskActivity", mock.Anything, "VISUAL_TASK", mock.Anything). + Return(map[string]any{ + "commodities": []any{map[string]any{"id": "item-1", "failure_reason": "visual reason"}}, + }, nil).Once() + + env.RegisterWorkflowWithOptions(GraphInterpreterWorkflow, workflow.RegisterOptions{Name: "GraphInterpreterWorkflow"}) + env.SetStartWorkflowOptions(client.StartWorkflowOptions{ID: "parallel-conflict-test-1"}) + + env.ExecuteWorkflow(GraphInterpreterWorkflow, def, map[string]any{}) + + s.True(env.IsWorkflowCompleted()) + s.NoError(env.GetWorkflowError()) + + var result WorkflowInstance + s.NoError(env.GetWorkflowResult(&result)) + + items, ok := result.WorkflowVariables["commodities"].([]any) + s.Require().True(ok) + s.Require().Len(items, 1) + item, _ := items[0].(map[string]any) + + // psplit's outgoing edges are e3 (-> lab_task) and e4 (-> visual_task); sorted by edge ID, + // e3 is applied before e4, so visual_task's (later) value wins the conflicting field. + s.Equal("visual reason", item["failure_reason"], + "a genuinely conflicting field resolves via the fixed, documented branch order — not a merge algorithm decision") +} + +func (s *ParallelGatewayTestSuite) TestParallelSplit_MergeByID_ItemMissingID_Fails() { + env := s.NewTestWorkflowEnvironment() + + acts := &Activities{} + env.RegisterActivityWithOptions(acts.ExecuteTaskActivity, activity.RegisterOptions{Name: "ExecuteTaskActivity"}) + env.RegisterActivityWithOptions(acts.WorkflowCompletedActivity, activity.RegisterOptions{Name: "WorkflowCompletedActivity"}) + mockWorkflowCompletedIgnoringChildren(env) + + def := buildParallelMergeWorkflow(map[string]string{"commodities": "id"}) + + // An item is missing the required "id" field + env.OnActivity("ExecuteTaskActivity", mock.Anything, "LOAD_ITEMS", mock.Anything). + Return(map[string]any{ + "items": []any{ + map[string]any{"name": "No ID item"}, + }, + }, nil).Once() + + passThrough := func(_ context.Context, _ string, inputs map[string]any) (map[string]any, error) { + return map[string]any{"commodities": inputs["commodities"]}, nil + } + env.OnActivity("ExecuteTaskActivity", mock.Anything, "LAB_TASK", mock.Anything).Return(passThrough) + env.OnActivity("ExecuteTaskActivity", mock.Anything, "VISUAL_TASK", mock.Anything).Return(passThrough) + + env.RegisterDelayedCallback(func() { + val, err := env.QueryWorkflow("GetStatus") + s.Require().NoError(err) + var instance WorkflowInstance + s.Require().NoError(val.Get(&instance)) + s.Equal(NodeStatusAwaitingAdmin, instance.NodeInfo["psplit"].Status) + + env.SignalWorkflow(AdminResolutionSignalName, AdminResolutionSignal{ + NodeID: "psplit", + Action: AdminActionAbort, + }) + }, 2*time.Second) + + env.RegisterWorkflowWithOptions(GraphInterpreterWorkflow, workflow.RegisterOptions{Name: "GraphInterpreterWorkflow"}) + env.SetStartWorkflowOptions(client.StartWorkflowOptions{ID: "missing-id-test"}) + + env.ExecuteWorkflow(GraphInterpreterWorkflow, def, map[string]any{}) + + s.True(env.IsWorkflowCompleted()) + err := env.GetWorkflowError() + s.Error(err) + s.Contains(err.Error(), "missing required ID field \"id\"") +} + +func (s *ParallelGatewayTestSuite) TestParallelSplit_MergeByID_InvalidItemType_Fails() { + env := s.NewTestWorkflowEnvironment() + + acts := &Activities{} + env.RegisterActivityWithOptions(acts.ExecuteTaskActivity, activity.RegisterOptions{Name: "ExecuteTaskActivity"}) + env.RegisterActivityWithOptions(acts.WorkflowCompletedActivity, activity.RegisterOptions{Name: "WorkflowCompletedActivity"}) + mockWorkflowCompletedIgnoringChildren(env) + + def := buildParallelMergeWorkflow(map[string]string{"commodities": "id"}) + + // LOAD_ITEMS returns invalid non-slice items + env.OnActivity("ExecuteTaskActivity", mock.Anything, "LOAD_ITEMS", mock.Anything). + Return(map[string]any{ + "items": "not-a-slice", + }, nil).Once() + + passThrough := func(_ context.Context, _ string, inputs map[string]any) (map[string]any, error) { + return map[string]any{"commodities": inputs["commodities"]}, nil + } + env.OnActivity("ExecuteTaskActivity", mock.Anything, "LAB_TASK", mock.Anything).Return(passThrough) + env.OnActivity("ExecuteTaskActivity", mock.Anything, "VISUAL_TASK", mock.Anything).Return(passThrough) + + env.RegisterDelayedCallback(func() { + val, err := env.QueryWorkflow("GetStatus") + s.Require().NoError(err) + var instance WorkflowInstance + s.Require().NoError(val.Get(&instance)) + s.Equal(NodeStatusAwaitingAdmin, instance.NodeInfo["psplit"].Status) + + env.SignalWorkflow(AdminResolutionSignalName, AdminResolutionSignal{ + NodeID: "psplit", + Action: AdminActionAbort, + }) + }, 2*time.Second) + + env.RegisterWorkflowWithOptions(GraphInterpreterWorkflow, workflow.RegisterOptions{Name: "GraphInterpreterWorkflow"}) + env.SetStartWorkflowOptions(client.StartWorkflowOptions{ID: "invalid-slice-test"}) + + env.ExecuteWorkflow(GraphInterpreterWorkflow, def, map[string]any{}) + + s.True(env.IsWorkflowCompleted()) + err := env.GetWorkflowError() + s.Error(err) + s.Contains(err.Error(), "invalid items") +} + +func (s *ParallelGatewayTestSuite) TestParallelValidation_MergeByID_EmptyIDField_Fails() { + def := buildParallelMergeWorkflow(map[string]string{"commodities": ""}) + err := ValidateParallelGateways(def) + s.Error(err) + s.Contains(err.Error(), "has empty ID field") +} diff --git a/workflow/parallel_validate.go b/workflow/parallel_validate.go new file mode 100644 index 0000000..55a7ba1 --- /dev/null +++ b/workflow/parallel_validate.go @@ -0,0 +1,87 @@ +// SPDX-License-Identifier: Apache-2.0 +// Copyright (c) 2026 Lanka Software Foundation + +package engine + +import ( + "fmt" + "strings" +) + +// ValidateParallelGateways checks structural invariants and topological containment for +// PARALLEL_SPLIT / PARALLEL_JOIN pairs in the workflow definition, mirroring +// ValidateBatchGateways for BATCH_SPLIT / BATCH_JOIN. Call this at parse-time or at the start +// of GraphInterpreterWorkflow before execution begins. +func ValidateParallelGateways(def WorkflowDefinition) error { + nodesByID := make(map[string]*Node, len(def.Nodes)) + parallelSplits := make(map[string]*Node) + parallelJoins := make(map[string]*Node) + joinToSplit := make(map[string]string) + + for i, node := range def.Nodes { + nodesByID[node.ID] = &def.Nodes[i] + if node.Type != NodeTypeGateway { + continue + } + switch node.GatewayType { + case GatewayTypeParallelSplit: + parallelSplits[node.ID] = &def.Nodes[i] + case GatewayTypeParallelJoin: + if node.ParallelJoin == nil || node.ParallelJoin.GatewayNodeID == "" { + return fmt.Errorf("PARALLEL_JOIN node %q: parallel_join.gateway_node_id is required", node.ID) + } + for varPath, idField := range node.ParallelJoin.MergeByID { + if strings.Contains(varPath, ".") { + return fmt.Errorf("PARALLEL_JOIN node %q: merge_by_id key %q must be a top-level workflow variable (nested dot-paths are not supported)", node.ID, varPath) + } + if idField == "" { + return fmt.Errorf("PARALLEL_JOIN node %q: merge_by_id key %q has empty ID field", node.ID, varPath) + } + } + parallelJoins[node.ID] = &def.Nodes[i] + joinToSplit[node.ID] = node.ParallelJoin.GatewayNodeID + } + } + + // 1. Every PARALLEL_SPLIT must have exactly one paired PARALLEL_JOIN. + pairedJoins := make(map[string]string) // splitNodeID -> joinNodeID + for joinID, splitID := range joinToSplit { + if _, exists := parallelSplits[splitID]; !exists { + return fmt.Errorf("PARALLEL_JOIN node %q references non-existent PARALLEL_SPLIT node %q", joinID, splitID) + } + if existingJoin, alreadyPaired := pairedJoins[splitID]; alreadyPaired { + return fmt.Errorf("PARALLEL_SPLIT node %q has multiple PARALLEL_JOIN nodes: %q and %q", splitID, existingJoin, joinID) + } + pairedJoins[splitID] = joinID + } + + for splitID := range parallelSplits { + if _, paired := pairedJoins[splitID]; !paired { + return fmt.Errorf("PARALLEL_SPLIT node %q has no paired PARALLEL_JOIN", splitID) + } + } + + // 2. Build forward and reverse edge mappings. + forwardEdges := make(map[string][]Edge) + reverseEdges := make(map[string][]string) + for _, edge := range def.Edges { + forwardEdges[edge.SourceID] = append(forwardEdges[edge.SourceID], edge) + reverseEdges[edge.TargetID] = append(reverseEdges[edge.TargetID], edge.SourceID) + } + + // 3. PARALLEL_JOIN is a converging gateway; it cannot have more than 1 outgoing edge. + for joinID := range parallelJoins { + if count := len(forwardEdges[joinID]); count > 1 { + return fmt.Errorf("PARALLEL_JOIN node %q cannot have more than 1 outgoing edge, got %d", joinID, count) + } + } + + // 4. Sub-graph topological containment, same shape as the batch gateway check. + for splitID, joinID := range pairedJoins { + if err := validateGatewayRegion(splitID, joinID, GatewayTypeParallelSplit, "PARALLEL_SPLIT", "PARALLEL_JOIN", "parallel region", pairedJoins, nodesByID, forwardEdges, reverseEdges); err != nil { + return err + } + } + + return nil +} diff --git a/workflow/workflow.go b/workflow/workflow.go index 96300a2..40d8002 100644 --- a/workflow/workflow.go +++ b/workflow/workflow.go @@ -40,6 +40,9 @@ func GraphInterpreterWorkflow(ctx workflow.Context, def WorkflowDefinition, init if err := ValidateBatchGateways(def); err != nil { return nil, fmt.Errorf("workflow definition validation failed: %w", err) } + if err := ValidateParallelGateways(def); err != nil { + return nil, fmt.Errorf("workflow definition validation failed: %w", err) + } instance := &WorkflowInstance{ ID: workflow.GetInfo(ctx).WorkflowExecution.ID, @@ -436,46 +439,10 @@ func (g *graphInterpreter) handleGatewayNode(ctx workflow.Context, nodeInfo *Nod return fmt.Errorf("no matching conditions found at exclusive gateway %s", node.ID) case GatewayTypeParallelSplit: - nodeInfo.Status = NodeStatusCompleted - nodeInfo.UpdatedAt = workflow.Now(ctx) - var futures []workflow.Future - for _, e := range outEdges { - match, err := EvaluateCondition(e.Condition, g.instance.WorkflowVariables) - if err != nil { - return err - } - if match { - f, s := workflow.NewFuture(ctx) - edge := e // Capture locally for coroutine - workflow.Go(ctx, func(c workflow.Context) { - err := g.transitionTo(c, edge) - s.Set(nil, err) - }) - futures = append(futures, f) - } - } - for _, f := range futures { - if err := f.Get(ctx, nil); err != nil { - return err - } - } - return nil + return g.handleParallelSplitGateway(ctx, nodeInfo, node, outEdges) case GatewayTypeParallelJoin: - for _, e := range inEdges { - if g.edgeTokens[e.ID] <= 0 { - return nil // Wait for other branches - } - } - for _, e := range inEdges { - g.edgeTokens[e.ID]-- // Consume tokens - } - if len(outEdges) > 0 { - nodeInfo.Status = NodeStatusCompleted - nodeInfo.UpdatedAt = workflow.Now(ctx) - return g.transitionTo(ctx, outEdges[0]) - } - return nil + return g.handleParallelJoinGateway(ctx, nodeInfo, node, outEdges) case GatewayTypeExclusiveJoin: consumed := false