Fix Grid and Graph 500 errors for cyclic TaskGroup dependencies

airflow-core/src/airflow/serialization/definitions/taskgroup.py

@@ -236,10 +236,8 @@ def topological_sort(
         """
         Sort children topologically — a task always comes after its upstream dependencies.
 
-        See ``TaskGroup.topological_sort`` in task-sdk for the algorithm. Cycles are
-        treated as corrupt input: ``DAG.check_cycle`` rejects cyclic Dags before
-        serialization, so a cycle reaching this code indicates malformed serialized data,
-        and we raise ``ValueError`` rather than silently looping forever.
+        See ``TaskGroup.topological_sort`` in task-sdk for the algorithm, including how siblings
+        that depend on each other in a cycle are ordered.
         """
         children = self.children
         if not children:
@@ -329,44 +327,16 @@ def _sweep_projection(self, nodes: list[DAGNode], projected: list[tuple[int, ...
                 emitted[i] = 1
                 order_append(nodes[i])
             if len(next_pending) == len(pending):
-                raise ValueError(f"A cyclic dependency occurred in dag: {self.dag_id}")
+                return self._sort_cyclic_projection(nodes, projected)
             pending = next_pending
         return order
 
     def _sort_via_pass_numbering(
         self, nodes: list[DAGNode], projected: list[tuple[int, ...]]
     ) -> list[DAGNode]:
-        n = len(nodes)
-        in_degree = [len(deps) for deps in projected]
-        successors: list[list[int]] = [[] for _ in range(n)]
-        for i, deps in enumerate(projected):
-            for d in deps:
-                successors[d].append(i)
-
-        pass_of = [0] * n
-        queue: deque[int] = deque(i for i in range(n) if in_degree[i] == 0)
-        processed = 0
-        while queue:
-            i = queue.popleft()
-            my_pass = 1
-            for d in projected[i]:
-                d_pass = pass_of[d]
-                if d < i:
-                    if d_pass > my_pass:
-                        my_pass = d_pass
-                elif d_pass + 1 > my_pass:
-                    my_pass = d_pass + 1
-            pass_of[i] = my_pass
-            processed += 1
-            for s in successors[i]:
-                in_degree[s] -= 1
-                if in_degree[s] == 0:
-                    queue.append(s)
-
-        if processed != n:
-            raise ValueError(f"A cyclic dependency occurred in dag: {self.dag_id}")
-
-        sorted_indices = sorted(range(n), key=lambda i: (pass_of[i], i))
+        sorted_indices = self._compute_pass_order(projected)
+        if len(sorted_indices) != len(nodes):
+            return self._sort_cyclic_projection(nodes, projected)
         return [nodes[i] for i in sorted_indices]
 
     def add(self, node: DAGNode) -> DAGNode:

airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_grid.py

@@ -1063,6 +1063,32 @@ def test_task_converted_to_task_group_doesnt_crash(self, session, dag_maker, tes
             {"id": "task_b", "label": "task_b"},
         ]
 
+    def test_structure_filter_creating_task_group_cycle(self, dag_maker, test_client, session):
+        """Dropping ``g.guard`` makes ``g.second`` a root of ``g``, so ``g`` and ``bridge`` depend on each other."""
+        with dag_maker(dag_id="filtered_task_group_cycle", serialized=True, session=session):
+            with TaskGroup(group_id="g"):
+                first = EmptyOperator(task_id="first")
+                second = EmptyOperator(task_id="second")
+                guard = EmptyOperator(task_id="guard")
+                guard >> second
+            bridge = EmptyOperator(task_id="bridge")
+            first >> bridge >> second
+        dag_maker.sync_dagbag_to_db()
+
+        response = test_client.get(
+            "/grid/structure/filtered_task_group_cycle?root=g.first&include_downstream=true"
+        )
+
+        assert response.status_code == 200
+        assert response.json() == [
+            {"id": "bridge", "label": "bridge"},
+            {
+                "id": "g",
+                "label": "g",
+                "children": [{"id": "g.first", "label": "first"}, {"id": "g.second", "label": "second"}],
+            },
+        ]
+
     # Tests for root, include_upstream, and include_downstream parameters
     @pytest.mark.parametrize(
         ("params", "expected_task_ids", "description"),

airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_structure.py

@@ -393,6 +393,28 @@ def test_mapped_operator_in_task_group(self, dag_maker, test_client, session):
         assert mapped_in_group["is_mapped"] is True
         assert mapped_in_group["operator"] == "PythonOperator"
 
+    def test_filter_creating_task_group_cycle(self, dag_maker, test_client, session):
+        """Dropping ``g.guard`` makes ``g.second`` a root of ``g``, so ``g`` and ``bridge`` depend on each other."""
+        with dag_maker(dag_id="filtered_task_group_cycle", serialized=True, session=session):
+            with TaskGroup(group_id="g"):
+                first = EmptyOperator(task_id="first")
+                second = EmptyOperator(task_id="second")
+                guard = EmptyOperator(task_id="guard")
+                guard >> second
+            bridge = EmptyOperator(task_id="bridge")
+            first >> bridge >> second
+        dag_maker.sync_dagbag_to_db()
+
+        response = test_client.get(
+            "/structure/structure_data",
+            params={"dag_id": "filtered_task_group_cycle", "root": "g.first", "include_downstream": True},
+        )
+
+        assert response.status_code == 200
+        nodes = response.json()["nodes"]
+        assert [node["id"] for node in nodes] == ["bridge", "g"]
+        assert [child["id"] for child in nodes[1]["children"]] == ["g.first", "g.second"]
+
     def test_ui_colors_passed_through_to_graph(self, dag_maker, test_client, session):
         """Both raw hex colors and Chakra palette tokens reach the graph unchanged, for operators and groups."""
 

airflow-core/tests/unit/utils/test_task_group.py

@@ -1351,6 +1351,140 @@ def spy(self, nodes, projected):
         assert position[f"r{i}"] < position[f"r{i + 1}"]
 
 
+# Deserialization orders a group's children by label, so these Dags declare children in label order
+# to give the Task SDK sort the same input as the serialized one.
+
+
+def _make_sibling_groups_cycle():
+    with DAG("sibling_groups_cycle", schedule=None, start_date=DEFAULT_DATE) as dag:
+        end = EmptyOperator(task_id="end")
+        with TaskGroup("group1"):
+            a1 = EmptyOperator(task_id="a1")
+            a2 = EmptyOperator(task_id="a2")
+        with TaskGroup("group2"):
+            b1 = EmptyOperator(task_id="b1")
+            b2 = EmptyOperator(task_id="b2")
+        start = EmptyOperator(task_id="start")
+        start >> [a1, b2]
+        a1 >> b1
+        b2 >> a2
+        [a2, b1] >> end
+    return dag
+
+
+def _make_group_bridged_by_outside_task():
+    with DAG("group_bridged_by_outside_task", schedule=None, start_date=DEFAULT_DATE) as dag:
+        bridge = EmptyOperator(task_id="bridge")
+        with TaskGroup("group"):
+            a = EmptyOperator(task_id="a")
+            b = EmptyOperator(task_id="b")
+        a >> bridge >> b
+    return dag
+
+
+def _make_bridged_group_among_siblings():
+    with DAG("bridged_group_among_siblings", schedule=None, start_date=DEFAULT_DATE) as dag:
+        after = EmptyOperator(task_id="after")
+        bridge = EmptyOperator(task_id="bridge")
+        with TaskGroup("group"):
+            a = EmptyOperator(task_id="a")
+            b = EmptyOperator(task_id="b")
+        for i in range(3):
+            EmptyOperator(task_id=f"x{i}")
+        a >> bridge >> b >> after
+    return dag
+
+
+def _make_nested_bridged_group():
+    with DAG("nested_bridged_group", schedule=None, start_date=DEFAULT_DATE) as dag:
+        with TaskGroup("outer"):
+            after = EmptyOperator(task_id="after")
+            bridge = EmptyOperator(task_id="bridge")
+            with TaskGroup("inner"):
+                a = EmptyOperator(task_id="a")
+                b = EmptyOperator(task_id="b")
+        a >> bridge >> b >> after
+    return dag
+
+
+def _make_three_group_ring():
+    with DAG("three_group_ring", schedule=None, start_date=DEFAULT_DATE) as dag:
+        groups = {}
+        for group_id in ("g0", "g1", "g2"):
+            with TaskGroup(group_id):
+                groups[group_id] = (EmptyOperator(task_id="first"), EmptyOperator(task_id="second"))
+        groups["g1"][0] >> groups["g0"][1]
+        groups["g2"][0] >> groups["g1"][1]
+        groups["g0"][0] >> groups["g2"][1]
+    return dag
+
+
+def _get_topological_orders(group_dict):
+    return {
+        group_id: [node.node_id for node in group.topological_sort(group_dict=group_dict)]
+        for group_id, group in group_dict.items()
+    }
+
+
+@pytest.mark.parametrize(
+    ("make_dag", "expected"),
+    [
+        pytest.param(
+            _make_sibling_groups_cycle,
+            {
+                None: ["start", "group1", "group2", "end"],
+                "group1": ["group1.a1", "group1.a2"],
+                "group2": ["group2.b1", "group2.b2"],
+            },
+            id="sibling-groups",
+        ),
+        pytest.param(
+            _make_group_bridged_by_outside_task,
+            {None: ["bridge", "group"], "group": ["group.a", "group.b"]},
+            id="group-bridged-by-outside-task",
+        ),
+        # Only two of six siblings depend on a later sibling, which keeps the sweep.
+        pytest.param(
+            _make_bridged_group_among_siblings,
+            {
+                None: ["bridge", "group", "x0", "x1", "x2", "after"],
+                "group": ["group.a", "group.b"],
+            },
+            id="bridged-group-among-siblings",
+        ),
+        # Two of outer's three children depend on a later sibling, which selects pass numbering.
+        pytest.param(
+            _make_nested_bridged_group,
+            {
+                None: ["outer"],
+                "outer": ["outer.bridge", "outer.inner", "outer.after"],
+                "outer.inner": ["outer.inner.a", "outer.inner.b"],
+            },
+            id="nested-bridged-group",
+        ),
+        # Two of the three siblings depend on a later sibling, which selects pass numbering.
+        pytest.param(
+            _make_three_group_ring,
+            {
+                None: ["g0", "g1", "g2"],
+                "g0": ["g0.first", "g0.second"],
+                "g1": ["g1.first", "g1.second"],
+                "g2": ["g2.first", "g2.second"],
+            },
+            id="three-group-ring",
+        ),
+    ],
+)
+def test_topological_sort_task_group_cycle(make_dag, expected):
+    """Siblings that depend on each other only at the group level are ordered instead of raising."""
+    dag = make_dag()
+    dag.check_cycle()
+    serialized = create_scheduler_dag(dag)
+
+    assert _get_topological_orders(serialized.task_group.get_task_group_dict()) == expected
+    assert _get_topological_orders(dag.task_group.get_task_group_dict()) == expected
+
+
 def test_task_group_arrow_with_setup_group():
     with DAG(dag_id="setup_group_teardown_group") as dag:
         with TaskGroup("group_1") as g1: