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

Summary

When infrastructure kills a task, Airflow cannot tell that apart from the task’s own code failing. A pod evicted during a node drain and a Python exception both arrive as a plain failure, because a worker that is killed outright never gets to raise anything.

Two costs follow. The eviction spends one of the retries the Dag author set aside for bugs in their own code, so with retries=2 three node drains can exhaust a task before it ever runs to completion. It also pages on-call as though the task broke, during what was routine cluster maintenance.

AIP-97 adds two things:

• Airflow classifies why a task failed as infra, application, timeout, or manual, with an optional reason token such as PreemptionByScheduler. Only a confirmed backend signal counts as infra; anything ambiguous stays unclassified and behaves exactly as it does now.
• Confirmed infrastructure failures draw on a separate infra_retries budget instead of the retries the Dag author set aside for their own code.

The same classification also reaches listeners, callbacks, and metrics, which is useful but secondary.

infra_retries defaults to 0, so an upgrade changes no retry behavior until an operator opts in.

The reference implementation is apache/airflow#66405.

What changes

For Dag authors

Take a task with retries=2 that is hit by three node drains in a row. It never gets to run its own code to completion:

Today that Dag is red because of cluster maintenance, and the two retries the author reserved for their own bugs were spent on something else. Opting in is one keyword:

PythonOperator(task_id="load", retries=2, infra_retries=3, ...)

Existing Dags need no change and keep their current retry behavior.

A task callback can also read the cause, so an owner is not paged for a drain:

def alert_owner(context):
    if context["failure_kind"] == "infra":
        return
    notify_owner(context["task_instance"], context["exception"])

For platform teams

The classification is what makes the budget safe to grant. An executor may report infra only from a documented backend signal, such as a Kubernetes DisruptionTarget condition or an ECS SpotInterruption stop code. A lost worker or a missed heartbeat is not evidence of a cause, so it stays unclassified and cannot spend the budget.

A deployment sets the fleet-wide default with [core] default_task_infra_retries, which ships as 0.

A listener receives the same cause, so disruption can be recorded for capacity planning rather than paged:

@hookimpl
def on_task_instance_failed(self, previous_state, task_instance, error, failure_kind):
    if failure_kind == "infra":
        record_infra_disruption(task_instance)
        return
    alert_on_call(task_instance, error)

The cause is delivered when the failure is handled rather than stored on the task-instance row. Consumers that need a durable record write it to a lifecycle event, metric, or state store.

Motivation

What a disruption looks like today

Current logs show that the worker disappeared, but not why. When the worker has time to shut down, Airflow may record:

{local_task_job_runner.py} INFO - Task exited with return code Negsignal.SIGKILL

If the node disappears, there may be no final worker log. The scheduler notices later that the task stopped heartbeating:

{scheduler_job_runner.py} ERROR - Detected a task instance without a heartbeat

Neither message distinguishes a task failure from an infrastructure disruption. Both paths currently consume a retry.

What Airflow can see

When the worker survives, it can report its own exception. When it does not, the cause must come from the component that observed it, such as the API or the executor backend:

The four kinds have distinct sources:

• infra: a documented backend signal says that something outside the task ended it.
• application: the task raised an exception, exited non-zero, or exceeded an application-owned limit.
• timeout: the task exceeded execution_timeout.
• manual: someone deliberately marked the task or Dag run failed.

Only infra can use the additional infrastructure retry budget. A reason without a reliable classification remains unclassified and follows Airflow’s current retry behavior.

Existing Airflow behavior

If a pod dies before its task starts, the Kubernetes executor puts it back on the queue without spending a retry. Airflow therefore already avoids charging users for a confirmed pre-start infrastructure failure. AIP-97 applies the same accounting after execution begins, but only when the backend can identify the cause.

Other schedulers make a similar distinction. Kubernetes Job podFailurePolicy can ignore DisruptionTarget. Nomad has separate reschedule and restart counters. Spark records whether an executor loss was exitCausedByApp. Flyte separates SYSTEM and USER failures and gives system failures their own budget.

Why existing retry mechanisms do not cover this case

AIP-105 evaluates a retry policy in the worker process. If a disruption kills that process, the policy never runs. AIP-103 can resume a task correctly, but reaching the resume still consumes one of the user’s retries.

The boundary is whether the worker reaches its Python error path. An infrastructure problem that raises an exception in a live worker, such as a database timeout or DNS error, belongs to AIP-105. AIP-97 covers failures that kill the worker before it can report an exception. The AIP-105 discussion reached the same conclusion. The two AIPs cover different failure domains.

