AIP-97 Disruption Readiness (Transparent Infra Failure Retries through Context Capture and Propagation)

Summary

When you give a task retries=2, you’re telling Airflow "if my code fails, give it two tries." Maybe it calls an API that sometimes fails. Those retries are meant for your code.

But Airflow spends them on things your code had nothing to do with. Say the worker pod running the task is evicted partway through, because the cluster needs its node back. Your code didn’t fail, but Airflow can’t tell that apart from a crash. It sees the task stop, records a failure, and uses up one of your two retries. If it happens again, both are gone and the task fails, even though the code never ran to completion. The same gap shows up in alerting: on-call gets paged for what looks like a task failure when a node was only drained for maintenance.

The reason Airflow can’t tell the difference is that when something kills a task from outside, the worker dies before it can explain itself. The only thing that reaches a failure listener is a generic error. (A listener is a plugin Airflow calls whenever a task fails; it’s the hook that alerting, lineage, and metrics tools plug into.) Nothing downstream can separate "the node went away" from "the code threw an exception."

Airflow already handles a small version of this. If a pod dies before its task even starts (the image won’t pull, say), the Kubernetes executor quietly puts it back on the queue without spending a retry. AIP-97 carries that same idea across to a task that’s already running.

Two small additions, both opt-in:

• Failure context. When a task fails, tell the listener why: infrastructure, the task’s own code, a timeout, or someone stopping it by hand. The tools that hang off those hooks (alerting, metrics, lineage, lifecycle tracking) can then treat a node drain differently from a real bug.
• Transparent infra failure retry. When infrastructure was the cause, don’t charge the attempt to the user’s retries.

The failure kind is a new argument on the listener that any executor can set. The KubernetesExecutor, CeleryExecutor, and local executor are each wired up in a POC, and nothing in the design is tied to any one of them. A listener that doesn’t declare the argument keeps working as before, and an executor that can’t classify a failure leaves it unset, so behavior there is unchanged.

POC: apache/airflow#66405 (all of it on one branch), landing as the three smaller PRs in What defines this AIP as done?.

One color key across the diagrams: blue is the infrastructure path, amber the user’s own code, indigo the failure_kind and reason signal, slate the neutral steps, and red versus green is before versus after.

Where AIP-97 plugs in

The change lives in three seams: the executor bridge sets failure_kind from what its backend reports, the scheduler refunds an infra attempt by bumping the max_tries it already tracks and hands the kind to the listener, and the listener spec gains one argument. The metadata schema doesn’t change.

AIP-105 already drew this line on its discussion thread. Its retry policies run inside the worker, right after the code throws, so they only come into play when the worker is alive to catch the exception. Once the worker itself is killed, that code never runs. AIP-97 covers the other side of that line: the failures where the worker is already gone, and something outside it has to work out what happened.

Motivation

What a disruption looks like today

When infrastructure takes down a running task, one of two things lands in the logs. If the worker gets a moment to wind down, the task log trails off with something like:

Server indicated the task shouldn't be running anymore. Terminating process
Task killed!

If the whole node goes away and takes the process with it, there is no final log line at all. The scheduler only finds out later, when the task stops sending heartbeats, and fails it:

Task did not emit heartbeat within time limit (N seconds) and will be terminated.

Read either one and you can’t say whose fault it was. Nothing in there says "the platform did this." To alerting, to lineage, and to the retry counter, it reads the same as a task that raised ValueError on line 40. So the task’s on_failure_callback fires, an alert goes out to on-call as though the code had broken, and a retry is spent, over a disruption the author had no hand in.

What we want instead

We want the disruption seen for what it is: infrastructure, not the user’s code. Two people read a task failure, and today neither can tell a drain from a bug.

The platform team runs a cluster-wide listener. Hand it the reason and it can record the disruption for capacity tracking instead of alerting someone about code that didn’t break:

@hookimpl
def on_task_instance_failed(self, previous_state, task_instance, error, failure_kind):
    if failure_kind == "infra":
        # a drain or an eviction: record it for capacity tracking, skip the alert
        record_infra_disruption(task_instance)
    else:
        alert_on_call(task_instance, error)

The DAG author writes their own on_failure_callback. It runs in the DAG processor after the worker is already gone, so it reads the same reason from the task context:

def alert_owner(context):
    if context["failure_kind"] == "infra":
        return  # a drain or an eviction, not this DAG's code; don't alert
    notify_owner(context["task_instance"], context["exception"])

