Production Airflow: Idempotency and Backfills

Consider a hypothetical daily orders pipeline built as an Airflow DAG. On Monday night the load task commits its rows and then loses the connection while waiting for the database’s reply. Airflow marks the attempt failed and retries it, the second attempt inserts the same rows again, and Tuesday’s dashboard shows double revenue while every task in the run is green. On Wednesday an engineer starts a 90-day backfill to repair a calculation bug. It finishes after the live run has published a correction, and it replaces that correction with older numbers. On Thursday a month-end spike fills the shared extract pool with heavy tasks, and a task that needs two slots waits all morning while one-slot tasks keep passing it.

None of these is an Airflow defect. Each is a rule the pipeline had to state and the scheduler could not state for it: what a repeated write does to its destination, which failures deserve another attempt, how shared capacity is rationed, how a task waits for something outside Airflow, and how a historical rerun and a live run publish into the same place. This article works through those rules with programs that reproduce each failure and the repair.

It assumes the vocabulary of Apache Airflow Fundamentals: DAGs, Tasks, and Scheduling (a DAG run, a task instance, a data interval) and the repeat-safe load ideas of Building Incremental and Idempotent Pipelines. How the scheduler, workers, and metadata database fail, and how to recover from each, is the subject of Airflow Operations: Retries, Backfills, and Safe Recovery, which this article leaves to that follow-up.

The programs run on Python 3.12. Those that import Airflow ran on Apache Airflow 3.3.2 with a SQLite metadata database on one machine, using Airflow’s local test runner for tasks and no scheduler service. All source data is synthetic. SQLite stands in for the destination; where a warehouse behaves differently, the text names the property to look for.

Idempotency belongs to the write, not to the task

Airflow retries a task by running its body again from the beginning, against whatever the failed attempt left behind. The Airflow best-practices page draws the consequence for databases in one sentence: “Do not use INSERT during a task re-run”, because the statement can produce duplicate rows, and an upsert is the suggested replacement. That advice is right for one destination shape and incomplete for others. Whether a repeat is harmless depends on what the write does when the first attempt’s effects are already present. The glossary entry on idempotency defines the property (repeating the same logical operation has the same intended effect as applying it once); the question here is which write patterns have it, and against which failures.

The experiment is one task that writes one day of orders. The first attempt reads the source, writes, and then loses the acknowledgement, so the destination holds the result while the scheduler records a failure. The retry reads the source again. In one case the source is unchanged. In the other, a correction arrived between the attempts: order o3 was canceled and order o2 changed from 20 to 25, so the correct total drops from 60 to 35. Five write patterns face the same two cases. Four are shown here: an append, an upsert keyed by order, a delete-and-insert inside one transaction, and a stage-and-swap.

def w_append(db, rows, attempt):
    db.execute("BEGIN")
    db.executemany("INSERT INTO append_t VALUES (?,?,?)", [(DAY, o, a) for o, a in rows])
    db.execute("COMMIT")

def w_upsert(db, rows, attempt):
    db.execute("BEGIN")
    db.executemany("INSERT INTO keyed_t VALUES (?,?,?) ON CONFLICT(day, order_id) DO UPDATE SET amount=excluded.amount",
                   [(DAY, o, a) for o, a in rows])
    db.execute("COMMIT")

def w_partition_txn(db, rows, attempt):
    db.execute("BEGIN")
    db.execute("DELETE FROM part_t WHERE day=?", (DAY,))
    db.executemany("INSERT INTO part_t VALUES (?,?,?)", [(DAY, o, a) for o, a in rows])
    db.execute("COMMIT")

def w_stage_swap(db, rows, attempt):
    db.execute("DELETE FROM stage_t WHERE attempt=?", (attempt,))   # leftovers of this same attempt id
    db.executemany("INSERT INTO stage_t VALUES (?,?,?,?)", [(attempt, DAY, o, a) for o, a in rows])
    db.execute("INSERT INTO live VALUES (?,?) ON CONFLICT(day) DO UPDATE SET attempt=excluded.attempt", (DAY, attempt))

The fifth pattern, delete-then-insert as two separate commits (w_partition_split), is in the same file. The harness runs attempt 1 on the first version of the source, lets it commit, then runs attempt 2 on the second version and compares the destination’s total with the total of the source that attempt 2 read.

pattern                            case                         rows  sum expected  verdict
append                             same input on retry             6  120       60  WRONG
append                             source corrected before retry    5   95       35  WRONG
upsert                             same input on retry             3   60       60  ok
upsert                             source corrected before retry    3   65       35  WRONG
partition replace (1 txn)          same input on retry             3   60       60  ok
partition replace (1 txn)          source corrected before retry    2   35       35  ok
delete, then insert (2 commits)    same input on retry             3   60       60  ok
delete, then insert (2 commits)    source corrected before retry    2   35       35  ok
stage + pointer swap               same input on retry             3   60       60  ok
stage + pointer swap               source corrected before retry    2   35       35  ok

What a reader sees if the worker dies inside the write
  single transaction, killed before commit : sum = 60 (old result intact)
  two commits, killed between them         : sum = 0 (partition is empty until a retry succeeds)
  stage + swap, try2 died while staging    : sum = 60 (readers still follow try1)

Read the table of results row by row. The append doubles the day in both cases (120 against 60, and 95 against 35, because the first attempt’s three rows stay and the second attempt’s two rows join them). The upsert survives a retry on identical input: the same three keys receive the same values. It fails when the recomputed set shrinks. Order o3 is no longer in the source, so nothing in the second attempt touches its row, and the day totals 65 instead of 35. A key-based write only adds or changes keys that the new computation produces; it cannot notice a key the computation stopped producing. If the source can cancel, merge, or drop rows, an upsert needs a second step that deletes keys absent from the new result, or the destination should be replaced by partition. The entries on upsert and merge and partition overwrite describe those two shapes.

The partition replace, the delete-then-insert, and the stage-and-swap all end correct in both cases. They differ in what a reader can see on the way. The last three lines of output show a worker killed inside each write. In a single transaction, a kill before the commit leaves the old result intact. With two separate commits, a kill between the delete and the insert leaves the partition empty, and it stays empty until a retry succeeds; the final state is correct, but a dashboard that refreshed in the gap reported zero. With stage-and-swap, the dead attempt wrote only into its own staging rows, and readers follow the live pointer, which still names the previous attempt.

Write pattern	Retry with the same input	Source rows disappear	Reader sees a partial state	What it requires
Append	Duplicates	Duplicates	No	Nothing can make it repeat-safe without a dedup key or a ledger
Upsert by key	Safe	Stale rows remain	No	A stable key; a separate step for removed keys
Delete and insert, one transaction	Safe	Safe	No	A destination with multi-statement transactions over the whole partition; the complete desired contents, including an empty set
Delete and insert, two commits	Safe after the retry	Safe after the retry	Yes, between the commits	Readers that tolerate a gap, which is rarely the case
Stage, then swap a pointer	Safe	Safe	No	A single atomic pointer update; readers that resolve the pointer once; cleanup of abandoned staging data

The transaction row depends on the destination. A relational database provides it directly (the entry on SQL transactions covers the boundary of what a transaction protects). A directory of files on object storage does not offer a multi-file atomic replace, and a warehouse may or may not let one transaction cover the delete and the insert for the table in question. Where there is no such transaction, the stage-and-swap shape is the portable answer: build the complete candidate under a name derived from the operation, validate it, then move one reference. The entry on atomic publication explains why the reference has to be a single update and why readers must resolve it once.

When two attempts overlap

Retries are sequential: the next attempt starts after the previous one has failed. Overlap happens when an attempt is declared lost while its process is still alive (a worker that cannot reach the metadata database for a while, or a task that someone clears while it runs). The lost attempt may read the old source, wake up late, and commit after its replacement. Each commit is atomic, so the destination never shows a partial result, yet the last committer wins and it is the stale one.

class StaleWriter(Exception): pass

def replace_partition(version, rows, guarded):
    db.execute("BEGIN IMMEDIATE")
    try:
        if guarded:
            db.execute("INSERT OR IGNORE INTO part_meta VALUES (?, 0)", (DAY,))
            cur = db.execute("UPDATE part_meta SET source_version=? WHERE day=? AND source_version < ?", (version, DAY, version))
            if cur.rowcount != 1:
                raise StaleWriter(f"source_version {version} is not newer than the stored one")
        db.execute("DELETE FROM part_t WHERE day=?", (DAY,))
        db.executemany("INSERT INTO part_t VALUES (?,?,?)", [(DAY, o, a) for o, a in rows])
        db.execute("COMMIT")
    except Exception:
        db.execute("ROLLBACK"); raise

guarded=False: late attempt committed; partition sum = 60 (corrected value is 35)
guarded=True : late attempt rejected (source_version 1 is not newer than the stored one); partition sum = 35 (corrected value is 35)

The unguarded run ends at 60, which is the first attempt’s stale result. The guarded run rejects the late writer because the guard is part of the same transaction as the replacement: it moves a stored version forward only if the incoming version is higher, and rolls everything back otherwise. Which value to compare depends on what staleness means. When corrections arrive, compare the version of the input snapshot an attempt read. When a lost attempt wakes up after its replacement began, the task instance’s attempt number orders the two. The principle is the one in fencing tokens: a number that only increases, checked by the resource that is being protected, in the same atomic step as the write. A scheduler that has decided an attempt is lost cannot stop that attempt from writing; only the destination can.

What makes two attempts the same operation

A repeat is recognizable only if the operation has an identity that every attempt computes the same way. The usual identity combines the interval the run owns, the version of the input it reads, and the version of the transformation. What decides whether the work is already done must derive from that identity: the partition key, the candidate version id that a pointer or a compare-and-swap names, and an entry in a batch load ledger. A staging area may still be named per attempt, as in the stage-and-swap above, so that overlapping attempts never share rows. Values that change per attempt, such as the wall clock, a random id, or the attempt number, belong in diagnostics and in ordering guards, not in identity. If the identity includes a per-attempt value, no later attempt can tell that an earlier one already did the work.

Effects outside the destination are not covered by any of this. A database transaction does not retract an email or a webhook, and a primary key does not stop a second HTTP call. Keep those effects in their own task after the validated publish, give each a recorded identity so a repeat can detect that it already happened, and set retries on that task according to whether the receiving system tolerates duplicates. The glossary entry on task instances and retries summarizes how attempts relate to a task instance.

Retries: classify the failure before spending attempts

A retry is a bet that the next attempt will meet different conditions. The bet is sound for transient causes, wasteful for permanent ones, and dangerous when the outcome of the failed attempt is unknown. Three classes follow from that. A transient failure (a connection reset, a throttled request, a briefly unavailable service) is worth retrying with a delay. A permanent failure (an expired credential, a missing object, a schema mismatch) is not; retrying only postpones the alert by the sum of the delays. An ambiguous failure is one that may have changed state before it was reported, such as the lost acknowledgement above; retrying it is safe exactly when the write is idempotent. The entry on networks and I/O for data engineers examines the ambiguous case for network calls in more detail.

Counting attempts and choosing which failures to retry

In Airflow, retries counts retries after the first attempt, so retries=3 permits up to four attempts. The task instance sits in up_for_retry between attempts. A task can raise AirflowFailException to fail immediately regardless of the remaining retries, which suits conditions that retrying cannot fix. Airflow 3.3 adds retry policies: a retry_policy on a task maps exception types to an action, which can retry with a specific delay, fail immediately, or fall back to the default behavior. The documentation states two limits: the policy can fail a task earlier but cannot raise the number of retries, and AirflowFailException always takes precedence.

The next program defines three tasks that all allow three retries. One raises ConnectionError on its first two attempts. Two raise PermissionError on every attempt, one of them under a policy that classifies it as permanent.

POLICY = ExceptionRetryPolicy(rules=[
    RetryRule(exception=ConnectionError, action=RetryAction.RETRY, retry_delay=timedelta(seconds=1), reason="transient network"),
    RetryRule(exception=PermissionError, action=RetryAction.FAIL, reason="credential problem"),
])

@dag(dag_id="retry_policy_demo", start_date=pendulum.datetime(2024, 10, 1, tz="UTC"), schedule=None, catchup=False)
def retry_policy_demo():
    @task(retries=3, retry_delay=timedelta(seconds=1), retry_policy=POLICY)
    def network_blip():
        ti = get_current_context()["ti"]
        print(f"network_blip try_number={ti.try_number}")
        if ti.try_number < 3:
            raise ConnectionError("connection reset")
        print("network_blip ok")

    @task(retries=3, retry_delay=timedelta(seconds=1), retry_policy=POLICY)
    def bad_credential():
        ti = get_current_context()["ti"]
        print(f"bad_credential try_number={ti.try_number}")
        raise PermissionError("token expired")

    @task(retries=3, retry_delay=timedelta(seconds=1))
    def bad_credential_no_policy():
        ti = get_current_context()["ti"]
        print(f"bad_credential_no_policy try_number={ti.try_number}")
        raise PermissionError("token expired")

bad_credential             final state: failed   attempts made: 1
bad_credential_no_policy   final state: failed   attempts made: 4
network_blip               final state: success  attempts made: 3

The task with the policy and a transient error retried twice and succeeded on the third attempt. The same error class under the same policy that maps PermissionError to FAIL stopped after one attempt, and the log recorded the decision with its reason (“credential problem”). Without a policy, the third task spent all four attempts on an error that could not clear. Over a real schedule, with real delays, that is the difference between an alert in seconds and an alert after the longest retry chain. The local runner executes tasks one after another, so these results show the decisions, not the timing.

Spacing the attempts

Delays matter when many tasks fail for the same reason. A shared database that restarts makes every task that touches it fail at once, and tasks that all wait the same fixed delay return together. Airflow’s exponential backoff multiplies the delay by retry_exponential_backoff for each failed try and adds an offset derived from a hash of the DAG id, task id, logical date, and try number, so tasks spread out deterministically instead of randomly. In the 3.3.2 source, for a base delay \(b\) in seconds, a multiplier \(m\), a failed try number \(n\), and a cap \(c\) (max_retry_delay), the delay is

\[ d_n = \min\bigl(c,\; B_n + (h \bmod B_n)\bigr), \qquad B_n = \lceil b\, m^{\,n-1} \rceil \]

where \(h\) is the hash. The term \(h \bmod B_n\) lies between 0 and \(B_n – 1\), so before the cap applies every task waits from \(B_n\) up to just under \(2B_n\) seconds. The next program calls Airflow’s own method for this computation on stand-in objects, with all 200 hypothetical tasks failing at the same instant.

def delay_seconds(task_id, try_number, base, multiplier, max_delay=None):
    task = SimpleNamespace(retry_delay=base, retry_exponential_backoff=multiplier, max_retry_delay=max_delay)
    ti = SimpleNamespace(retry_delay_override=None, task=task, try_number=try_number, dag_id="orders",
                         task_id=task_id, logical_date=LOGICAL, end_date=END, log=logging.getLogger("x"))
    return (TaskInstance.next_retry_datetime(ti) - END).total_seconds()

retry delay for one task (base 30 s, multiplier 2, max_retry_delay 600 s):
  after failed try 1:     35 s   (uncapped window [30, 60) )
  after failed try 2:     69 s   (uncapped window [60, 120) )
  after failed try 3:    187 s   (uncapped window [120, 240) )
  after failed try 4:    403 s   (uncapped window [240, 480) )
  after failed try 5:    558 s   (uncapped window [480, 960) )
  after failed try 6:    600 s   (uncapped window [960, 1920) )

  fixed 30 s delay          : busiest 10 s bucket starts 200 retries (bucket begins 30 s after the failure)
  exponential, hash-jittered: busiest 10 s bucket starts  72 retries (bucket begins 30 s after the failure)
  delays range from 30 s to 59 s
  try 1 with max_retry_delay=600:  30 distinct delays among 200 tasks, range 30-59 s
  try 4 with max_retry_delay=600: 135 distinct delays among 200 tasks, range 242-478 s
  try 6 with max_retry_delay=600:   1 distinct delays among 200 tasks, range 600-600 s

The first block shows the doubling window for one task. The second compares 200 tasks that fail together. With a fixed delay all 200 retries start in one 10-second bucket. With the backoff offset, the busiest bucket starts 72, because the offset spreads the first retry only over a window as wide as the base delay (30 to 59 seconds, 30 distinct values). The third block shows a trap: once the cap binds, the offset is cut off too. At try 6 the uncapped window is 960 to 1,920 seconds, the cap is 600, and all 200 tasks wait exactly 600 seconds, so the last retries arrive in lockstep. A cap should sit above the windows that matter, or the final attempts of correlated failures should be protected another way, for example by a pool.

Marc Brooker’s analysis of retry schemes for AWS compares variants and reports that full jitter, which draws each wait uniformly from zero up to the exponential bound, does the least total client work while finishing about as fast as the alternatives. Airflow’s offset is a different scheme: no attempt retries sooner than its exponential base, and the spread is at most one extra base. That is a reasonable default for scheduled batch work, where nobody waits on one attempt, and it is the reason to treat the delay settings as a load-shaping tool and not as a politeness setting. A retried task instance is scheduled again like any other and passes the scheduler’s pool check again, so a pool caps how many retries hit a shared dependency at once whatever their delays.

