Production Patterns: Idempotent, Observable, Alerting

The three habits that separate a DAG you trust in production from one you babysit: safe re-runs, a callback that pages you, and a source database you don't melt.

A DAG that works on the first run is easy. A DAG that works on the second run — after the first one half-failed at 2am and the retry fired against a table that already has today’s rows in it — is the one you actually want. Production isn’t about the happy path. It’s about what happens when a task runs twice, or runs late, or runs against a source that’s already on fire. Three habits get you there: make every task idempotent so re-running is free, make failures loud so you hear about them before your users do, and cap concurrency so one greedy DAG can’t take down the database it reads from. Everything in this post is a variation on those three.

Idempotency: re-running has to be safe

The single most useful property a task can have is that running it again changes nothing. Run it once, run it five times — same result. That’s what makes retries harmless and backfills survivable, and it’s almost always violated by the same mistake: appending.

@task
def load_orders():
    rows = fetch_orders()
    db.execute("INSERT INTO orders SELECT * FROM staging")  # runs twice, doubles the data

Retry that task and you’ve got duplicate rows. Backfill a week and you’ve got seven copies of the overlap. The fix is to scope every write to the run’s data interval and replace that slice, never add to it. The blunt, portable version is delete-then-insert:

Before this works, check your interval. Every window in this section relies on data_interval_start and data_interval_end being different values. On a default Airflow 3 install they are not: a cron schedule uses CronTriggerTimetable, whose data interval is zero-width, so >= start AND < end matches nothing and this task silently does no work. Set AIRFLOW__SCHEDULER__CREATE_CRON_DATA_INTERVALS=True (the foundations scheduling chapter explains why, and what the alternative is). If you inherit a DAG and the numbers look wrong, print both bounds first — if they’re equal, you’ve found it.

@task
def load_orders(data_interval_start=None, data_interval_end=None):
    rows = fetch_orders(since=data_interval_start, until=data_interval_end)
    db.execute(
        "DELETE FROM orders WHERE ordered_at >= %s AND ordered_at < %s",
        (data_interval_start, data_interval_end),
    )
    db.bulk_insert("orders", rows)

Now the task owns exactly one window of the table, and running it again just rewrites that same window. The other half of this rule is where the window comes from: it comes from the run, not the wall clock. The moment a task calls datetime.now() to decide which data to pull, backfills are dead — every historical run would grab today’s rows instead of the day it’s supposed to represent. Airflow hands you data_interval_start and data_interval_end for exactly this reason. Use them, and a backfill of last March produces last March’s numbers.

Delete-insert has one failure mode worth naming: the gap between the DELETE and the bulk_insert. If the task dies in that window, you’ve deleted a day of data and written nothing back, and now the table is worse than before you ran. Two fixes. The first is a transaction — wrap both statements so the delete only commits if the insert does:

@task
def load_orders(data_interval_start=None, data_interval_end=None):
    rows = fetch_orders(since=data_interval_start, until=data_interval_end)
    with db.begin() as tx:  # commits together, rolls back together
        tx.execute(
            "DELETE FROM orders WHERE ordered_at >= %s AND ordered_at < %s",
            (data_interval_start, data_interval_end),
        )
        tx.bulk_insert("orders", rows)

Now a mid-task crash rolls the whole thing back and the retry starts from the same clean state it did the first time. That’s the property you’re actually chasing: not “the task succeeds” but “the table is the same after a failed-and-retried run as after a clean one.”

The second fix, when you key on a natural identifier rather than a time window, is an upsert — a MERGE that inserts new rows and overwrites existing ones by primary key, with no delete at all:

@task
def load_customers():
    rows = fetch_changed_customers()
    db.execute("""
        MERGE INTO customers AS t
        USING staging_customers AS s
        ON t.customer_id = s.customer_id
        WHEN MATCHED THEN UPDATE SET name = s.name, email = s.email, updated_at = s.updated_at
        WHEN NOT MATCHED THEN INSERT (customer_id, name, email, updated_at)
                             VALUES (s.customer_id, s.name, s.email, s.updated_at)
    """)

Run this twice and the second run matches every row it just wrote and updates them to identical values — a no-op with extra steps, which is exactly what idempotent means. MERGE is the right tool when the grain is a business key (a customer, an order, a book) rather than a time slice, because it doesn’t care how many times it runs or in what order relative to its neighbors.