PythonOperator(task_id="load", on_failure_callback=alert_owner, ...)

And it means giving back the retry the disruption ate, so retries=2 is two tries at the user’s code, not one try and one eviction.

The signal exists; Airflow just drops it

Start with the hook itself, on_task_instance_failed(previous_state, task_instance, error). All it learns about the failure is error, which is a None, a string, or an exception. That’s plenty when the worker caught the failure and can pass the exception along. For the failures the worker doesn’t survive, there’s nothing meaningful to put there:

Kubernetes makes it especially plain. The executor already looks at the dead pod and captures the pod status, the container’s reason, and the exit code in a FailureDetails record. That is everything you’d need to know it was an eviction, and none of it reaches the listener. By the time the listener runs, it’s been flattened down to a string.

The four kinds

A failure comes back as one of four kinds:

• infra: something outside the task ended it. A drained node, an evicted or preempted pod, a reclaimed spot instance, a pod deleted while its task was still running. This is the one that earns a free retry.
• application: the task’s own code is at fault. It raised, exited non-zero, or used more memory than its limit allowed and got OOM-killed for it.
• timeout: the task ran past its execution_timeout. (A DAG-level dagrun_timeout behaves differently: it skips whatever is still running rather than failing it, so it never lands here.)
• manual: a person marked it failed, either the task or the whole DAG run.

Why keep application and manual apart instead of folding both into one "user" bucket? Because they answer different questions. "My code crashed" and "someone stopped this on purpose" aren’t the same event to whoever’s reading the alert, and Temporal draws the same line between an application failure and a cancellation. The refund only ever looks at infra; the other three stay on Airflow’s existing retry path, so a real bug is never mistaken for infrastructure and handed a free attempt.

Airflow already does this, for one case

When a pod fails while its task is still queued and hasn’t started (the image won’t pull, for instance), the Kubernetes executor puts it back on the queue and doesn’t charge a retry. It stops at that line because once the task is actually running, Airflow can’t tell an eviction from a crash without knowing the cause. That is the gap the failure kind fills, and the refund extends the same rule to a task that is already running.

Considerations

Alongside AIP-105

AIP-105 already handles the failures the worker catches: a Python exception, run through a retry policy in the worker process. AIP-97 takes the other half: the failures where the worker is already dead and something outside has to classify them. They never touch the same code. The failure kind is an argument on a listener, not an input to a retry policy, and you can adopt either one without the other.

Works with AIP-103

AIP-103 and AIP-97 work together. AIP-103 lets a task carry its progress across attempts, so the next run resumes instead of starting over. AIP-97 makes sure that when infrastructure interrupted it, the retry does not come out of the user’s budget and the reason is recorded. Together, a spot reclamation costs the user almost nothing: the work resumes, and the retries are untouched. A consumer that wants the failure kind to outlive the attempt can store it in AIP-103’s task_state store.

Telling an eviction from a manual stop

An eviction and an operator marking a task failed can both reach a running task as a termination signal, so the signal looks the same for both. The difference is the task’s state. When a person stops a task, Airflow writes down the outcome they chose before the task is torn down, so by the time the executor reports in, the instance is already terminal and there’s no way to mistake it for infrastructure. Only a task that was still running when something external killed it gets called infra. We ran each of these against a live cluster (a manual mark, an execution_timeout, a DAG-run stop), and none is refunded; only a genuine external kill is.

Out of scope

What change do you propose to make?

Pillar 1: carry the reason to the listener

It’s one new argument, failure_kind, on the on_task_instance_failed hook. Its value is a short string enum (infra, application, timeout, or manual), so a listener compares it with a plain == "infra". It’s optional: pluggy matches hook arguments by name, so a listener that never mentions it is called just like before. Where the failure is seen decides who fills it in: the worker sets application or timeout for the failures it catches, and the scheduler sets infra for a task killed from outside. Next to it rides the executor’s short reason token (Evicted, PodDeleted, and the like), handed to the listener at the moment of failure rather than stored on the row, so a consumer that wants it to outlive the failure records it itself (a lifecycle event, a metric tag, its own store). It’s the same move #56272 made for DAG-run notifications, a transient msg on the hook. How that token best reaches the listener is an open question below.