Existing production use

At least one large Airflow 2 deployment already grants replacement attempts when its executor detects a pod disruption. That implementation adjusts retry accounting from the task process. Airflow 3 removes the task-side database session it depends on, so the deployment cannot carry the behavior forward without support in the scheduler.

The design

Every change lands in a process Airflow already runs. Executors run inside the scheduler, so classification and the budget decision sit next to the database the scheduler already owns, and the worker keeps its existing boundary.

The diagram links to the current implementations of the worker kinds, manual kind, executor contract, scheduler handoff, listener, callbacks, metrics, and logs.

Classify the cause

AIP-97 adds an optional failure_kind argument to on_task_instance_failed. Its value is one of the four strings defined above:

def on_task_instance_failed(previous_state, task_instance, error, failure_kind): ...

pluggy matches hook arguments by name. A listener that does not declare failure_kind continues to be called as before.

The producer can also attach a short reason token such as Evicted or SpotInterruption. The component that observed the failure supplies the classification. The worker sets application or timeout for failures it catches. The API sets manual when a user marks a task failed. An executor sets infra only when its backend reports a specific disruption.

The kind and reason are delivered when the failure is handled rather than stored on the task-instance row. This part of the proposal needs no schema migration and can be backported independently of the infrastructure retry budget.

The same classification reaches each failure consumer. The task’s on_failure_callback and on_retry_callback receive it through context. ti_failures and operator_failures carry it as a bounded metric tag, with live backend evidence that the added tag does not create a cardinality problem. The scheduler logs each infrastructure retry together with the reason that justified it. Dag callbacks, fleet-level dashboards, and scheduler logs therefore describe the same failure in the same terms.

Adding the kind to the ordinary task-finished log, so that it is visible even when infra_retries=0, is not yet implemented. It is listed under what defines this AIP as done.

Give infrastructure failures their own retry budget

infra_retries is the number of additional attempts available for confirmed infrastructure failures. It defaults from [core] default_task_infra_retries, whose default value is 0. A Dag author can override it on an individual task:

PythonOperator(
    task_id="load",
    retries=2,
    infra_retries=3,
    ...
)

With this configuration, three confirmed infrastructure failures can be retried without using the two attempts configured through retries. After the infrastructure budget is exhausted, Airflow follows its existing retry behavior.

Airflow implements the additional budget by storing infra_retries_used under a reserved AIP-103 task-state key. When an infra failure occurs and budget remains, the scheduler increments that counter and max_tries in the same transaction. Retry eligibility already uses max_tries, including for retries=0. The separate counter survives scheduler restarts and task clears, and [state_store] clear_on_success can remove it after the task succeeds.

The task state store is durable storage. Its default metastore backend writes to the task_state_store table added by AIP-103, but AIP-97 does not add a column or migration. The infra_retries setting itself travels with the serialized task. Airflow must reserve the counter key so task code cannot overwrite scheduler accounting.

Airflow checks the infrastructure budget before it chooses retry callbacks and email. This keeps the notification type and the persisted task-instance state consistent, including when a task with retries=0 receives its first infrastructure retry.

The sequence traces one evicted pod through the typed producer, the budget gate, and AIP-103 task state. The highlighted region is where the ordering above is enforced.

If the community does not want a new task-level budget, a smaller fallback stays available (POC). A deployment-wide [core] max_infra_retries cap increments the existing max_tries field after a confirmed infrastructure failure, and it adds no state-store key and no schema. That version can be backported further, but it cannot expose a reliable per-task infrastructure budget, because a task clear changes the values used to infer how many replacement attempts were granted. The inference is deliberately conservative, so an exhausted cap cannot reopen.

The executor contract

The shared executor contract makes the proposal useful beyond Kubernetes.

Today BaseExecutor.change_state(key, state, info=None) types info as Any, so there is no failure-cause contract for an executor author to implement. AIP-97 adds optional typed arguments to the failure path that every executor already calls:

def fail(self, key, info=None, *,
         failure_kind: TaskFailureKind | None = None,
         reason: str | None = None) -> None: ...

reason is a short, stable token owned by the backend rather than a free-form error message. It does not imply a classification. For example, WorkerLost and SIGKILL are useful diagnostic details, but neither identifies who killed the process. The scheduler can pass those reasons to listeners while leaving failure_kind unset. Only failure_kind=infra can use infra_retries.

The existing info parameter remains unchanged because it already carries unrelated state-dependent values such as external_executor_id. Typing it would break out-of-tree executors.

The worker and API supply application, timeout, and manual regardless of executor. Executor-specific infra classification remains limited to stable backend evidence:

The first implementation recognizes this bounded list:

Generic pod deletion, WorkerLostError, SIGKILL, OOMKilled, and heartbeat loss do not appear in this list because none identifies an external infrastructure cause by itself.

Separating a disruption from a manual stop

A termination signal alone cannot distinguish a disruption from a manual stop. The API records a requested terminal state before it terminates the process. If the executor later reports the process exit, the task instance is already terminal and Airflow does not reclassify it. An executor may report infra only while the task is still running and the backend has supplied a specific disruption cause.

Open decision: how infrastructure retries are budgeted

Failure classification is useful by itself, but it does not solve the retry-accounting problem that motivated this proposal. The remaining choice is how Airflow should represent the additional attempts granted for infrastructure failures.

The first-class option is preferred for four reasons:

• infra_retries makes the additional budget visible in the Dag instead of hiding it in scheduler accounting.
• A default of 0 preserves current behavior, while deployments and individual tasks can choose a different budget.
• The AIP-103 counter survives scheduler restarts and task clears.
• Every executor uses the same budget once it can report a confirmed infra cause.

This does not add an INFRA_FAILED task state. infra is a failure cause, and infra_retries is an additional budget used while the task moves through the existing UP_FOR_RETRY and FAILED states.

The implementation also follows two classification rules. It does not infer a cause from liveness loss, and it does not require every executor to provide one. Kubernetes disruption reasons and ECS SpotInterruption currently meet the standard for a stable terminal cause. Other signals remain unclassified unless their backend provides an equally specific contract.

Impact

Dag authors do not need to change their Dags. Listener authors can accept the new optional arguments. Operators can set a default infrastructure budget, and Dag authors can override it per task. Executors that do not implement the cause contract keep their current behavior.

Neither option needs a new schema migration. The preferred option uses the AIP-103 task state store and therefore requires Airflow 3.3 or later. The fallback changes only the existing max_tries value and can be backported independently. Under the preferred option, the reserved counter key is writable only by Airflow, not by task code.

Safety boundaries

An executor may set failure_kind=infra only from a documented backend cause. Ambiguous signals can still be reported as reasons, but they cannot use infra_retries. This keeps worker loss and missed heartbeats from silently changing retry behavior.

The infrastructure budget is 0 by default and finite when enabled. It can run a task more times than retries alone would allow, so operators should use it for tasks that already tolerate retry.

Future work

Some platforms announce a disruption before they terminate a process. Kubernetes can set DisruptionTarget before deleting a pod, and the task SDK already receives SIGTERM. A later change could preserve that notice and correlate it with the terminal result. AIP-97 still needs the post-failure path because a hard kill may provide no notice, and a notice alone does not show whether the task stopped.

Out of scope

What defines this AIP as done

1. The worker and API report application, timeout, and manual through the shared listener contract regardless of executor.
2. BaseExecutor.fail() accepts the optional typed failure_kind and reason fields, with documentation for in-tree and out-of-tree executor authors.
3. Kubernetes and ECS report independent, backend-confirmed infra causes.
4. Celery reports WorkerLost as a reason with no kind, confirming that ambiguous evidence remains observable without using the infrastructure retry budget.
5. infra_retries defaults from deployment configuration and can be overridden per task. Its durable counter is used only for infra. Retry callbacks and email are selected after the budget check, so they match the state stored in the database, including at retries=0.
6. failure_kind reaches the other consumers. The Dag author’s callbacks and the ti_failures and operator_failures metric tags already carry it in #66405. The remaining piece is the ordinary task-finished scheduler log, which today records the kind only when an infrastructure retry is granted.
7. The configuration, the listener contract, and the executor contract are documented for Dag authors, deployment managers, and executor authors.

These changes can land as separate, reviewable PRs. The current POCs cover the listener, Kubernetes, and budget path (#66405), ECS (#28), Celery (#15), and the stateless fallback (#16). All four are rebased on the current contract and repeat their end-to-end checks against it.

Verification

The evidence below records the paths exercised while developing the proposal. Every retained PR has been rebased onto the current contract and repeats its own checks. Full commands and raw output belong in those PRs.