The hard case is the one that has no MERGE: an external side effect. Charging a card, sending an email, kicking a downstream API — a retry of that isn’t a rewrite, it’s a second charge. The pattern here is an idempotency key: derive a deterministic token from the run (the task’s logical date plus the entity), hand it to the external system, and let them dedupe. Every serious payments and messaging API supports this precisely because retries are unavoidable.

@task
def charge_subscriptions(logical_date=None):
    for sub in due_subscriptions(logical_date):
        # same key on every retry of this run → the provider charges once
        payments.charge(
            amount=sub.amount,
            idempotency_key=f"sub-{sub.id}-{logical_date:%Y-%m-%d}",
        )

The key has to be a pure function of the run and the entity — never a UUID generated inside the task, because that changes on every attempt and defeats the whole mechanism. Get this right and the answer to “the task failed after charging half the customers, is it safe to retry?” becomes an unhesitating yes.

Retries: the whole reason idempotency matters

Idempotency is the property; retries are why you want it. A task with no retries is a task that pages a human the first time a network blip drops a connection. Airflow’s retry machinery turns most transient failures into a footnote in the logs, and it’s controlled entirely through default_args or the task decorator:

from datetime import timedelta
from airflow.sdk import dag, task

@dag(
    schedule="@daily",
    catchup=False,
    default_args={
        "retries": 4,
        "retry_delay": timedelta(minutes=2),
        "retry_exponential_backoff": True,
        "max_retry_delay": timedelta(minutes=30),
        "execution_timeout": timedelta(minutes=45),
    },
)
def load_bookshop():
    ...

Each knob earns its place. retries is how many times Airflow re-queues a failed task before giving up and marking it failed for real. retry_delay is the wait between attempts. retry_exponential_backoff=True turns that fixed delay into a doubling one — two minutes, then four, then eight — which is the correct behaviour against a source that’s failing because it’s overloaded, since hammering it every two minutes just prolongs the outage. max_retry_delay caps that exponential growth so you don’t end up waiting six hours between the last two attempts.

execution_timeout is the one people forget, and it’s the most important for protecting the rest of the system. Without it, a task that hangs — a query that never returns, an API that accepts the connection and then goes silent — occupies a worker slot forever, and a few of those will quietly starve your whole deployment. execution_timeout kills the task after its budget and lets the retry logic take over. Set it to something a healthy run comfortably fits inside; a task that’s running twice its normal duration is a task that’s usually stuck.

The catch that ties this back to the previous section: retries are only safe if the task is idempotent. Airflow doesn’t know whether attempt two is picking up cleanly or doubling your data — it just re-runs the same code. retries=4 on a non-idempotent append task isn’t resilience, it’s four chances to corrupt the table. Every retry you configure is a bet that re-running is harmless, which is exactly the bet idempotency lets you win.

Catchup and backfill: idempotency at scale

The place all of this gets stress-tested is backfill, because a backfill runs your task dozens of times in quick succession over historical intervals. Airflow’s catchup controls whether it does this automatically. With catchup=True, deploying a DAG with a start_date three months ago tells the scheduler to run every missed daily interval since — ninety runs, right now.

@dag(
    schedule="@daily",
    start_date=datetime(2026, 3, 1),
    catchup=True,
    max_active_runs=3,
)
def load_bookshop():
    ...

catchup=True is only sane on top of genuine idempotency — those ninety runs each rewrite their own interval, so the end state is correct regardless of order or overlap. max_active_runs=3 is the guardrail: it caps how many DAG runs execute at the same time, so the catchup marches through history three intervals at a time instead of trying to launch all ninety at once and stampeding your source. Most teams keep production DAGs on catchup=False and reach for the explicit airflow backfill create command (or the UI, in 3.x — the old CLI daemon is gone) when they actually want history reprocessed, which makes the intent deliberate rather than a side effect of a start_date.

Two ordering knobs matter when a run genuinely depends on the run before it. depends_on_past=True blocks a task from starting until the same task succeeded in the previous run — the right call when today’s incremental needs yesterday’s to have landed first. wait_for_downstream=True goes further, waiting for the previous run’s downstream tasks too. Both are powerful and both are traps: one failed run with depends_on_past set halts every run after it until you intervene, which is correct for a strict chain and catastrophic for a DAG that just happens to have it switched on. Reach for them when the data model is genuinely sequential, not as a default.

Callbacks: the full family, and where they run

By default a failed task turns red in the UI and does nothing else. In production you want it to reach out. Every task and DAG takes a family of callbacks — functions Airflow invokes with the task context at specific moments in the lifecycle:

  • on_success_callback — the task (or run) finished cleanly.
  • on_failure_callback — it failed for good, after exhausting retries.
  • on_retry_callback — a single attempt failed and another is queued.
  • on_execute_callback — the task is about to start running.
def notify_slack(context):
    ti = context["task_instance"]
    post_to_slack(
        channel="#data-alerts",
        text=f":red_circle: {ti.dag_id}.{ti.task_id} failed "
             f"for {context['logical_date']} — {ti.log_url}",
    )

@dag(schedule="@daily", catchup=False, default_args={"on_failure_callback": notify_slack})
def load_bookshop():
    ...

Setting a callback in default_args wires every task in the DAG at once. You can also set callbacks at the DAG level — @dag(on_failure_callback=...) fires when the run as a whole fails, separately from the per-task ones. The precedence is worth internalizing: a task-level callback and a DAG-level callback are not either/or, they’re different events. The task callback runs once per failed task; the DAG callback runs once for the run. Put “which task broke” logic in the task callback and “the pipeline didn’t finish” logic on the DAG.

Route the different callbacks at different volumes. on_retry_callback is signal you want in a channel you skim, not a page — a retry that eventually succeeds means the system healed itself, which is reassuring, not urgent. on_failure_callback is the one that should reach PagerDuty. on_execute_callback is niche but handy for audit (“this task touched production at this time”). Route retries to Slack, hard failures to the pager, and you’ll stop training yourself to ignore alerts.

Now the sharp edge that catches everyone: callbacks run in the scheduler and dag-processor context, not in your task’s worker. They execute in the same process that’s scheduling every other DAG in your deployment, and Airflow deliberately swallows any exception they raise — a callback that blows up will not fail your task, it’ll just vanish into the scheduler logs. Two consequences. First, keep them cheap and fast: a callback that makes a slow synchronous HTTP call is stealing time from the scheduler’s core loop, and a hundred of them firing at once is a scheduler you’ve quietly degraded. Second, don’t put anything load-bearing in a callback — it’s for notification, not for logic you’re relying on to run. If it must happen, make it a task.

Notifiers: reusable, testable alerting

Hand-rolled callback functions have a smell: every DAG grows its own slightly-different notify_slack, and none of them are tested. Airflow 3.x’s answer is notifiers — a BaseNotifier subclass that packages the “how do I tell someone” logic into a reusable object you drop into any callback slot. The provider packages ship the common ones; SlackNotifier is the one you’ll reach for first:

from airflow.providers.slack.notifications.slack import SlackNotifier

slack_failed = SlackNotifier(
    slack_conn_id="slack_default",
    text="{{ ti.dag_id }}.{{ ti.task_id }} failed for {{ ds }} — {{ ti.log_url }}",
    channel="#data-alerts",
)

@dag(schedule="@daily", catchup=False, default_args={"on_failure_callback": slack_failed})
def load_bookshop():
    ...

The text is a Jinja template rendered against the task context, so you’re not writing f-string plumbing in every DAG. Because a notifier is a plain object, you can define it once in a shared module, unit-test it in isolation, and reuse it across every pipeline you own. Writing your own is a two-method affair — subclass BaseNotifier, declare the templated fields, implement notify(context):

from airflow.sdk import BaseNotifier

class PagerDutyNotifier(BaseNotifier):
    template_fields = ("summary",)

    def __init__(self, summary, severity="critical"):
        self.summary = summary
        self.severity = severity

    def notify(self, context):
        pagerduty.trigger(summary=self.summary, severity=self.severity)

Same object works as on_failure_callback, on_retry_callback, or a DAG-level callback — the notifier abstraction is exactly the “define the channel once, attach it everywhere” shape that raw functions never quite reach.

Deadline alerts: the modern replacement for SLAs

If you’re coming from 2.x, the biggest thing to unlearn here: SLAs are gone. The old sla=timedelta(...) argument that fired a miss callback when a task ran long no longer exists — port it and it’s silently ignored, which is the worst kind of removed. Deadline-style alerting — “tell me if this run hasn’t finished in time” — is now a first-class DAG-level deadline under AIP-86, expressed with a DeadlineAlert:

from datetime import timedelta
from airflow.sdk import DeadlineAlert, DeadlineReference, AsyncCallback

