ChickenBenny opened a new issue, #76190:
URL: https://github.com/apache/airflow/issues/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:
```python
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:
```python
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.
### Operating System
_No response_
### Deployment
None
### Apache Airflow Provider(s)
_No response_
### Versions of Apache Airflow Providers
_No response_
### Official Helm Chart version
Not Applicable
### Kubernetes Version
_No response_
### Helm Chart configuration
_No response_
### Docker Image customizations
_No response_
### 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?
- [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)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