Timeouts and retries combine into a worst case. With \(r\) retries, a per-attempt execution_timeout \(T\), and delays \(d_1, \dots, d_r\), a task that always fails by timing out occupies its slot for up to \((r+1)\,T + \sum d_i\). Compare that with the deadline the output protects: if consumers need the data by a given time, the worst case plus time to notice and repair has to fit before it. The entry on data freshness shows how to derive such a deadline from a consumer’s requirement.

Pools: slot arithmetic with the scheduler’s own admission step

A pool is a named capacity that limits how many of its tasks run at once. Each task instance counts pool_slots slots (one by default) against the pool while it is queued or running, and the scheduler queues no more tasks than the pool has open slots. A task that names no pool uses default_pool, which the documentation says starts with 128 slots. The glossary entry on Airflow pools defines the pieces; the experiments below check its arithmetic against the scheduler’s code.

Reading a configuration does not show what the scheduler will do with it, so the experiment calls the scheduler’s admission step directly. In Airflow 3.3.2 the method SchedulerJobRunner._executable_task_instances_to_queued receives the scheduled task instances, orders them by priority weight, logical date, and map index, and returns the ones that fit the pool. It is a private method, used here as a probe and not as an API. The DAG below defines an 8-slot pool named etl, four extract tasks that need two slots, three check tasks that need one slot, a task that needs nine slots, and a high-priority task that needs two.

import pendulum
from airflow.sdk import DAG
from airflow.providers.standard.operators.empty import EmptyOperator

with DAG("pool_demo", start_date=pendulum.datetime(2024, 10, 1, tz="UTC"), schedule=None, catchup=False) as dag:
    for i in range(1, 5):
        EmptyOperator(task_id=f"extract_{i}", pool="etl", pool_slots=2)
    for i in range(1, 4):
        EmptyOperator(task_id=f"check_{i}", pool="etl", pool_slots=1)
    EmptyOperator(task_id="giant", pool="etl", pool_slots=9)
    EmptyOperator(task_id="heavy", pool="etl", pool_slots=2, priority_weight=100)

Each scenario creates a DAG run, puts chosen task instances in the scheduled or running state, runs one admission step, and reports which task instances were admitted and how many slots stayed open. The two helpers that do the work are short.

def set_states(dr, **states):
    for task_id, st in states.items():
        session.execute(update(TI).where(TI.dag_id == "pool_demo", TI.run_id == dr.run_id, TI.task_id == task_id).values(state=st))
    session.commit(); session.expire_all()

def admit():
    queued = runner._executable_task_instances_to_queued(max_tis=50, session=session)
    session.commit()
    return sorted(ti.task_id for ti in queued)

def open_slots():
    return Pool.get_pool("etl", session=session).open_slots(session=session)

A. 3 extracts (2 slots each) + 3 checks (1 slot each) eligible
  open before: 8; admitted: ['check_1', 'check_2', 'extract_1', 'extract_2', 'extract_3']; still waiting: ['check_3']; open after: 0
B. 4 extracts + 3 checks eligible
  open before: 8; admitted: ['extract_1', 'extract_2', 'extract_3', 'extract_4']; still waiting: ['check_1', 'check_2', 'check_3']; open after: 0
C. only the 9-slot task eligible
  open before: 8; admitted: []; still waiting: ['giant']; open after: 8
D. 3 extracts running; 1 extract + 3 checks eligible
  open before: 2; admitted: ['extract_4']; still waiting: ['check_1', 'check_2', 'check_3']; open after: 0
E. heavy (2 slots, priority_weight=100) versus one-slot checks, 3 extracts keep running
  round 1 (check_1 running; heavy and check_2 eligible): ['check_2'] open after: 0
  round 2 (check_1 done; check_3 eligible)             : ['check_3'] open after: 0
  round 3 (check_2 done; only heavy eligible)          : [] open after: 1
  round 4 (check_3 done)                               : ['heavy'] open after: 0
F. 4 extracts are deferred (2 slots each); 1 check eligible
  include_deferred=False: open slots 8; admitted: ['check_1']
  include_deferred=True : open slots 0; admitted: []

Scenario A matches the arithmetic in the glossary entry. Three extracts occupy six of eight slots, leaving two, so two of the three one-slot checks fit and the third waits. In scenario B four extracts fill all eight slots and every check waits for an extract to finish. The mix in A depends on the order among tasks of equal priority: here the extracts were taken first. With a different order, checks could take slots first and an extract could be left waiting, so a pool shared by tasks of different weights should set priority_weight deliberately and not rely on tie order. Scenario C is a configuration error that produces no error in the DAG run. A task that asks for nine slots in an eight-slot pool is never admitted and stays in scheduled. The scheduler source logs a warning that the requested slots exceed the pool’s total, and it counts the task as starving, which the scheduler reports through its pool.starving_tasks metric when metrics are enabled. A growing starving count with open slots remaining is the signature of a request that cannot fit.

Scenario D shows that running tasks are subtracted before anything new is admitted: with three extracts running, two slots remained, and exactly one more extract fit. Scenario E is the one that surprises people. The task named heavy needs two slots and has a priority weight of 100, far above the checks’ weight of 1. In round 1 only one slot was open, so the scheduler skipped it and admitted a one-slot check. In round 2 one slot opened again and another check took it. In round 3 a slot sat idle while the heavy task waited. Only in round 4, when two slots were free at the same moment, was it admitted. The admission loop does not hold slots back for a higher-priority task that does not yet fit; it moves on to lower-priority tasks that do. A steady stream of small tasks can therefore delay a large one for as long as small tasks keep taking each slot as it frees. The remedies are structural: a separate pool for heavy tasks, lowering the slot count a heavy task requests, or limiting how many small tasks may enter in front of it.

Scenario F uses the pool option that controls whether deferred tasks occupy slots. With four deferred extracts of two slots each, the pool reported eight open slots when deferred tasks were not counted and none when they were, and the check was admitted only in the first configuration. A deferred task has released its worker, but it may still represent work that holds an external resource. If a deferred task is merely waiting for a file, it should not count. If it is waiting for a long query it submitted to a warehouse, and the pool exists to cap warehouse concurrency, it should count, because the pool is the only place that limit is enforced.

