fail_fast Dag with a task removed mid-run puts the scheduler into a crash loop #76190

Under which category would you file this issue?

Airflow Core

Apache Airflow version

main (d561f37ea5, 2026-10-29); the same code path is present in 3.3.2

What happened and how to reproduce it?

In a Dag with fail_fast=True, if a task is removed from the Dag (deleted or renamed) while a run is in progress, the next task in that run that finally fails triggers a KeyError in _stop_remaining_tasks. This happens twice:

1. Execution API – the worker's PATCH /execution/task-instances/{id}/state (FAILED) returns 500. The catch-all in ti_update_state calls _handle_fail_fast_for_dag again and hits the same KeyError, so the TI is never marked FAILED.
2. Scheduler – once the worker gives up, the executor reports the task as failed, the scheduler sees a state mismatch and calls ti.handle_failure(), which runs fail_fast again and raises the same KeyError. Nothing catches it, so _run_scheduler_loop exits and the scheduler process dies.

When the scheduler is restarted (k8s/systemd/supervisor), it resets the still-RUNNING TI as orphaned and schedules a new try. The task fails again and the cycle repeats indefinitely.

Where it breaks

_stop_remaining_tasks (airflow-core/src/airflow/models/taskinstance.py) iterates over every TI of the run in the DB and looks each one up in the Dag:

for ti in tis:
    if ti.task_id == task_instance.task_id or ti.state in (SUCCESS, FAILED):
        continue
    if task_teardown_map:
        teardown = task_teardown_map[ti.task_id]                          # Execution API path
    else:
        teardown = task_instance.task.dag.task_dict[ti.task_id].is_teardown  # scheduler path

For an unversioned bundle (e.g. the default local Dags folder), DBDagBag._version_from_dag_run resolves the run to the latest Dag version. If a task was removed or renamed while the run was active, it is not in that version, but its TI is still in the run. That TI can be REMOVED (the scheduler marks it in _filter_tis_and_exclude_removed) or still in any unfinished state if the scheduler has not got to it yet. Either way it passes the state check above, so the lookup raises KeyError. The trigger is the task missing from the resolved Dag version, not the REMOVED state itself.

A realistic way to hit this: a long-running fail_fast Dag is in progress when CI/CD deploys a new version of the Dag file that renames or removes one of its tasks. Any task in that run that later fails for good brings the scheduler down.

Steps to reproduce (breeze, --backend postgres --executor LocalExecutor; any deployment with an unversioned bundle should behave the same)

Put this Dag in the Dags folder. To make the repro deterministic, the Dag simulates the deployment from inside the run: boom creates a flag file that drops victim from the next parse, waits for that re-parse, then fails:

import pathlib
import time

from airflow.sdk import DAG, task

FLAG = pathlib.Path(__file__).with_name("remove_victim")

with DAG(dag_id="fail_fast_removed_task", schedule=None, fail_fast=True, catchup=False):

    @task(retries=0)
    def boom():
        FLAG.touch()
        time.sleep(25)  # longer than [dag_processor] min_file_process_interval
        raise RuntimeError("boom")

    boom()

    if not FLAG.exists():

        @task
        def victim():
            time.sleep(300)

        victim()

1. Make sure remove_victim does not exist and the Dag has been parsed with both tasks (set [dag_processor] min_file_process_interval to 10 to speed things up).
2. Trigger the Dag.
3. About 30 seconds later, boom fails.

Observed

API server:

Error updating Task Instance state. Setting the task to failed.
...
KeyError: 'victim'

Worker (task log), followed by Workload execution failed.:

Starting call to 'airflow.sdk.api.client.Client._request_with_retry', this is the 1st time calling it.
...
Starting call to 'airflow.sdk.api.client.Client._request_with_retry', this is the 4th time calling it.

Scheduler:

[error] Executor LocalExecutor(parallelism=32) reported that the task instance <TaskInstance: ...boom ... [running]> finished with state failed, but the task instance's state ...
[error] Exception when executing SchedulerJob._run_scheduler_loop
Traceback (most recent call last):
  File ".../airflow/jobs/scheduler_job_runner.py", line 1884, in _execute
  File ".../airflow/jobs/scheduler_job_runner.py", line 2103, in _run_scheduler_loop
  File ".../airflow/jobs/scheduler_job_runner.py", line 1470, in _process_executor_events
  File ".../airflow/jobs/scheduler_job_runner.py", line 1829, in process_executor_events
    ti.handle_failure(error=msg, session=session)
  File ".../airflow/models/taskinstance.py", line 2809, in handle_failure
  File ".../airflow/models/taskinstance.py", line 2757, in fetch_handle_failure_context
    _stop_remaining_tasks(task_instance=ti, session=session)
  File ".../airflow/models/taskinstance.py", line 221, in _stop_remaining_tasks
    teardown = task_instance.task.dag.task_dict[ti.task_id].is_teardown
