8. State, dependencies and retries

A task runs when its upstream dependencies satisfy its trigger rule. The default rule is so sensible that most people never learn there are others, and then meet one of the two situations where the default is wrong.

Trigger rules

@task(trigger_rule="none_failed_min_one_success")
def summarise():
    ...

Rule	Runs when
`all_success`	default — every upstream succeeded
`all_failed`	every upstream failed
`all_done`	every upstream finished, whatever the outcome
`all_skipped`	every upstream was skipped
`one_success`	at least one succeeded (does not wait for the rest)
`one_failed`	at least one failed (does not wait)
`one_done`	at least one finished, success or failure
`none_failed`	none failed — skipped is acceptable
`none_failed_min_one_success`	none failed and at least one succeeded
`none_skipped`	nothing upstream was skipped
`always`	unconditionally

Three of these earn their place:

`all_done` — cleanup that must always run

@task(trigger_rule="all_done")
def release_lock():
    ...          # runs whether the pipeline succeeded or failed

The pattern for anything that must happen regardless: releasing a lock, tearing down a cluster, sending a “pipeline finished” notification with the outcome in it.

`none_failed_min_one_success` — the join after a branch

This is the one everyone hits.

   branch ──┬─► path_a  (chosen, succeeds)
            └─► path_b  (SKIPPED)
                   │
                   ▼
                 join     ← all_success requires ALL upstreams
                            to have SUCCEEDED. path_b was
                            skipped, not successful.
                            → join is SKIPPED. Silently.

The pipeline stops after the branch and nothing looks wrong — no failure, no error, just a run that ends early with a lot of grey boxes. none_failed_min_one_success is the correct rule for a join after a branch, and forgetting it is the single most common Airflow logic bug.

`one_success` — first one wins

Try three sources; proceed as soon as any works. Note that it does not wait for the others, so they keep running while downstream proceeds.

Retries

@task(
    retries=3,
    retry_delay=timedelta(minutes=5),
    retry_exponential_backoff=True,
    max_retry_delay=timedelta(hours=1),
)
def call_flaky_api():
    ...

With exponential backoff: 5 min, 10, 20, 40, capped at max_retry_delay.

Guidance that actually holds:

Retry things that fail transiently. Network calls, rate limits, temporary unavailability, contention.

Do not retry things that fail deterministically. A KeyError in your code will KeyError three more times, twenty minutes apart, and then fail — turning a two-minute failure notification into an hour-long one.

Use exponential backoff for anything external. Retrying a struggling API every five minutes is how you keep it struggling.

Set execution_timeout on everything (chapter 3). A hung task holds a worker slot forever, and retries does not help because it never fails.

To fail immediately on a known-permanent error:

from airflow.exceptions import AirflowFailException

@task(retries=5)
def load(record):
    if not valid(record):
        raise AirflowFailException("bad input — retrying will not help")

AirflowFailException skips remaining retries. Its counterpart AirflowSkipException marks the task skipped instead — the right way to say “there was nothing to do today”, which is different from success and very different from failure.

Idempotency

Not a style preference. A correctness requirement, because Airflow will run your task more than once whether you plan for it or not:

   - a retry after a transient failure
   - a zombie task killed and rerun (chapter 5)
   - a manual clear and rerun
   - a backfill over a period already processed
   - two schedulers racing (rare, but the system
     is designed assuming at-least-once execution)

# ✗ NOT IDEMPOTENT — running twice doubles the data
@task
def load(**context):
    rows = extract(context["data_interval_start"])
    warehouse.execute("INSERT INTO daily_sales VALUES ...", rows)

# ✓ IDEMPOTENT — running twice produces the same result
@task
def load(**context):
    start, end = context["data_interval_start"], context["data_interval_end"]
    rows = extract(start)
    warehouse.execute(
        "DELETE FROM daily_sales WHERE day = %s", (start.date(),))
    warehouse.execute("INSERT INTO daily_sales VALUES ...", rows)

The patterns that work:

Delete-then-insert for the interval. Simple, obvious, correct.

Partition overwrite. Write to s3://bucket/table/date=2026-08-29/, replacing whatever was there. This is why partitioned storage and Airflow fit together so well.

MERGE / upsert on a key. INSERT ... ON CONFLICT DO UPDATE in PostgreSQL; MERGE in most warehouses.

Write to a staging location, then swap atomically. The safest for large outputs — a failed run leaves the previous data intact.

The anti-patterns: appending without a key, incrementing a counter, sending an email (genuinely hard to make idempotent — use a deduplication key, or accept it), and anything that reads “the current state” and adds to it.

If a task is not idempotent, every retry is a data bug waiting to happen. And you cannot turn retries off, because zombie detection will rerun it anyway.

Clearing and rerunning

airflow tasks clear daily_sales \
  --start-date 2026-08-01 --end-date 2026-08-30 \
  --task-regex "transform.*" --downstream

Clearing removes a task’s state so the scheduler reruns it. --downstream clears everything after it too, which is almost always what you want — if transform produced different output, load needs to run again.

In the UI: select tasks, Clear. The dialog’s options map to the same flags.

Clearing is the standard recovery operation and it is why idempotency matters so much. You will clear tasks. You will clear them over date ranges. It has to be safe.

Dependencies

extract >> transform >> load                    # chain
extract >> [transform_a, transform_b] >> load   # fan out, fan in

from airflow.sdk import chain, cross_downstream
chain(a, [b, c], d)                             # a → (b,c) → d
cross_downstream([a, b], [c, d])                # every left → every right

In TaskFlow, dependencies come from data flow:

load(transform(extract()))                      # dependencies inferred

Mixing both, when a task depends on another without consuming its output:

data = extract()
notify = send_notification()
data >> notify                                  # order without data flow

Cross-run dependencies

@dag(...)
def incremental():
    @task(depends_on_past=True)
    def apply_delta():
        ...

depends_on_past=True means this task instance will not run until the same task in the previous run succeeded. For genuinely sequential state — a running balance, an incremental merge — this is necessary.

It is also a foot-gun with a sharp edge: one failure blocks every subsequent run, forever, until someone intervenes. A DAG with depends_on_past that fails on a Friday has done nothing by Monday. And a catchup with it set is strictly serial, so no parallelism at all.

wait_for_downstream=True is stricter still — it waits for the previous run’s downstream tasks too. Rarely the right answer.

Use depends_on_past only where the data genuinely requires ordering. Most pipelines that use it do not need it, and pay a large availability cost for a property they were not relying on.

What the states mean for a run

   every task SUCCESS or SKIPPED        → run SUCCESS
   any task FAILED (with no downstream
     recovery)                          → run FAILED
   still work to do                     → run RUNNING

A subtlety worth knowing: a run can be success with skipped tasks in it. A short-circuit that correctly determined there was no data today produces a green run where most of the graph is grey. That is right, and it means “the run succeeded” does not mean “the work happened” — which is why chapter 13 argues for alerting on data checks, not just on task state.

What to take from this chapter

The default trigger rule is all_success. The three worth knowing: all_done for cleanup that must always run, none_failed_min_one_success for a join after a branch, and one_success for first-one-wins.
The join-after-branch bug is the most common Airflow logic error: skipped is not successful, so all_success skips the join and the pipeline stops silently.
Retry transient failures with exponential backoff; do not retry deterministic ones. Use AirflowFailException to stop retrying, AirflowSkipException for “nothing to do”.
Set execution_timeout on everything — retries do not help a task that never finishes.
Idempotency is a correctness requirement, not a style. Retries, zombie reruns, manual clears and backfills all mean at-least-once execution. Use delete-then-insert, partition overwrite, upsert, or staging-then-swap.
Clearing is the standard recovery operation — usually with --downstream. It only works if tasks are idempotent.
depends_on_past=True blocks every subsequent run after one failure and serialises backfills. Use it only where the data genuinely requires ordering.
A run can succeed with most of its tasks skipped. “Run succeeded” is not “the work happened” — alert on data, not only on state.

Next: XComs — what they are for, the size limit that matters, and why passing data through Airflow is usually a mistake.

Written by Sepuri Sai Krishna