async def deadline_missed(context):
    post_to_slack(channel="#data-alerts", text="bookshop load missed its 2h deadline")

@dag(
    schedule="@daily",
    catchup=False,
    deadline=DeadlineAlert(
        reference=DeadlineReference.DAGRUN_LOGICAL_DATE,
        interval=timedelta(hours=2),
        callback=AsyncCallback(deadline_missed),
    ),
)
def load_bookshop():
    ...

Three parts. reference is where the clock starts. interval is how much time you’re granting from that reference point. callback is what fires if the run hasn’t finished by reference + interval. Anchoring to a reference you choose is the real upgrade over sla: “done within two hours of the logical date” is now directly expressible, instead of the old “within N minutes of a task starting.”

The DeadlineReference variants are the vocabulary worth knowing:

  • DAGRUN_LOGICAL_DATE — the clock starts at the run’s logical date, the interval it represents. Best when the deadline is about the data being fresh: “yesterday’s sales should be in the mart within two hours of midnight.”
  • DAGRUN_QUEUED_AT — the clock starts when the run was actually queued. Best when the deadline is about execution time regardless of how far behind the schedule is: “however late we started, this shouldn’t take more than 90 minutes to run.” During a backfill, where logical dates are historical, this is the reference that still means something.
  • A fixed reference — a wall-clock deadline like “must be done by 6am,” independent of when the run started, for the report someone opens at the start of the business day.

The callback wrapper is the other choice, and it maps onto the same machinery you met with deferrable operators. SyncCallback runs the function synchronously in the scheduler context — fine for something instant. AsyncCallback hands the callback to the triggerer to run as an async coroutine, which is what you want when the notification itself does I/O (an HTTP POST to Slack or PagerDuty), because it keeps that I/O off the scheduler’s hot loop and on the process built to wait on async work. The same subprocess-vs-triggerer distinction that makes deferrable sensors cheap makes async deadline callbacks cheap: slow, waiting work belongs on the triggerer, not the scheduler. Import both, plus DeadlineAlert and DeadlineReference, from airflow.sdk — the AIP-86 surface lives in the Task SDK alongside dag and task.

Concurrency: three ceilings, three different jobs

The bookshop’s orders live in the store’s production Postgres, and that database has a day job serving the actual store. If a backfill spins up thirty task instances that all open connections and run extracts at once, you’ve turned a data pipeline into a denial-of-service attack on the thing paying the bills. Airflow has three distinct concurrency ceilings, and the reason people get bitten is that they’re solving different problems and it’s easy to reach for the wrong one.

[core] parallelism is the global ceiling. It’s a deployment-wide setting — the absolute maximum number of task instances running across every DAG at once. This is the number that protects your workers and your metadata database from the sum of everything you run. It’s an infrastructure knob, set in config, not something you tune per pipeline.

max_active_tasks_per_dag (set per-DAG, or as the [core] default) caps how many tasks run simultaneously within a single DAG run. A DAG that fans out into two hundred mapped tasks would happily try to run all two hundred; this is what stops one wide DAG from eating the whole global budget and starving its neighbours. It’s about fairness between DAGs.

Pools cap concurrency across DAGs by resource, which is the one that actually protects the fragile source. Define a source_db pool with four slots in the UI, assign every task that touches that database to it — from any DAG — and Airflow will never run more than four of them at a time no matter how many runs across how many pipelines are in flight:

@task(pool="source_db", pool_slots=1)
def extract_orders(data_interval_start=None, data_interval_end=None):
    ...

The distinction between the three: parallelism protects the cluster, max_active_tasks_per_dag enforces fairness between DAGs, and pools protect a shared resource regardless of which DAG reaches for it. Only the pool travels with the resource, which is why it’s the right tool for “don’t melt the source database.”

pool_slots is the lever most people never touch and occasionally need: a task can claim more than one slot. A cheap metadata query takes pool_slots=1; a heavyweight full-table extract that hammers the source as hard as four normal queries should take pool_slots=4, so it accounts for its real cost against the pool’s capacity. It lets you size a pool by load rather than by task count, which is far more honest when your tasks aren’t uniform.

When more work is queued than a pool can run, priority_weight decides who goes first. Higher weight wins the next free slot. By default a task’s effective priority is rolled up from its downstream chain via weight_rule="downstream" (the default) — a task with a lot of work depending on it outranks a dead-end. Set weight_rule="upstream" to invert that, or weight_rule="absolute" to make priority_weight mean exactly what it says with no rollup, which is the predictable choice when you want a specific critical extract to always jump the queue:

@task(pool="source_db", priority_weight=10, weight_rule="absolute")
def extract_todays_orders(...):
    ...  # always grabs the next free source_db slot ahead of backfill work

That’s the combination that keeps a fragile source both safe and responsive: a pool bounds the total load, pool_slots accounts for uneven task cost, and priority_weight makes sure the daily run isn’t stuck behind a three-month backfill for the same four slots.

Observability: know what happened without guessing

Alerts tell you something broke. Observability tells you what, and ideally warns you before it does. Three surfaces matter.

Logs are the first stop. Every task instance streams its stdout and stderr to the log backend, reachable from the UI or straight from a callback via ti.log_url. In production, don’t leave logs on the workers’ local disk — a Kubernetes pod dies and takes its logs with it, right when you need them. Configure remote logging to S3, GCS, or Elasticsearch so logs outlive the worker that wrote them:

AIRFLOW__LOGGING__REMOTE_LOGGING=True
AIRFLOW__LOGGING__REMOTE_BASE_LOG_FOLDER=s3://bookshop-airflow-logs/
AIRFLOW__LOGGING__REMOTE_LOG_CONN_ID=aws_logs

The UI transparently reads from the remote store, so the debugging experience is unchanged — the logs are just somewhere durable. Elasticsearch additionally makes them searchable across every run, which is the difference between “grep one task’s log” and “find every task that hit this error last week.”

The health endpoint is what your uptime monitor should be hitting so you learn the scheduler died from a page, not from a pile of tasks stuck in queued. The api-server exposes /api/v2/monitor/health, returning the status of the core components:

{
  "metadatabase": { "status": "healthy" },
  "scheduler": { "status": "healthy", "latest_scheduler_heartbeat": "2026-05-25T04:12:07Z" },
  "triggerer": { "status": "healthy", "latest_triggerer_heartbeat": "2026-05-25T04:12:06Z" }
}

Point your monitor at it and alarm on anything that isn’t healthy. The field that actually catches the sneaky failures is latest_scheduler_heartbeat. A scheduler can be up — the process is running, the endpoint responds — while its scheduling loop is wedged and no new tasks are being queued. The heartbeat going stale is the earliest signal of that, well before the symptom (nothing running) becomes obvious. The equivalent StatsD/OpenTelemetry metric is scheduler_heartbeat; alarm when it stops incrementing, and you catch a stuck scheduler in a minute instead of when someone notices the mart is empty.

Metrics are how you catch the slow degradation a single failure won’t show. Airflow emits scheduler and task metrics over StatsD, and 3.x speaks OpenTelemetry, so task durations, queue depths, and pool utilization land in the same tracing stack as the rest of your services instead of a bespoke dashboard. The ones worth a chart: task duration trending up (a source getting slower), pool slots pinned at capacity (you’ve under-provisioned and everything’s queuing), and the scheduler heartbeat above. A failure pages you; a metric lets you fix the thing before it fails.

Secrets don’t belong in Variables

The last production habit is the least glamorous: stop putting credentials in Airflow Variables and Connections as plaintext. Anyone with UI access can read them, and they end up in database backups. Point Airflow at a secrets backend instead via AIRFLOW__SECRETS__BACKEND — an environment-variable backend for simple setups, or AWS/GCP/Vault Secrets Manager for the real thing. You configure the backend once, keep referencing connections and variables by name in your DAGs, and Airflow resolves them from the vault at runtime. The DAG code doesn’t change; the secret just stops living in the metadata database.

Final thoughts

None of these are Airflow features so much as production disciplines that Airflow happens to make cheap. Idempotency is the one that pays for itself first — the day a task fails halfway and you retry it without a second thought, instead of manually diffing tables to figure out what already ran, you’ll understand why it’s the pattern everything else rests on. Retries lean on it, backfills lean on it, and the whole alerting stack exists to tell you the rare times it wasn’t enough. Wire the deadlines and notifiers so the pipeline tells you when it’s late, cap the pools so it can’t hurt the systems it reads from, and point a monitor at that heartbeat so a dead scheduler is a page and not a mystery. A pipeline you can retry without thinking, that shouts when it slips, and that can’t melt its source, is a pipeline you can sleep through.

Next: Testing DAGs and Wiring Up CI

Comments