The same failure_kind reaches every consumer of a failure. Beyond the platform listener, the DAG author’s own on_failure_callback and on_retry_callback get it through the task context (POC), so a per-DAG alert can skip a drain the way the listener does. The failure metrics ti_failures and operator_failures carry it as a tag (POC, live e2e), so a dashboard can tell infra churn from real bugs instead of counting both as one failure spike. And the scheduler writes it into the TaskInstance Finished log on every infra failure (POC), so an operator reading the logs can tell an eviction from a crash even when the refund is off. Without these, Pillar 1 tells the platform team the reason but leaves the DAG author’s alerts, the failure dashboards, and the operator’s logs blind.

How a failure gets classified

Classification happens in two steps, and the first one doesn’t care which executor you’re on. Airflow already knows whether a task reported its own failure or just went quiet: when the code throws, the worker marks the task failed before the executor has any say; when the worker is killed, nothing gets marked, and the task still looks like it’s running when the executor reports it gone. That mismatch, where the executor says it finished but the task never said so, is the sign that something external happened, and it reads the same on every executor.

The second step is where an executor adds what only it can see. The KubernetesExecutor is one example: it reads the pod, so an eviction, a preemption, a lost node, or a pod deleted while its task ran is infra, and a container that died on its own, including one that blew past its own memory limit, is application. A lost worker under the CeleryExecutor, or a SIGKILL under the LocalExecutor, classifies the same way, through the same executor hook, without any core change. An executor that can’t tell leaves the failure unclassified, and Airflow falls back to today’s behavior rather than guessing.

Pillar 2: refund the retry infra used

When a failure comes back as infra, Airflow refunds the retry it used rather than counting it, so retries=2 still buys two real tries at the user’s code. A per-task ceiling, [core] max_infra_refunds (default 3), bounds how many times it will do this. A task stuck on a bad node that keeps evicting it refunds up to that cap, and after that an eviction spends a real retry like any other failure and the task fails normally, so it can’t retry forever. The refund is transparent to the retry budget but never silent. The scheduler logs the classification on every infra failure, whether or not it refunds, so an operator can tell an eviction from a crash even with the refund off (POC). When it refunds, the reason and running count go to the task log (infra failure (Evicted) refunded attempt ...; max_tries now 2 (refund 1/3), user retry budget preserved); when the cap is reached, it logs why it spent a real retry instead. And because failure_kind and the reason reach the listener, the callbacks, and the metrics, everything downstream can report exactly what happened. It stays off until an operator turns it on, and it needs no schema change; it reuses the retry count Airflow already keeps.

Counting the refund: max_tries is the source of truth

The refund works by lifting max_tries, and retry eligibility now reads max_tries rather than the user’s configured retries. TaskInstance.is_eligible_to_retry was the last place that still short-circuited on retries, which is why a retries=0 task couldn’t be rescued no matter how high the ceiling went. It now mirrors the execution API’s _is_eligible_to_retry, which already dropped retries in favor of max_tries ("we can handle using max_tries"). So the scheduler-detected path and the worker-reported path agree, and the refund reaches a retries=0 task the same way it reaches a retries=2 one. Behavior is unchanged everywhere except the one case the refund creates: max_tries lifted past the user’s retries.

Three other ways to cover retries=0 were considered and rejected. A dedicated infra_retry_count column, the shape Flyte and Nomad use, keeps retries pristine and reads cleanest for a UI, but it reintroduces the one migration this design avoids and adds a second retry counter to reason about; the migration-free property is what keeps AIP-97 backportable, and aligning the gate covers retries=0 without it. A bypass in handle_failure that re-checks eligibility off the refund’s return value is contained, but it leaves the scheduler and worker eligibility paths divergent, the exact split this removes, and would need a database integration test to pin. And porting the checkpoint-resume bypass from an internal 2.9.2 deployment, which force-allows a retry on resume, needs AIP-103 checkpoint state that this design deliberately doesn’t keep and predates the execution API split.

What problem does it solve?

• Everything that consumes task failures (alerting, metrics, lineage, capacity planning, lifecycle tracking) can read the kind and the reason and separate infra churn from real bugs. Large deployments already build on these lifecycle listeners to emit tracking and metrics events for every DAG and task run, and today an eviction and a code exception look identical in that data. The failure kind is the dimension that’s missing.
• An eviction, or any infra kill, stops eating the retries a user set aside for their own code.
• Anyone consuming failures gets one clear place to read where a failure came from.

Why is it needed?