What a pool does not limit

A pool limits tasks that name it. It does not limit tasks in other Airflow deployments or traffic from outside Airflow, and it limits concurrency, not request rate. A pool of four slots allows four simultaneous tasks whether each makes one request or ten thousand; the entry on concurrency and request rate separates the two quantities. The slot weights are an allocation policy chosen by people, not a measurement of connections or CPU. Verify the protection by watching the destination’s actual connection count and the pool’s open and starving counts during a busy period.

Sensors in poke mode complicate the arithmetic. The Airflow sensor documentation notes that in poke mode a sensor holds a worker slot and a pool slot for its whole runtime. If eight poke sensors wait in an eight-slot pool for files that eight extract tasks in the same pool would create, no extract is ever admitted: it is scenario B with sensors as the occupants, with no open slot for any extract, and the sensors wait until they time out. The next section covers the alternatives.

Waiting for an outside condition: poke, reschedule, and deferral

Many tasks cannot start until something outside Airflow is ready: a file lands, a partner’s export completes, a table reaches a row count. Airflow offers three ways to wait. A sensor in poke mode, the default, keeps its worker slot for its entire runtime and sleeps between checks. In reschedule mode the sensor frees the worker between checks and the scheduler starts it again after poke_interval. A deferrable operator or sensor hands the wait to a trigger, a small asynchronous coroutine that runs inside a process called the triggerer; when the trigger fires, the scheduler resumes the task on a worker. The glossary entry on deferrable operators and sensors gives the definitions. The comparison below is about cost.

Mode	Holds a worker while waiting	Where the checking runs	Needs	Timeouts to know
Poke	Yes, plus a pool slot	Inside the task process	Nothing extra	timeout and execution_timeout measure the same elapsed time
Reschedule	Only during each check	A fresh task start per check	Interval long enough not to load the scheduler	timeout counts elapsed time since the first check; execution_timeout counts only running time
Deferral	Only at the start and at resume	A trigger in the triggerer	A running triggerer and an operator with a trigger	execution_timeout covers total runtime including the deferred time; defer() takes its own timeout

The Airflow documentation puts the trade-off as latency: something checked every second suits poke, something checked every minute suits reschedule. The base sensor class documents that a reschedule interval should be more than one minute to avoid loading the scheduler. These statements follow from fixed costs. Every reschedule is a new task start, and every start has overhead on the scheduler, the executor, and the worker. The next program prices that overhead with hypothetical numbers: a file arrives at a random time between 0 and 90 minutes after the wait begins, every task start or resume holds a worker slot for 20 seconds of fixed overhead, and each check of the source costs one second. These are illustrative assumptions, not measurements of any deployment.

rng = random.Random(7)
N, OVERHEAD, CHECK = 20000, 20.0, 1.0
arrivals = [rng.uniform(0, 90 * 60) for _ in range(N)]

def poke(a, interval):
    checks = math.floor(a / interval) + 1
    detect = checks * interval - a
    return a + detect, detect, checks                       # slot is held from the first check until the one that succeeds

def reschedule(a, interval):
    checks = math.floor(a / interval) + 1
    detect = checks * interval - a
    return checks * (OVERHEAD + CHECK), detect, checks      # a fresh task start for every check

def deferrable(a, interval):
    checks = math.floor(a / interval) + 1
    detect = checks * interval - a
    return 2 * OVERHEAD + CHECK, detect, checks             # one start before deferral, one resume after the trigger fires

mode                                     worker-slot s detect delay s checks of source
poke, every 60 s                                  2718           29.9             45.3
reschedule, every 60 s                             951           29.9             45.3
reschedule, every 300 s                            199          150.2              9.5
deferrable, trigger polls every 30 s                41           15.1             90.1
deferrable, trigger polls every 300 s               41          150.2              9.5

Two results stand out. First, the interval sets the detection delay, not the mode. With arrivals spread uniformly, the average delay is half the interval for every mode (about 30 seconds at 60 seconds, 150 seconds at 300 seconds, 15 seconds at 30 seconds). Second, the modes differ in slot cost. Poking every minute holds a slot for about 2,700 seconds on average, which is the wait itself. Rescheduling every minute cuts that to about 950 seconds, because the fixed overhead is paid 45 times. Rescheduling every five minutes cuts it to about 200. Deferral costs 41 seconds regardless of the wait, since only the start and the resume use a worker.

Deferral does not reduce traffic to the source. A trigger that polls every 30 seconds makes 90 checks per wait on average, ten times the checks of a five-minute reschedule. The saving is in worker and scheduler resources, and the poll interval is still a decision about detection delay against load on the external service. What changes is the cost of choosing a short interval: a trigger is a coroutine, and by default one triggerer runs up to 1,000 triggers at once (the [triggerer] capacity setting). Because all of a triggerer’s triggers share one event loop, the documentation warns that a trigger that makes a blocking call without awaiting it can stall the whole triggerer process.

What the resumed task knows

The program below runs a small deferrable operator and a reschedule-mode sensor in Airflow’s test runner. The operator defers to a trigger that waits for a marker file and returns its contents. When the task resumes, it must decide whether that marker is the one it was waiting for.

class WaitForMarker(BaseOperator):
    """Deferrable wait for a delivery marker; resumes only on the interval it was built for."""
    def __init__(self, expected_start, **kw):
        super().__init__(**kw)
        self.expected_start = expected_start          # constructor argument: part of the task definition
    def execute(self, context):
        self.scratch = "set during execute()"         # ordinary instance state
        print(f"execute(): try_number={context['ti'].try_number}; deferring")
        threading.Thread(target=deliver_marker, args=(2.0, "2024-10-01" if MODE == "good" else "2024-09-30"), daemon=True).start()
        from wait_triggers import MarkerTrigger
        self.defer(trigger=MarkerTrigger(MARKER), method_name="resume", timeout=30)
    def resume(self, context, event=None):
        print(f"resume(): try_number={context['ti'].try_number}; scratch survived? {hasattr(self, 'scratch')}; "
              f"expected_start still set? {self.expected_start}")
        print(f"resume(): event={event}")
        if event["interval_start"] != self.expected_start:
            raise ValueError(f"marker is for {event['interval_start']}, expected {self.expected_start}")
        return event["version"]

def marker_ready():
    pokes.append(len(pokes) + 1)
    return len(pokes) >= 3                              # the source "delivers" on the third check

import asyncio, json, os
from airflow.triggers.base import BaseTrigger, TriggerEvent

class MarkerTrigger(BaseTrigger):
    """Waits for a marker file, then reports what the marker says."""
    def __init__(self, path, poll_interval=0.2):
        super().__init__()
        self.path, self.poll_interval = path, poll_interval
    def serialize(self):
        return ("wait_triggers.MarkerTrigger", {"path": self.path, "poll_interval": self.poll_interval})
    async def run(self):
        while not os.path.exists(self.path):
            await asyncio.sleep(self.poll_interval)
        with open(self.path) as f:
            yield TriggerEvent(json.load(f))

The marker writer in the experiment delivered a marker for the right interval in one run (good) and for the previous day’s interval in the other (stale).

execute(): try_number=1; deferring
resume(): try_number=1; scratch survived? False; expected_start still set? 2024-10-01
resume(): event={'interval_start': '2024-10-01', 'version': 7}
RESULT deferred_wait success try_number 1 reschedule rows 0
RESULT reschedule_wait success try_number 1 reschedule rows 2
RESULT sensor callable invocations: 3

execute(): try_number=1; deferring
resume(): try_number=1; scratch survived? False; expected_start still set? 2024-10-01
resume(): event={'interval_start': '2024-09-30', 'version': 7}
RESULT deferred_wait failed try_number 1 reschedule rows 0
RESULT reschedule_wait success try_number 1 reschedule rows 2
RESULT sensor callable invocations: 3

In the good run the deferral did not count as a retry: try_number stayed at 1 across the start and the resume. The method that resumed the task ran on a freshly constructed operator. The attribute assigned during execute() was gone, while the constructor argument expected_start, which is part of the task’s definition, was present. The Airflow documentation states the rule: the only way to pass state from the old operator instance to the new one is the resume method name and its keyword arguments, which must be serializable.

In the stale run the trigger fired and the wait finished, and the task still failed, because the resume method compared the marker’s interval with the one the task owned. A trigger proves only the condition it was given. A marker that does not name the interval and the version of what it announces cannot tell a delivery for today from a leftover from yesterday, and a wait that returns on any file in a directory releases downstream work on stale input. Put the interval and version into the marker, check them on resume, and still validate row counts or checksums before publishing, because a marker proves that a delivery was announced, not that it is complete. The timeout=30 argument of defer() bounds the wait in the other direction: a marker that never arrives fails the task instead of leaving it deferred forever.

The reschedule sensor was invoked three times, which produced two reschedule records, and its try_number also stayed at 1. Rescheduling is not a retry and does not consume the task’s retries, so the sensor’s timeout is what bounds the wait. It measures elapsed time from the first check, reschedule gaps included, and a sensor that exceeds it fails without being retried, as the sensor documentation describes. Set it from the freshness deadline of the work that follows, as with the retry worst case.

Apply the cost model to a file expected in about an hour. Poke holds a worker slot for the whole hour, 3,600 seconds. Rescheduling every five minutes starts the task about 13 times, roughly 270 seconds of slot time, and delays detection by up to five minutes. A deferrable sensor holds a worker for about 40 seconds and can poll as often as the source tolerates. The triggerer is the new constraint: at the default capacity of 1,000 triggers per triggerer, 2,500 waits open at once need at least three triggerers. Choose poke for waits that end within minutes at a short interval, reschedule when the wait is long and several minutes of detection delay are acceptable, and deferral when the provider ships a deferrable variant of the operator you need, or the condition is naturally asynchronous (such as a job submitted to an external system), you operate enough triggerer capacity, and any trigger code you write never blocks.

Backfills as an operation with a plan

A scheduler creates runs for intervals that are due. A backfill is an explicit request to create runs for a chosen range of past intervals, usually to apply a fix or fill a gap. The entry on Airflow catchup separates it from the scheduler’s catch-up policy, and the entry on data backfill lists what a date range alone leaves undefined: the input versions, the transformation version, and the plan for publishing. The follow-up article covers how a backfill fits into recovery. Here the focus is on the controls that decide how much harm a backfill can do.

Plan with a dry run

The Airflow 3 CLI can show what a backfill would create before it creates anything. The following command was run against a DAG with a daily data-interval schedule that starts on October 1, 2024. It printed the parameters and the logical dates it would attempt.

airflow backfill create --dag-id daily_load --from-date 2024-10-01 --to-date 2024-10-05 --dry-run --reprocess-behavior failed

Performing dry run of backfill.
Printing params:
    - dag_id = daily_load
    - from_date = 2024-10-01 00:00:00+00:00
    - to_date = 2024-10-05 00:00:00+00:00
    - reprocess_behavior = ReprocessBehavior.FAILED
    - run_on_latest_version = True
Runs to be attempted:
| logical_date              | partition_key   | partition_date   |
| 2024-10-01 00:00:00+00:00 |                 |                  |
| 2024-10-02 00:00:00+00:00 |                 |                  |
| 2024-10-03 00:00:00+00:00 |                 |                  |
| 2024-10-04 00:00:00+00:00 |                 |                  |
| 2024-10-05 00:00:00+00:00 |                 |                  |

Three details in that output matter. First, the end date is inclusive: five logical dates for a window from October 1 to October 5. For a data-interval schedule the logical date is the start of the interval, so the run for October 5 owns the interval that ends on October 6. If the end date is today, that interval has not ended, and the run would read data that is still arriving. Second, reprocess_behavior decides which existing runs are replaced. The documentation defines none (the default: create nothing for a logical date that already has a run), failed (create a new run when the existing one failed), and completed (create a new run when the existing one completed or failed), and says a run that is still running or queued is never duplicated. Choosing failed repairs an outage and leaves successes alone. Choosing completed recomputes everything in the range, which a code fix needs and an outage repair does not. Third, run_on_latest_version = True: the CLI help for Airflow 3.3.2 says backfills run on the latest bundle version unless told otherwise. A backfill therefore recomputes history with the current code. That is what a bug fix needs, and it is a different thing from replaying history as it originally ran. A DAG whose code changed since the original runs will produce different outputs for the same intervals, and the destination write has to treat the result as a replacement, which is the reason for the partition replace in the first section.

Throughput is set by the tightest limit

The max_active_runs option of a backfill applies to the backfill and is independent of the DAG’s own max_active_runs, according to the documentation. Neither is the only bound. Task instances of a backfill run are ordinary task instances for the scheduler: in the admission code read for the pool experiments, the query selects task instances of running DAG runs and treats pool slots without regard to how the run was created. So the effective concurrency is the smaller of the runs allowed and the number of tasks the pool can hold. The next program computes the duration of a hypothetical 90-day backfill whose runs each contain one 20-minute task that needs two slots of an 8-slot pool.

DAYS, MINUTES, SLOTS_PER_TASK, POOL = 90, 20, 2, 8

def hours(backfill_max_active_runs, pool_slots_left):
    by_runs = backfill_max_active_runs
    by_pool = pool_slots_left // SLOTS_PER_TASK
    concurrency = min(by_runs, by_pool)
    waves = math.ceil(DAYS / concurrency)
    return concurrency, waves * MINUTES / 60

backfill max_active_runs pool slots left for it tasks at once  hours
                       1                      8             1   30.0
                       3                      8             3   10.0
                       3                      6             3   10.0
                       6                      8             4    7.7
                       6                      6             3   10.0
                       6                      2             1   30.0

Raising max_active_runs from 3 to 6 changes the duration only from 10.0 to 7.7 hours, because the pool allows at most four of these tasks at once. If live runs already hold two of the eight slots, six allowed runs are worth no more than three and the backfill takes 10.0 hours; if they hold six, the backfill gets one task at a time and takes 30 hours. Two choices follow from this. The backfill can have its own pool, so its load is bounded and the live schedule keeps its slots, or the date range can be split so that the repair of recent intervals finishes first. The --run-backwards option runs the most recent intervals first, for cases where recent data matters most.

A backfill also does not remove the need for an idempotent destination. A backfill run for a logical date can run alongside the live run for a later date, or can replace output that a live run has already corrected. Part of the defense is the write pattern from the first section. The remaining part is how the result becomes visible, which is the subject of the next section.

Publishing without clobbering: compare-and-swap

Many pipelines end with a small update that makes a result current: a pointer, a manifest, a “latest” row that readers consult. Two writers can target it, a live run and a backfill, and each decided what to publish from the state it observed earlier. The mechanism that prevents a stale decision from taking effect is compare-and-swap (CAS): an update that applies only if the current value still equals the value the writer expected, with the comparison and the change forming one atomic operation.

The program below uses a SQLite pointer table with a version name and a revision counter. Four experiments follow in order: an unconditional update, the same race under a conditional update, a value that changes and changes back, and a commit whose acknowledgement is lost.

# ---------- 1. read, then write unconditionally ----------
reset("v1")
live, backfill = conn(), conn()
seen_live = live.execute("SELECT version FROM current").fetchone()[0]          # both observe v1
seen_back = backfill.execute("SELECT version FROM current").fetchone()[0]
live.execute("UPDATE current SET version='v2-live', revision=revision+1")      # live run publishes the fresh result
backfill.execute("UPDATE current SET version='v3-backfill', revision=revision+1")   # backfill finishes later
print("1. unconditional update      : both observed", seen_live, "; pointer ends at", state())

# ---------- 2. compare-and-swap ----------
def cas_publish(c, expected, new):
    cur = c.execute("UPDATE current SET version=?, revision=revision+1 WHERE id=1 AND version=?", (new, expected))
    return cur.rowcount == 1
reset("v1")
live, backfill = conn(), conn()
ok_live = cas_publish(live, "v1", "v2-live")
ok_back = cas_publish(backfill, "v1", "v3-backfill")
print("2. compare-and-swap          : live ok =", ok_live, "; backfill ok =", ok_back, "; pointer ends at", state())

# ---------- 3. ABA: the value comes back, the history does not ----------
reset("v1")
slow = conn(); observed = slow.execute("SELECT version, revision FROM current").fetchone()   # ('v1', 0)
other = conn()
cas_publish(other, "v1", "v2")          # a newer result is published ...
cas_publish(other, "v2", "v1")          # ... and then rolled back to v1 by an operator
ok_value_only = cas_publish(slow, observed[0], "v3-from-stale-input")
print("3. ABA, compare value only   : slow publisher observed", observed, "-> succeeded =", ok_value_only, "; pointer", state())
reset("v1")
slow = conn(); observed = slow.execute("SELECT version, revision FROM current").fetchone()
other = conn(); cas_publish(other, "v1", "v2"); cas_publish(other, "v2", "v1")
cur = slow.execute("UPDATE current SET version=?, revision=revision+1 WHERE id=1 AND version=? AND revision=?",
                   ("v3-from-stale-input", observed[0], observed[1]))
print("   compare value and revision : succeeded =", cur.rowcount == 1, "; pointer", state())

class AckLost(Exception): pass
def publish_task(attempt, candidate, expected, reconcile):
    c = conn()
    ok = cas_publish(c, expected, candidate)
    if attempt == 1:
        raise AckLost("connection reset after commit")       # attempt 1 committed, then lost the reply
    if ok: return "published"
    if reconcile and c.execute("SELECT version FROM current").fetchone()[0] == candidate:
        return "already published by an earlier attempt"
    return "CONFLICT (task fails)"

1. unconditional update      : both observed v1 ; pointer ends at ('v3-backfill', 2)
2. compare-and-swap          : live ok = True ; backfill ok = False ; pointer ends at ('v2-live', 1)
3. ABA, compare value only   : slow publisher observed ('v1', 0) -> succeeded = True ; pointer ('v3-from-stale-input', 3)
   compare value and revision : succeeded = False ; pointer ('v1', 2)
4. reconcile=False  attempt 1: attempt 1 raised AckLost
4. reconcile=False  attempt 2: CONFLICT (task fails)
   pointer after both attempts: ('v2-2024-10-01', 1)
4. reconcile=True   attempt 1: attempt 1 raised AckLost
4. reconcile=True   attempt 2: already published by an earlier attempt
   pointer after both attempts: ('v2-2024-10-01', 1)

Experiment 1 reads and then writes without a condition. Both writers observed v1, the live run published v2-live, and the backfill, which had been computing from the old state, then wrote v3-backfill over it. The revision counter reached 2 and the newest result was lost. Each statement was atomic, so atomicity was not the problem; the read and the write were separate steps and another writer fit between them.

In experiment 2, the update carries the condition WHERE version = expected and the writer checks how many rows changed. The live run’s update changed one row. The backfill’s update changed none, because the pointer no longer held v1, and the pointer stayed at v2-live. In a task, a zero row count is raised as a publication conflict. A conflict is a signal that the writer’s basis is out of date, not an error to retry blindly. The writer has to read the current version and decide: rebuild the candidate on top of the new state, compare it with what is now published and discard it if it is older, or fail the task so a person looks. A loop that rereads the pointer and repeats the same swap with the old candidate reproduces experiment 1 with extra steps.

Experiment 3 shows what an equality check cannot see. The slow publisher observed v1. Meanwhile another writer published v2 and then an operator rolled back to v1. The slow publisher’s compare-on-value update succeeded, because the value matched, even though two changes had happened since it looked, and its candidate was built from input that predates both. This is the ABA problem. Comparing the revision as well as the version closes the gap. In a second run of the same sequence, the update also required the revision observed earlier (0); it had advanced to 2, so the update changed no rows. A counter that only increases distinguishes states that a value alone cannot.

Experiment 4 is the lost acknowledgement from the first section, now at the publish step. The first attempt’s CAS committed and the connection dropped before the reply. On retry the pointer already holds the candidate, so the CAS finds the expected value v1 missing and changes no rows. Without a reconciliation step the task reports a conflict and fails although the work is done. With one, the task reads the pointer, sees that it already names this operation’s candidate, and finishes as a success. That works only if the candidate’s name is derived from the operation’s identity (the interval, input version, and code version), not from the attempt. The first attempt and the second must compute the same name.

What CAS guarantees and what it leaves open

CAS guards one claim: the state the writer based its change on is still current. It does not make a repeat safe by itself, which is idempotency’s job, and the reconciliation step in experiment 4 is a small piece of idempotency added on top. It does not order writer generations either. A fencing token, as in Martin Kleppmann’s account of distributed locks, is a number issued in increasing order that the protected resource uses to reject writers whose number has gone backwards. The version guard in the overlapping-attempts experiment was of that kind, and a CAS on a pointer is related but answers a different question: not who is the newest writer, but whether the thing I read is still the thing that is there. Fencing needs the resource to enforce the order, and so does CAS. A comparison done in application code before an unconditional update is not CAS.

Finally, a successful swap says nothing about the quality of what was published. A pointer can move atomically to a candidate that is wrong or incomplete. Validate the candidate (row counts against a delivery manifest, keys against the source, the interval it covers) before the swap, and use the swap only to make a validated candidate visible. The entry on reconciliation describes what such comparisons can and cannot establish.

Failure injections to run before trusting a DAG

Each mechanism above can be tested without waiting for production to supply the failure. For a DAG that writes data, run these injections in a copy of the destination, and read the destination, not the run status, to judge the result.

1. Kill the task after the write commits and before it returns, then let it retry. Expect the destination to hold the day exactly once.
2. Change the source between the first attempt and the retry so that a row disappears. Expect the removed row to disappear from the destination.
3. Start a second attempt of the same interval while the first is paused, let the first commit last, and confirm the destination rejects it.
4. Run a backfill over an interval whose live run has just published a correction, and confirm the correction survives, or that the conflict is reported to a person.
5. Raise a permanent error and a transient error from the same task, and confirm the permanent one fails on the first attempt and the transient one is retried with a spread of delays.
6. Request more pool slots than the pool has, and confirm an alert fires on the starving count before the data deadline passes.
7. Deliver a marker for the wrong interval and confirm the waiting task refuses to continue.

Passing these confirms the rules in this article, and it does not replace a test of the deployment itself. Behavior that depends on the executor, the metadata database’s isolation, and the warehouse’s transaction support has to be tested in an environment that matches production, which includes the scheduler and metadata failures that the operations follow-up covers.

References

• Apache Airflow 3.3.2 documentation, Best Practices
• Apache Airflow 3.3.2 documentation, Tasks (retries, retry policies, timeouts)
• Apache Airflow 3.3.2 documentation, Pools
• Apache Airflow 3.3.2 documentation, Sensors
• Apache Airflow documentation, Deferrable Operators & Triggers
• Apache Airflow 3.3.2 documentation, Backfill
• Marc Brooker, Exponential Backoff And Jitter (AWS Architecture Blog)
• Martin Kleppmann, How to do distributed locking
• SQLite documentation, UPSERT