KeyError: 'victim'
Shutting down LocalExecutor; waiting for running tasks to finish.

After the scheduler restarts:

Reset the following 1 orphaned TaskInstances:
    <TaskInstance: ...boom ... [running]>
Setting the following tasks to queued state: ... boom ... (try_number=3)

Left running for about 30 minutes, boom (retries=0, max_tries=0) reached try 13, every scheduler job from that period ended failed, and the Dag run stayed running.

What you think should happen instead?

boom should be marked FAILED once, victim should be left alone, and the Dag run should end as failed.

Suggested fix, limited to _stop_remaining_tasks:

• Skip TIs whose task is not in the Dag, using .get() instead of [...] in both lookup branches. Checking for REMOVED alone would not be enough, because the TI can still be unfinished before the scheduler marks it.
• Fetch only unfinished TIs: dag_run.get_task_instances(state=State.unfinished, session=session). Today the loop only skips SUCCESS and FAILED, so other finished states (SKIPPED, UPSTREAM_FAILED, REMOVED) are rewritten to SKIPPED. State.unfinished includes None, and fetch_task_instances turns that into state IS NULL, so TIs that are not scheduled yet are still skipped by fail_fast. A plain state NOT IN (<finished>) filter would drop them, because NULL NOT IN (...) is not true.

This matches what marking a Dag run as failed already does in airflow/api/common/mark_tasks.py (task_id.in_(task_ids) together with or_(state.is_(None), state.not_in(State.finished))).

Skipping a missing task does not leave a stray process behind. Once the scheduler marks that TI REMOVED, its next heartbeat gets a 409 and the supervisor terminates it.

The Execution API path and both scheduler paths all go through _stop_remaining_tasks, so this one change removes the 500, the scheduler crash and the restart loop.

Worth considering separately (not needed to fix this bug):

• Whether a failure in handle_failure for one TI should be able to take down the whole scheduler loop. Neither process_executor_events nor the heartbeat-timeout purge guards the call, and the purge timer is not non_fatal.
• The catch-all in ti_update_state calls _handle_fail_fast_for_dag again, so any exception from fail_fast handling becomes a permanent 500 instead of the intended "set the task to failed" fallback.

Impact

• The scheduler crashes repeatedly, which stops scheduling for every Dag in the deployment, not just the affected one.
• retries is ignored. The failing task is re-run on every scheduler restart, so tasks with side effects run more times than configured.
• It does not stop by itself. It needs manual intervention, such as failing the run or restoring the task.
• fail_fast does not do its job. Other tasks are not stopped and the run never finishes.
• It is hard to diagnose. The UI only shows a task that keeps retrying; the KeyError appears only in the API server and scheduler logs.

Conditions: fail_fast=True, an unversioned bundle, a task removed or renamed while a run is active (the removed task does not need to be running), and a later final failure in that run.

Changing a Dag's structure mid-run may reasonably give imperfect results for that run, but it should not crash the scheduler, bypass retries, or leave the run stuck forever.

Operating System

Debian (breeze CI image), host WSL2

Deployment

Other Docker-based deployment

Deployment details

breeze start-airflow --backend postgres --executor LocalExecutor on main.

Anything else?

• The Execution API part (task_teardown_map[ti.task_id] and the catch-all re-calling _handle_fail_fast_for_dag) is the same in 3.3.2. The scheduler part (task_dict[ti.task_id], ti.handle_failure in process_executor_events with no surrounding handler, and the latest-version fallback in _version_from_dag_run) is also the same in 3.3.2. This is from reading the 3.3.2 source; I reproduced it only on main.
• On the API side, every supervisor path that reports a final FAILED goes through the same handler: the task raising, the task process dying without a message (state inferred from the exit code), or the task runtime failing to start.
• The scheduler reaches _stop_remaining_tasks from two places, both via ti.handle_failure. One is process_executor_events, reproduced above. The other is the heartbeat-timeout purge (_find_and_purge_task_instances_without_heartbeats → _purge_task_instances_without_heartbeats), which loads the task the same way so that fail_fast applies, and runs in a timer registered without non_fatal=True. A worker that disappears without reporting (node loss, network partition) would therefore likely crash the scheduler the same way, without going through the Execution API first. Not reproduced yet.
• Versioned bundles should not be affected because the run stays on its own Dag version, which still has the task. Not verified.

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct