Skip to content

Commit fd6f270

Browse files
committed
Fix Grid and Graph 500 errors for cyclic TaskGroup dependencies
Since 3.3.1 the Grid and Graph views fail with HTTP 500 for Dags whose TaskGroups depend on each other when each group is treated as a single unit, even though the task graph is acyclic and the Dag parses and runs normally. Filtering a view by a root task can create the same group-level cycle in a Dag that has none, so the views have to tolerate these cycles whether or not such Dags are later rejected at parse time.
1 parent a0f9f7c commit fd6f270

4 files changed

Lines changed: 209 additions & 12 deletions

File tree

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

Lines changed: 79 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -236,10 +236,13 @@ def topological_sort(
236236
"""
237237
Sort children topologically — a task always comes after its upstream dependencies.
238238
239-
See ``TaskGroup.topological_sort`` in task-sdk for the algorithm. Cycles are
240-
treated as corrupt input: ``DAG.check_cycle`` rejects cyclic Dags before
241-
serialization, so a cycle reaching this code indicates malformed serialized data,
242-
and we raise ``ValueError`` rather than silently looping forever.
239+
See ``TaskGroup.topological_sort`` in task-sdk for the algorithm. Unlike the task-sdk
240+
variant, a cycle between siblings does not raise. ``DAG.check_cycle`` rejects task-level
241+
cycles, but a task with no upstream inside its own group counts as a root of that group,
242+
so edges routed through tasks outside the group can still make siblings depend on each
243+
other. ``partial_subset`` can also create such a cycle by dropping a task's in-group
244+
upstream. Grid and Graph must render these Dags, so the siblings on a cycle are ordered
245+
as one unit instead (see ``_sort_cyclic_projection``).
243246
"""
244247
children = self.children
245248
if not children:
@@ -329,14 +332,21 @@ def _sweep_projection(self, nodes: list[DAGNode], projected: list[tuple[int, ...
329332
emitted[i] = 1
330333
order_append(nodes[i])
331334
if len(next_pending) == len(pending):
332-
raise ValueError(f"A cyclic dependency occurred in dag: {self.dag_id}")
335+
return self._sort_cyclic_projection(nodes, projected)
333336
pending = next_pending
334337
return order
335338

336339
def _sort_via_pass_numbering(
337340
self, nodes: list[DAGNode], projected: list[tuple[int, ...]]
338341
) -> list[DAGNode]:
339-
n = len(nodes)
342+
sorted_indices = self._compute_pass_order(projected)
343+
if len(sorted_indices) != len(nodes):
344+
return self._sort_cyclic_projection(nodes, projected)
345+
return [nodes[i] for i in sorted_indices]
346+
347+
@staticmethod
348+
def _compute_pass_order(projected: list[tuple[int, ...]]) -> list[int]:
349+
n = len(projected)
340350
in_degree = [len(deps) for deps in projected]
341351
successors: list[list[int]] = [[] for _ in range(n)]
342352
for i, deps in enumerate(projected):
@@ -345,7 +355,7 @@ def _sort_via_pass_numbering(
345355

346356
pass_of = [0] * n
347357
queue: deque[int] = deque(i for i in range(n) if in_degree[i] == 0)
348-
processed = 0
358+
processed: list[int] = []
349359
while queue:
350360
i = queue.popleft()
351361
my_pass = 1
@@ -357,17 +367,74 @@ def _sort_via_pass_numbering(
357367
elif d_pass + 1 > my_pass:
358368
my_pass = d_pass + 1
359369
pass_of[i] = my_pass
360-
processed += 1
370+
processed.append(i)
361371
for s in successors[i]:
362372
in_degree[s] -= 1
363373
if in_degree[s] == 0:
364374
queue.append(s)
365375

366-
if processed != n:
367-
raise ValueError(f"A cyclic dependency occurred in dag: {self.dag_id}")
376+
# Children on or downstream of a cycle never reach in-degree 0 and are left out.
377+
return sorted(processed, key=lambda i: (pass_of[i], i))
368378

369-
sorted_indices = sorted(range(n), key=lambda i: (pass_of[i], i))
370-
return [nodes[i] for i in sorted_indices]
379+
def _sort_cyclic_projection(
380+
self, nodes: list[DAGNode], projected: list[tuple[int, ...]]
381+
) -> list[DAGNode]:
382+
# Each component is one unit placed by its first child; units follow the same pass
383+
# ordering as the acyclic path, and a unit's children keep insertion order.
384+
component_of = self._find_projection_components(projected)
385+
count = max(component_of) + 1
386+
members: list[list[int]] = [[] for _ in range(count)]
387+
component_deps: list[set[int]] = [set() for _ in range(count)]
388+
for i, deps in enumerate(projected):
389+
c = component_of[i]
390+
members[c].append(i)
391+
component_deps[c].update(component_of[d] for d in deps if component_of[d] != c)
392+
component_order = self._compute_pass_order([tuple(deps) for deps in component_deps])
393+
return [nodes[i] for c in component_order for i in members[c]]
394+
395+
@staticmethod
396+
def _find_projection_components(projected: list[tuple[int, ...]]) -> list[int]:
397+
"""Return each child's strongly connected component, numbered in order of its first child."""
398+
n = len(projected)
399+
successors: list[list[int]] = [[] for _ in range(n)]
400+
for i, deps in enumerate(projected):
401+
for d in deps:
402+
successors[d].append(i)
403+
404+
# Kosaraju: finish order along successors, then collect components along dependencies.
405+
visited = bytearray(n)
406+
finish_order: list[int] = []
407+
for start in range(n):
408+
if visited[start]:
409+
continue
410+
visited[start] = 1
411+
stack: list[tuple[int, Iterator[int]]] = [(start, iter(successors[start]))]
412+
while stack:
413+
node, remaining = stack[-1]
414+
for s in remaining:
415+
if not visited[s]:
416+
visited[s] = 1
417+
stack.append((s, iter(successors[s])))
418+
break
419+
else:
420+
stack.pop()
421+
finish_order.append(node)
422+
423+
root_of = [-1] * n
424+
for start in reversed(finish_order):
425+
if root_of[start] != -1:
426+
continue
427+
root_of[start] = start
428+
to_visit = [start]
429+
while to_visit:
430+
node = to_visit.pop()
431+
for d in projected[node]:
432+
if root_of[d] == -1:
433+
root_of[d] = start
434+
to_visit.append(d)
435+
436+
numbering: dict[int, int] = {}
437+
return [numbering.setdefault(root, len(numbering)) for root in root_of]
371438

372439
def add(self, node: DAGNode) -> DAGNode:
373440
# Set the TG first, as setting it might change the return value of node_id!

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

Lines changed: 26 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1063,6 +1063,32 @@ def test_task_converted_to_task_group_doesnt_crash(self, session, dag_maker, tes
10631063
{"id": "task_b", "label": "task_b"},
10641064
]
10651065

1066+
def test_structure_filter_creating_task_group_cycle(self, dag_maker, test_client, session):
1067+
"""Dropping ``g.guard`` makes ``g.second`` a root of ``g``, so ``g`` and ``bridge`` depend on each other."""
1068+
with dag_maker(dag_id="filtered_task_group_cycle", serialized=True, session=session):
1069+
with TaskGroup(group_id="g"):
1070+
first = EmptyOperator(task_id="first")
1071+
second = EmptyOperator(task_id="second")
1072+
guard = EmptyOperator(task_id="guard")
1073+
guard >> second
1074+
bridge = EmptyOperator(task_id="bridge")
1075+
first >> bridge >> second
1076+
dag_maker.sync_dagbag_to_db()
1077+
1078+
response = test_client.get(
1079+
"/grid/structure/filtered_task_group_cycle?root=g.first&include_downstream=true"
1080+
)
1081+
1082+
assert response.status_code == 200
1083+
assert response.json() == [
1084+
{"id": "bridge", "label": "bridge"},
1085+
{
1086+
"id": "g",
1087+
"label": "g",
1088+
"children": [{"id": "g.first", "label": "first"}, {"id": "g.second", "label": "second"}],
1089+
},
1090+
]
1091+
10661092
# Tests for root, include_upstream, and include_downstream parameters
10671093
@pytest.mark.parametrize(
10681094
("params", "expected_task_ids", "description"),

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

Lines changed: 22 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -393,6 +393,28 @@ def test_mapped_operator_in_task_group(self, dag_maker, test_client, session):
393393
assert mapped_in_group["is_mapped"] is True
394394
assert mapped_in_group["operator"] == "PythonOperator"
395395

396+
def test_filter_creating_task_group_cycle(self, dag_maker, test_client, session):
397+
"""Dropping ``g.guard`` makes ``g.second`` a root of ``g``, so ``g`` and ``bridge`` depend on each other."""
398+
with dag_maker(dag_id="filtered_task_group_cycle", serialized=True, session=session):
399+
with TaskGroup(group_id="g"):
400+
first = EmptyOperator(task_id="first")
401+
second = EmptyOperator(task_id="second")
402+
guard = EmptyOperator(task_id="guard")
403+
guard >> second
404+
bridge = EmptyOperator(task_id="bridge")
405+
first >> bridge >> second
406+
dag_maker.sync_dagbag_to_db()
407+
408+
response = test_client.get(
409+
"/structure/structure_data",
410+
params={"dag_id": "filtered_task_group_cycle", "root": "g.first", "include_downstream": True},
411+
)
412+
413+
assert response.status_code == 200
414+
nodes = response.json()["nodes"]
415+
assert [node["id"] for node in nodes] == ["bridge", "g"]
416+
assert [child["id"] for child in nodes[1]["children"]] == ["g.first", "g.second"]
417+
396418
def test_ui_colors_passed_through_to_graph(self, dag_maker, test_client, session):
397419
"""Both raw hex colors and Chakra palette tokens reach the graph unchanged, for operators and groups."""
398420

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

Lines changed: 82 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1351,6 +1351,88 @@ def spy(self, nodes, projected):
13511351
assert position[f"r{i}"] < position[f"r{i + 1}"]
13521352

13531353

1354+
def _make_sibling_groups_cycle():
1355+
with DAG("sibling_groups_cycle", schedule=None, start_date=DEFAULT_DATE) as dag:
1356+
start = EmptyOperator(task_id="start")
1357+
with TaskGroup("group1"):
1358+
a1 = EmptyOperator(task_id="a1")
1359+
a2 = EmptyOperator(task_id="a2")
1360+
with TaskGroup("group2"):
1361+
b1 = EmptyOperator(task_id="b1")
1362+
b2 = EmptyOperator(task_id="b2")
1363+
end = EmptyOperator(task_id="end")
1364+
start >> [a1, b2]
1365+
a1 >> b1
1366+
b2 >> a2
1367+
[a2, b1] >> end
1368+
return dag
1369+
1370+
1371+
def _make_group_bridged_by_outside_task():
1372+
with DAG("group_bridged_by_outside_task", schedule=None, start_date=DEFAULT_DATE) as dag:
1373+
with TaskGroup("group"):
1374+
a = EmptyOperator(task_id="a")
1375+
b = EmptyOperator(task_id="b")
1376+
bridge = EmptyOperator(task_id="bridge")
1377+
a >> bridge >> b
1378+
return dag
1379+
1380+
1381+
def _make_three_group_ring():
1382+
with DAG("three_group_ring", schedule=None, start_date=DEFAULT_DATE) as dag:
1383+
groups = {}
1384+
for group_id in ("g0", "g1", "g2"):
1385+
with TaskGroup(group_id):
1386+
groups[group_id] = (EmptyOperator(task_id="first"), EmptyOperator(task_id="second"))
1387+
groups["g1"][0] >> groups["g0"][1]
1388+
groups["g2"][0] >> groups["g1"][1]
1389+
groups["g0"][0] >> groups["g2"][1]
1390+
return dag
1391+
1392+
1393+
@pytest.mark.parametrize(
1394+
("make_dag", "expected"),
1395+
[
1396+
pytest.param(
1397+
_make_sibling_groups_cycle,
1398+
{
1399+
None: ["start", "group1", "group2", "end"],
1400+
"group1": ["group1.a1", "group1.a2"],
1401+
"group2": ["group2.b1", "group2.b2"],
1402+
},
1403+
id="sibling-groups",
1404+
),
1405+
pytest.param(
1406+
_make_group_bridged_by_outside_task,
1407+
{None: ["bridge", "group"], "group": ["group.a", "group.b"]},
1408+
id="group-bridged-by-outside-task",
1409+
),
1410+
# Two of the three siblings depend on a later sibling, which selects pass numbering.
1411+
pytest.param(
1412+
_make_three_group_ring,
1413+
{
1414+
None: ["g0", "g1", "g2"],
1415+
"g0": ["g0.first", "g0.second"],
1416+
"g1": ["g1.first", "g1.second"],
1417+
"g2": ["g2.first", "g2.second"],
1418+
},
1419+
id="three-group-ring",
1420+
),
1421+
],
1422+
)
1423+
def test_topological_sort_serialized_task_group_cycle(make_dag, expected):
1424+
"""Siblings that depend on each other only at the group level are ordered instead of raising."""
1425+
dag = make_dag()
1426+
dag.check_cycle()
1427+
serialized = create_scheduler_dag(dag)
1428+
group_dict = serialized.task_group.get_task_group_dict()
1429+
1430+
assert {
1431+
group_id: [node.node_id for node in group.topological_sort(group_dict=group_dict)]
1432+
for group_id, group in group_dict.items()
1433+
} == expected
1434+
1435+
13541436
def test_task_group_arrow_with_setup_group():
13551437
with DAG(dag_id="setup_group_teardown_group") as dag:
13561438
with TaskGroup("group_1") as g1:

0 commit comments

Comments
 (0)