Scheduler exits when handle_failure raises for a single task instance #76529

Under which category would you file this issue?

Airflow Core

Apache Airflow version

main; the same code path is present in 3.3.2

What happened and how to reproduce it?

Follow-up from #76191 (fixing #76190).

The scheduler calls ti.handle_failure() for each task instance inside a loop, with nothing that isolates an exception raised for one task instance:

• SchedulerJobRunner.process_executor_events, when the executor reports a failure the TI did not record itself
• the heartbeat-timeout purge (_purge_task_instances_without_heartbeats)

If handle_failure() raises for any reason, the exception leaves _run_scheduler_loop and the scheduler process exits. After a restart the same task instance is still in the same state, so it is picked up again, raises again, and the scheduler keeps crashing. In #76190 this was a KeyError from fail_fast handling. A task with retries=0 reached try 13 in about 30 minutes, because each restart reset the TI as orphaned and ran it again.

#76191 removes that specific KeyError, but any other exception raised from handle_failure() (a failure callback lookup, fail_fast handling, a data problem in one Dag run) has the same effect. One bad task instance can stop scheduling for every Dag.

To reproduce, make handle_failure() raise for one task instance. For example, on a build without #76191, follow the steps in #76190.

What you think should happen instead?

An exception from handle_failure() for one task instance should not stop the scheduler. It should log the error with the task instance key, and leave the TI in a state that does not cause it to be picked up again in the same way, for example FAILED without the extra handling. It should then continue with the remaining task instances.

Open points:

• If the exception comes from the database, the transaction may no longer be usable. Handling each TI inside a savepoint (session.begin_nested()) may be needed so that one TI's failure does not roll back the others.
• Which fallback state to use when handle_failure() itself cannot complete.

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?

• Related: Execution API re-runs fail_fast handling after it fails during a task state update #76530
• Related: Fix fail_fast crash on removed tasks and stop overwriting finished TIs #76191

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