fix: complete orphaned TaskGroup nodes stuck Running. Fixes #16450 (cherry-pick #16454 for 4.0) (#16465)

Signed-off-by: Alan Clucas <alan@clucas.org>
Co-authored-by: Alan Clucas <alan@clucas.org>
Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

Author: 3 people

workflow/controller/dag.go

@@ -57,6 +57,11 @@ type dagContext struct {
 	// are only computed once per operation
 	dependsLogic map[string]string
 
+	// taskGroupsToComplete collects the names of TaskGroup nodes that assessDAGPhase
+	// found stuck Running with all of their children fulfilled, mapped to the phase
+	// they should complete with. executeDAG marks them once assessment is done.
+	taskGroupsToComplete map[string]wfv1.NodePhase
+
 	// used for logging in the dag
 	log logging.Logger
 }
@@ -169,11 +174,23 @@ func (d *dagContext) assessDAGPhase(ctx context.Context, targetTasks []string, n
 		branchPhase := curr.phase
 
 		if !node.Fulfilled() {
-			return wfv1.NodeRunning, nil
-		}
-
-		// Only overwrite the branchPhase if this node completed. (If it didn't we can just inherit our parent's branchPhase).
-		if node.Completed() {
+			// A fan-out TaskGroup can be left Running with every expanded child
+			// already fulfilled, for example when a retry resets the group but never
+			// re-runs it because its dependents have already completed. executeDAGTask
+			// only visits unfulfilled tasks, so it never revisits such a group, which
+			// would then hold the DAG Running forever. Complete it from its children
+			// instead of blocking here.
+			groupPhase, ok := completableTaskGroupPhase(node, nodes)
+			if !ok {
+				return wfv1.NodeRunning, nil
+			}
+			if d.taskGroupsToComplete == nil {
+				d.taskGroupsToComplete = make(map[string]wfv1.NodePhase)
+			}
+			d.taskGroupsToComplete[node.Name] = groupPhase
+			branchPhase = groupPhase
+		} else if node.Completed() {
+			// Only overwrite the branchPhase if this node completed. (If it didn't we can just inherit our parent's branchPhase).
 			branchPhase = node.Phase
 		}
 
@@ -227,6 +244,28 @@ func (d *dagContext) assessDAGPhase(ctx context.Context, targetTasks []string, n
 	return result, nil
 }
 
+// completableTaskGroupPhase reports whether node is a TaskGroup that is not yet
+// fulfilled even though all of its expanded children are, and if so the phase it
+// should complete with (Succeeded unless a child failed or errored, matching the
+// aggregation executeDAGTask uses). Such a group is never revisited by
+// executeDAGTask, so it must be completed during DAG assessment.
+func completableTaskGroupPhase(node *wfv1.NodeStatus, nodes wfv1.Nodes) (wfv1.NodePhase, bool) {
+	if node.Type != wfv1.NodeTypeTaskGroup || len(node.Children) == 0 {
+		return "", false
+	}
+	phase := wfv1.NodeSucceeded
+	for _, childID := range node.Children {
+		child, err := nodes.Get(childID)
+		if err != nil || !child.Fulfilled() {
+			return "", false
+		}
+		if child.FailedOrError() {
+			phase = child.Phase
+		}
+	}
+	return phase, true
+}
+
 func (woc *wfOperationCtx) executeDAG(ctx context.Context, nodeName string, tmplCtx *templateresolution.TemplateContext, templateScope string, tmpl *wfv1.Template, orgTmpl wfv1.TemplateReferenceHolder, opts *executeTemplateOpts) (*wfv1.NodeStatus, error) {
 
 	node, err := woc.wf.GetNodeByName(nodeName)
@@ -320,6 +359,13 @@ func (woc *wfOperationCtx) executeDAG(ctx context.Context, nodeName string, tmpl
 		return nil, err
 	}
 
+	// Complete any orphaned TaskGroups that assessment found stuck Running with all
+	// children fulfilled. Done regardless of the overall DAG phase so a group is
+	// healed even while other tasks are still legitimately running.
+	for name, phase := range dagCtx.taskGroupsToComplete {
+		woc.markNodePhase(ctx, name, phase)
+	}
+
 	switch dagPhase {
 	case wfv1.NodeRunning:
 		return node, nil

workflow/controller/dag_test.go

@@ -4589,3 +4589,61 @@ func TestDAGSkippedInlineExpressionFallback(t *testing.T) {
 	require.NotNil(t, in.Value)
 	assert.Equal(t, "inline-fallback", in.Value.String())
 }
+
+// TestDAGOrphanedTaskGroupCompletes verifies that a fan-out TaskGroup left Running
+// with every expanded child already fulfilled — the state a retry can produce when
+// it resets the group but never re-runs it, because its dependents already
+// completed — does not hold the DAG Running forever. executeDAGTask never revisits
+// such a group, so the controller must complete it from its children during DAG
+// assessment.
+func TestDAGOrphanedTaskGroupCompletes(t *testing.T) {
+	wf := wfv1.MustUnmarshalWorkflow(`
+apiVersion: argoproj.io/v1alpha1
+kind: Workflow
+metadata:
+  name: dag-orphaned-taskgroup
+  namespace: argo
+spec:
+  entrypoint: main
+  templates:
+  - name: main
+    dag:
+      tasks:
+      - name: fanout
+        template: echo
+        withItems: [a, b]
+      - name: leaf
+        template: echo
+        depends: fanout
+  - name: echo
+    container:
+      image: alpine:3.23
+      command: [sh, -c, "exit 0"]
+`)
+	wf.Status.Phase = wfv1.WorkflowRunning
+	wf.Status.StartedAt = metav1.Now()
+
+	root := wf.Name
+	fanout := root + ".fanout"
+	child0 := root + ".fanout(0:a)"
+	child1 := root + ".fanout(1:b)"
+	leaf := root + ".leaf"
+	id := wf.NodeID
+
+	// The fan-out TaskGroup is Running while its children and the downstream leaf
+	// have all Succeeded — an orphaned group that nothing will otherwise complete.
+	wf.Status.Nodes = wfv1.Nodes{
+		id(root):   {ID: id(root), Name: root, Type: wfv1.NodeTypeDAG, Phase: wfv1.NodeRunning, TemplateName: "main", Children: []string{id(fanout)}},
+		id(fanout): {ID: id(fanout), Name: fanout, Type: wfv1.NodeTypeTaskGroup, Phase: wfv1.NodeRunning, BoundaryID: id(root), TemplateName: "echo", Children: []string{id(child0), id(child1)}},
+		id(child0): {ID: id(child0), Name: child0, Type: wfv1.NodeTypePod, Phase: wfv1.NodeSucceeded, BoundaryID: id(root), TemplateName: "echo", Children: []string{id(leaf)}},
+		id(child1): {ID: id(child1), Name: child1, Type: wfv1.NodeTypePod, Phase: wfv1.NodeSucceeded, BoundaryID: id(root), TemplateName: "echo", Children: []string{id(leaf)}},
+		id(leaf):   {ID: id(leaf), Name: leaf, Type: wfv1.NodeTypePod, Phase: wfv1.NodeSucceeded, BoundaryID: id(root), TemplateName: "echo"},
+	}
+
+	ctx := logging.TestContext(t.Context())
+	woc := newWoc(ctx, *wf)
+	woc.operate(ctx)
+
+	assert.Equal(t, wfv1.NodeSucceeded, woc.wf.Status.Nodes[id(fanout)].Phase, "orphaned TaskGroup should be completed from its children")
+	assert.Equal(t, wfv1.WorkflowSucceeded, woc.wf.Status.Phase, "workflow should complete")
+}