From 5222d9487364f7e5466067bfc2b8204c63a826c6 Mon Sep 17 00:00:00 2001 From: Eason09053360 <185830721+Eason09053360@users.noreply.github.com> Date: Tue, 22 Sep 2026 10:32:13 +0800 Subject: [PATCH] UI: Fix grid columns vanishing when a Dag version has been cleaned `_get_serdag` is declared `SerializedDAG | None` and logs when it returns None, but the caller guarded it with `if TYPE_CHECKING: assert serdag`, which never executes at runtime. Once `airflow db clean` removes a version's serialized Dag row, the resulting AttributeError is raised inside the `StreamingResponse` generator -- after the 200 has been sent. The client gets a silently truncated NDJSON body and loses every run queued behind the failing one, with no status code to detect it by. The guard was a real `raise HTTPException(404)` until #52302 replaced it with the no-op assert. A 404 is no longer available that far into the stream, and skipping the run would hide task instances that are still in the database, so an unresolvable version now falls through to the flat-node path the endpoint already uses for tasks the current structure no longer contains. That makes the missing serialized Dag an expected, handled condition, and the grid re-streams every visible run on each auto-refresh, so its log drops from error to warning -- matching the existing sites for the same condition in models/taskinstance.py. --- .../api_fastapi/core_api/routes/ui/grid.py | 29 ++++++----- .../core_api/routes/ui/test_grid.py | 52 ++++++++++++++++++- 2 files changed, 66 insertions(+), 15 deletions(-) diff --git a/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/grid.py b/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/grid.py index e4f9b02f7d1b0..151aa0ea1461a 100644 --- a/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/grid.py +++ b/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/grid.py @@ -104,7 +104,7 @@ def _get_serdag( if dag_version_id is not None: serdag = dag_bag.get_dag(dag_version_id, session=session) if serdag is None: - log.error("No serialized dag found", dag_id=dag_id, version_id=dag_version_id) + log.warning("No serialized dag found", dag_id=dag_id, version_id=dag_version_id) return serdag # Fallback: pre-3.0 upgrade — pick the oldest DagVersion for this dag_id. @@ -115,7 +115,7 @@ def _get_serdag( return None serdag = dag_bag.get_dag(oldest_version_id, session=session) if serdag is None: - log.error("No serialized dag found", dag_id=dag_id, version_id=oldest_version_id) + log.warning("No serialized dag found", dag_id=dag_id, version_id=oldest_version_id) return serdag @@ -429,21 +429,22 @@ def _build_ti_summaries( return None serdag = _get_serdag(dag_bag, dag_id, dag_version_id, session) - if TYPE_CHECKING: - assert serdag def get_node_summaries() -> Iterable[dict[str, Any]]: yielded_task_ids: set[str] = set() - for node, _ in _find_aggregates( - node=serdag.task_group, - parent_node=None, - ti_details=ti_details, - ): - if node["type"] in {"task", "mapped_task"}: - yielded_task_ids.add(node["task_id"]) - if node["type"] == "task": - node["child_states"] = None - yield node + # serdag is None once `airflow db clean` removes the version's serialized Dag row. Leaving + # yielded_task_ids empty then routes every task through the flat-node path below. + if serdag is not None: + for node, _ in _find_aggregates( + node=serdag.task_group, + parent_node=None, + ti_details=ti_details, + ): + if node["type"] in {"task", "mapped_task"}: + yielded_task_ids.add(node["task_id"]) + if node["type"] == "task": + node["child_states"] = None + yield node missing_task_ids = set(ti_details.keys()) - yielded_task_ids for task_id in sorted(missing_task_ids): detail = ti_details[task_id] diff --git a/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_grid.py b/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_grid.py index ad4450903df87..b67a4c7d8b789 100644 --- a/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_grid.py +++ b/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_grid.py @@ -23,7 +23,7 @@ import pendulum import pytest -from sqlalchemy import select +from sqlalchemy import delete, select from sqlalchemy.orm import Session from airflow._shared.timezones import timezone @@ -31,6 +31,7 @@ from airflow.models.dag_version import DagVersion from airflow.models.dagbag import DBDagBag from airflow.models.dagrun import DagRun, DagRunNote +from airflow.models.serialized_dag import SerializedDagModel from airflow.models.taskinstance import TaskInstance, TaskInstanceNote from airflow.providers.standard.operators.empty import EmptyOperator from airflow.providers.standard.operators.python import PythonOperator @@ -1260,6 +1261,55 @@ def test_grid_ti_summaries_stream_skips_missing_runs(self, session, test_client) assert len(summaries) == 1 assert summaries[0]["run_id"] == "run_1" + def test_grid_ti_summaries_stream_continues_past_cleaned_dag_version(self, session, test_client): + """A run whose version was cleaned still streams, and does not cut the runs after it.""" + cleaned_version = session.scalar( + select(DagVersion).where(DagVersion.dag_id == DAG_ID_5, DagVersion.version_number == 1) + ) + session.execute( + delete(SerializedDagModel).where(SerializedDagModel.dag_version_id == cleaned_version.id) + ) + session.commit() + + response = test_client.get( + f"/grid/ti_summaries/{DAG_ID_5}", params={"run_ids": ["run_5_1", "run_5_2"]} + ) + assert response.status_code == 200 + summaries = {summary["run_id"]: summary for summary in self._parse_ndjson(response)} + assert set(summaries) == {"run_5_1", "run_5_2"} + assert {node["task_id"] for node in summaries["run_5_1"]["task_instances"]} == { + "task_a", + "task_b", + "task_c", + "task_d", + "task_f", + } + # The run on the surviving version is unaffected. + assert {node["task_id"] for node in summaries["run_5_2"]["task_instances"]} == { + "task_a", + "task_b", + "task_c", + "task_d", + "task_e", + } + + def test_grid_ti_summaries_stream_without_serialized_dag_flattens_groups(self, session, test_client): + """With no structure to walk, a grouped Dag's tasks come back as flat leaf nodes.""" + session.execute(delete(SerializedDagModel).where(SerializedDagModel.dag_id == DAG_ID)) + session.commit() + + response = test_client.get(f"/grid/ti_summaries/{DAG_ID}", params={"run_ids": ["run_1"]}) + assert response.status_code == 200 + [summary] = self._parse_ndjson(response) + assert {node["task_id"]: node["state"] for node in summary["task_instances"]} == { + TASK_ID: "success", + "mapped_task_group.subtask": "success", + f"{TASK_GROUP_ID}.{MAPPED_TASK_ID}": "success", + f"{TASK_GROUP_ID}.{INNER_TASK_GROUP}.{INNER_TASK_GROUP_SUB_TASK}": "success", + MAPPED_TASK_ID_2: "success", + } + assert all(node["child_states"] is None for node in summary["task_instances"]) + def test_grid_ti_summaries_stream_empty_run_ids(self, session, test_client): """Streaming endpoint with no run_ids returns an empty body.""" session.commit()