Every mature scheduler that has run into this ends up keeping infra failures off the application retry budget. Kubernetes has podFailurePolicy (ignore a DisruptionTarget), Nomad keeps reschedule and restart as separate counters, Spark tracks exitCausedByApp, and Flyte sorts failures into SYSTEM and USER with their own system-retry budget. Airflow counts them all the same. It already carved out the exception for pods that die before they start; this finishes the job.

Are there any downsides?

No real ones. It adds one idea, that a failure has a kind, and the rest follows from it. A listener that doesn’t use it sees no difference. The retry change is the only shift in behavior, and it stays off until someone turns it on, with a ceiling so it can’t run away, the same thing Airflow already does for pods that die before they start. The classification is a best guess, and when Airflow can’t guess it doesn’t; the failure just falls through to today’s behavior, the way Flyte leaves an unknown cause unclassified.

Which users are affected?

• Listener authors who want the context add failure_kind to their hook. Opt-in.
• Executor authors write a small classifier for their backend. Opt-in, with the KubernetesExecutor, CeleryExecutor, and local executor each shown in a POC.
• DAG authors don’t change anything. For Pillar 1 they see nothing new; for Pillar 2, infra failures stop drawing down their retries.

How are users affected? (DB / migration)

No schema change, so no migration. Pillar 1 hands the failure kind and reason to the listener when the task fails, instead of storing them. Pillar 2 bumps max_tries, a column Airflow already has, behind an opt-in flag that defaults to today’s behavior. All of this keeps the change easy to backport, with no upgrade or rollback step.

What is the level of migration effort?

None to adopt. A listener or an executor opts in when it wants the context, and a deployment flips one flag for the retry change.

What defines this AIP as "done"?

Seven pieces, each its own PR. The consolidated POC is the core (pieces 1 to 5) on one branch; the numbered PRs below are how it’s meant to land, with the completeness and executor follow-ups POC’d on their own.

1. The foundation: the failure_kind argument on the hook, the four kinds, and the executor’s reason token, with the worker and the API filling them in. (PR 1/3.)
2. The Kubernetes classifier: reading the pod to call an eviction or a PodDeleted infra, and an own-limit OOM application, wired through to the scheduler. It also reads the DisruptionTarget condition, so a node drain or a preemption reads as infra rather than as an application crash. (PR 2/3.)
3. An end-to-end check of the foundation: a recording listener actually receives the kind from a real killed task, with the logs in the consolidated PR.
4. The retry refund: gated, capped, and only ever on infra, with unit tests pinning that scope. (PR 3/3.)
5. An end-to-end check of the refund on a real Kubernetes cluster, across a code crash, an OOM, a timeout, a manual stop, and a real eviction: only the eviction is refunded, and repeated evictions refund up to the cap and then fail. Logs in the consolidated PR.
6. The reason reaching the other consumers beyond the platform listener: the DAG author’s on_failure_callback and on_retry_callback get failure_kind through the task context (#19), the failure metrics ti_failures and operator_failures carry it as a tag (#20, live e2e), and the scheduler logs the classification on every infra failure, independent of the refund (#21). All POC’d, so a per-DAG alert, a dashboard, and an operator reading the logs can each separate infra from a bug the same way the listener can.
7. The remaining executor paths and the docs, each POC’d and landing as a follow-up: the Celery classifier (#15), the scheduler-side heartbeat path that also covers the local executor (#14), and the config plus listener docs (#18).

Open questions

1. How big should the infra budget be, and who sets it? The POC uses one server-wide cap. The alternative is a server default that a DAG or task author can raise, so a job that’s especially disruption-prone can ask for more headroom (POC).
2. How should the reason reach the listener? With the reason no longer kept in database state, the executor’s token is handed to the listener at failure time, and the shape is open. Fold it into the existing error argument, so an infra kill surfaces error="Evicted" alongside failure_kind=infra and adds no argument (POC); or give it a dedicated optional reason argument next to failure_kind, leaving error as the worker’s own error (POC). Folding into error is the smaller surface but puts a short token (for infra) and a full exception (for application) in the same field; a separate reason keeps them apart, which reads cleaner for a lineage or dashboard consumer that wants a short cause, at the cost of one optional kwarg. Either way pluggy matches by name, so a listener that ignores it is unchanged, and #56272 already set the precedent with a transient msg on the DAG-run hook. Both are POC’d so we can compare.