diff --git a/airflow-core/newsfragments/70113.bugfix.rst b/airflow-core/newsfragments/70113.bugfix.rst new file mode 100644 index 0000000000000..a7383671da696 --- /dev/null +++ b/airflow-core/newsfragments/70113.bugfix.rst @@ -0,0 +1 @@ +Fail queued DagRuns whose pinned dag version can no longer be resolved, instead of skipping them forever. diff --git a/airflow-core/src/airflow/jobs/scheduler_job_runner.py b/airflow-core/src/airflow/jobs/scheduler_job_runner.py index 9b9e85c07e353..0cd2677c7bc9c 100644 --- a/airflow-core/src/airflow/jobs/scheduler_job_runner.py +++ b/airflow-core/src/airflow/jobs/scheduler_job_runner.py @@ -2788,7 +2788,22 @@ def _update_state(dag: SerializedDAG, dag_run: DagRun): backfill_id = dag_run.backfill_id dag = dag_run.dag = cached_get_dag(dag_run) if not dag: - self.log.error("DAG '%s' not found in serialized_dag table", dag_run.dag_id) + # The serialized DAG for this run cannot be resolved. This happens when the run is + # pinned to a dag version (bundle_version set, created_dag_version_id populated) whose + # dag_version row was later deleted: the FK's ondelete="set null" clears + # created_dag_version_id and the pinned resolver does not fall back to the latest + # version. Without an explicit outcome the run is re-selected and skipped on every + # loop forever, with no failure signal. Fail it explicitly, mirroring how a SCHEDULED + # task instance whose serialized DAG cannot be found is failed elsewhere in the loop. + self.log.error( + "DAG '%s' for queued run %s could not be resolved " + "(created_dag_version_id=%s, bundle_version=%s); marking the run as failed", + dag_id, + run_id, + dag_run.created_dag_version_id, + dag_run.bundle_version, + ) + dag_run.set_state(DagRunState.FAILED) continue active_runs = active_runs_of_dags[(dag_id, backfill_id)] if backfill_id is not None: diff --git a/airflow-core/tests/unit/jobs/test_scheduler_job.py b/airflow-core/tests/unit/jobs/test_scheduler_job.py index dff3adea48b21..8f2fc346aac37 100644 --- a/airflow-core/tests/unit/jobs/test_scheduler_job.py +++ b/airflow-core/tests/unit/jobs/test_scheduler_job.py @@ -6015,6 +6015,60 @@ def test_start_dagruns(self, mock_get_backend, dag_maker, session): assert get_last_dagrun(dag.dag_id, session).creating_job_id == scheduler_job.id + def test_queued_dagrun_with_unresolvable_pinned_version_is_failed(self, dag_maker, session): + """ + A queued run whose pinned dag version can no longer be resolved is failed + explicitly instead of being skipped on every scheduler loop. + + Reproduces the state after the pinned dag_version row is deleted: the FK + ondelete="set null" clears created_dag_version_id while bundle_version + remains set, so the pinned resolver returns None and never falls back to + the latest version. The scheduler now marks the run FAILED rather than + re-selecting and skipping it forever. + """ + with dag_maker(dag_id="test_stuck_pinned_run"): + EmptyOperator(task_id="mytask") + dr = dag_maker.create_dagrun(run_type=DagRunType.MANUAL, state=State.QUEUED) + + # The run was created pinned to a bundle version... + dr.bundle_version = "0123456789abcdef" + # ...and the pinned dag_version row has since been deleted -> FK SET NULL + dr.created_dag_version_id = None + session.merge(dr) + session.flush() + + scheduler_job = Job() + self.job_runner = SchedulerJobRunner(job=scheduler_job, executors=[self.null_exec]) + + # Even across repeated loops the run reaches a terminal state instead of being skipped. + for _ in range(3): + self.job_runner._start_queued_dagruns(session) + session.flush() + + dr = session.scalars(select(DagRun).where(DagRun.dag_id == "test_stuck_pinned_run")).one() + assert dr.state == State.FAILED + + def test_queued_dagrun_without_bundle_version_falls_back_to_latest(self, dag_maker, session): + """ + Contrast case: same NULL created_dag_version_id, but without bundle_version + the resolver falls back to the latest serialized version and the run starts. + """ + with dag_maker(dag_id="test_unpinned_run_falls_back"): + EmptyOperator(task_id="mytask") + dr = dag_maker.create_dagrun(run_type=DagRunType.MANUAL, state=State.QUEUED) + dr.bundle_version = None + dr.created_dag_version_id = None + session.merge(dr) + session.flush() + + scheduler_job = Job() + self.job_runner = SchedulerJobRunner(job=scheduler_job, executors=[self.null_exec]) + self.job_runner._start_queued_dagruns(session) + session.flush() + + dr = session.scalars(select(DagRun).where(DagRun.dag_id == "test_unpinned_run_falls_back")).one() + assert dr.state == State.RUNNING + def test_extra_operator_links_not_loaded_in_scheduler_loop(self, dag_maker): """ Test that Operator links are not loaded inside the Scheduling Loop (that does not include