Fail stranded Dag runs instead of holding a max_active_runs slot - #71642

Deleting a task instance leaves its Dag run unable to reach a terminal state. The scheduler's deadlock check would otherwise fail such a run and release its max_active_runs slot, but that check is gated on should_schedule, where every clause is an all(...) — so a single unfinished task carrying max_active_tis_per_dag or max_active_tis_per_dagrun suppresses it for the entire run. On a Dag with max_active_runs=1, scheduling stops completely. The reporter observed four stranded runs in a single day, one of which blocked roughly ten scheduled runs over four hours.

Why removing the concurrency clauses is safe

Task-level concurrency limits are never evaluated as dependencies on this path:

• DepContext.deps defaults to an empty set, and both _get_ready_tis and _are_premature_tis build a bare DepContext(...).
• DEFAULT_OPERATOR_DEPS does not contain TaskConcurrencyDep. That dep appears only in REQUEUEABLE_DEPS, RUNNING_DEPS and SCHEDULER_QUEUED_DEPS, which are scheduler and executor paths.

A throttled task therefore still reports its dependencies as met and counts as runnable, so it could not have tripped the deadlock check in the first place. Gating on these limits protected nothing while disabling detection for the whole run.

PoolSlotsAvailableDep corroborates this. It lives in the same dep sets and produces the same "parked in SCHEDULED" shape, yet should_schedule never gated on pool starvation — and that has never produced the false positive the concurrency gate was presumably guarding against.

depends_on_past and DEFERRED/AWAITING_INPUT are deliberately retained. Those are evaluated as dependencies (PrevDagrunDep is in DEFAULT_OPERATOR_DEPS), so such a task can legitimately become runnable later and must not be treated as deadlocked.

Scope

This closes one of the three gates described in the issue. Explicitly not addressed here:

• A run still strands if any unfinished task carries depends_on_past or is deferred.
• The deleted task instance is still never re-created; both verify_integrity paths remain gated (scheduler_job_runner.py requires a Dag version change and is skipped entirely for bundle-versioned runs; taskinstance.py is unreachable while the run is non-terminal).

Failing the run does at least make the second reachable: once the run is terminal, clearing it can restore the missing row.

Upgrade impact

Dag runs already stranded in the metadata database will terminate as failed the next time the scheduler evaluates them. Expect a one-off burst of failure callbacks and on_dag_run_failed listener events on the first scheduler pass after upgrade.

Testing

airflow-core/tests/unit/models/test_dagrun.py: 233 passed, 1 pre-existing xfail.

• test_dagrun_deadlock_with_deleted_ti_and_concurrency_limit — parametrized over both removed attributes, reproducing the issue with an actual session.delete(ti), which is what DELETE /api/v2/dags/{dag_id}/dagRuns/{run_id}/taskInstances/{task_id} does. Fails on main.
• test_dagrun_no_deadlock_with_throttled_task — pins the safety argument above; it fails if TaskConcurrencyDep is ever added to the default dep set.
• test_dagrun_no_deadlock_with_deleted_ti_and_depends_on_past and test_dagrun_no_deadlock_with_deleted_ti_and_deferred_task — cover the two retained clauses. Neither had coverage before: removing either clause previously passed the entire file.

closes: #71540

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

Generated-by: Claude Code (Opus 5) following the guidelines

• Read the Pull Request Guidelines for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
• For fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
• When adding dependency, check compliance with the ASF 3rd Party License Policy.
• For significant user-facing changes create newsfragment: {pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.