apache / apache/airflow

Scheduler silently skips queued DagRuns forever when their pinned dag version can no longer be resolved

Open
#70,056 3 comments 0 reactions 0 assignees View on GitHub
area:core area:scheduler kind:bug priority:high
Dominant language
Python
Stars
46.9k
Forks
17.8k
Avg merge
2d 9h
Merged PRs (30d)
472

Description

### Apache Airflow version

Observed in production on 3.1.8; the code path is unchanged on 3.3.0 and current main (`b628e46`).

### What happened

A `DagRun` created by `TriggerDagRunOperator` sat in `QUEUED` for 8+ hours. The scheduler logged, on every loop:

```
DAG 'YB_DUMMY' not found in serialized_dag table
```

while the DAG itself was healthy: not paused, no import errors, visible in the UI, and new runs of the same DAG worked. Run-scoped API endpoints for the stuck run returned 404 while latest-version endpoints returned 200. The parent DAG's `ExternalTaskSensor` treats only `success`/`failed` as terminal, so it poked the queued run until its own 10h timeout.

The chain:

1. The run was created **pinned**: `bundle_version` set, `created_dag_version_id` pointing at the dag version current at creation time. The FK is declared `ondelete="set null"` ([dagrun.py L301-303](https://github.com/apache/airflow/blob/main/airflow-core/src/airflow/models/dagrun.py#L301-L303)).
2. The pinned `dag_version` row was later deleted (the FK's `SET NULL` exists precisely because version rows can be deleted). `created_dag_version_id` became NULL **silently**; `bundle_version` stayed set.
3. `DBDagBag._version_from_dag_run` ([dagbag.py L211-216](https://github.com/apache/airflow/blob/main/airflow-core/src/airflow/models/dagbag.py#L211-L216)) only falls back to the latest version when `bundle_version` is **not** set. For a pinned run it returns `created_dag_version_id` — now NULL — so `get_dag_for_run()` returns `None`.
4. `_start_queued_dagruns` handles that with `log.error(...); continue` ([scheduler_job_runner.py, "DAG '%s' not found in serialized_dag table"](https://github.com/apache/airflow/blob/main/airflow-core/src/airflow/jobs/scheduler_job_runner.py)). The run is neither started nor failed, and is re-selected and re-skipped on every subsequent loop, forever. No timeout applies to a QUEUED run in this state.

So a single race between "run created & pinned" and "pinned version row deleted" produces a run that is permanently stuck with no failure signal, no metric, and an error message that points at the wrong cause (the DAG *is* in the serialized_dag table — just not the pinned version).

### What you think should happen instead

The scheduler already handles the sibling case for task instances in the same function: when the serialized dag for a SCHEDULED TI cannot be found, the TIs are set to FAILED and the loop moves on. Queued runs deserve the same explicit outcome. Options, in my order of preference:

1. **Fail the run explicitly** when its pinned version cannot be resolved (message naming the pinned `created_dag_version_id`/`bundle_version`), consistent with the TI path.
2. **Fall back to the latest serialized version**, matching unpinned behavior — simpler for users, but arguably violates version-pinning semantics.
3. At minimum, emit a dedicated metric/log so the permanent skip is observable.

I'm happy to submit a PR for option 1 (or whichever direction maintainers prefer).

### How to reproduce

Two tests against current main — both pass (verified on Linux CI at `b628e46`; the first one "passing" *is* the bug: the run stays QUEUED through repeated scheduler loops). Branch with the tests: https://github.com/kjh0623/airflow/tree/fix/queued-dagrun-stuck-unresolvable-version

```python
def test_queued_dagrun_with_unresolvable_pinned_version_is_stuck_forever(self, dag_maker, session):
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()

self.job_runner = SchedulerJobRunner(job=Job(), executors=[self.null_exec])
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.QUEUED # skipped every loop: never started, never failed

def test_queued_dagrun_without_bundle_version_falls_back_to_latest(self, dag_maker, session):
"""Contrast: same NULL created_dag_version_id, but unpinned -> falls back to latest and 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()

self.job_runner = SchedulerJobRunner(job=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
```

On a live deployment: trigger a run into QUEUED (e.g. via `TriggerDagRunOperator` with `wait_for_completion=False` under max_active_runs pressure), then delete its `dag_version` row; the run stays QUEUED indefinitely with the log line above.

### Operating System

Kubernetes (observed); reproduction is OS-independent.

### Deployment

Official Docker image on Kubernetes, KubernetesExecutor, git-synced bundle (`bundle_version` populated on run creation).

### Anything else

Recovery in production required manually clearing the run so a freshly pinned run could be created. SQL to confirm the state:

```sql
SELECT run_id, state, bundle_version, created_dag_version_id
FROM dag_run
WHERE state = 'queued' AND bundle_version IS NOT NULL AND created_dag_version_id IS NULL;
```

### Willing to submit PR?

- [x] Yes I am willing to submit a PR!

### Code of Conduct

- [x] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)

Contributor guide

Open the contributing guide

Research direction

Start in airflow/jobs/scheduler_job_runner.py at _start_queued_dagruns and compare its unresolved-DAG handling with the sibling scheduled task-instance path; then read airflow/models/dagbag.py, especially _version_from_dag_run and get_dag_for_run. Run the two reproduction tests provided in the issue and add regression coverage for the maintainer-selected outcome, showing that an unresolvable pinned queued run no longer remains queued indefinitely.

Written by the indexing model from the issue text.

Assessment

Tech stack
python
Domain
backend
Issue type
Bug
Difficulty
3/5
Estimated time
1-2 days
Activity status
Quiet
Clarity
Mostly clear
Newbie friendliness
68/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.