Sensors and Deferrable Operators: Waiting Without Wasting

Half of data engineering is waiting for a file to land. Do it wrong and a sensor pins a worker for hours; do it right and the task steps aside until the world is ready.

The bookshop’s orders don’t arrive on your schedule. An upstream job drops a CSV into object storage sometime after midnight — usually by 00:20, occasionally at 02:00 when their source system is slow. Your load can’t start until that file exists. So you write a sensor: a task whose whole job is to wait until a condition is true, then let the pipeline proceed.

Sensors are one of Airflow’s oldest ideas and one of its easiest to misuse. The naive version works fine in a demo and quietly wrecks your cluster in production, because “wait for two hours” and “hold a worker slot for two hours” turn out to be the same sentence unless you say otherwise.

A sensor is a poll with a timeout

At heart a sensor repeatedly asks a yes/no question — is the file here? is the partition ready? — until the answer is yes or it gives up. In TaskFlow you write one with @task.sensor:

from airflow.sdk import task
from airflow.sdk.definitions.context import Context


@task.sensor(poke_interval=60, timeout=2 * 60 * 60, mode="reschedule")
def wait_for_orders_csv(hook, key):
    return hook.object_exists(key)

The function returns truthy when the condition is met. Between checks, the sensor sleeps for poke_interval and tries again, giving up after timeout. The interesting argument is mode, and it’s the one people get wrong.

In poke mode (the default), the sensor holds its worker slot the entire time it waits. It sits in the process, sleeps sixty seconds, wakes, checks, sleeps again. For a five-second wait that’s fine. For a two-hour wait it means one of your worker slots is doing nothing but napping — and if you have a handful of these across your DAGs, they can starve real work out of the pool. This is the classic Airflow footgun: a cluster that looks busy and produces nothing.

In reschedule mode, the sensor checks once, and if the answer is no it releases the slot and asks the scheduler to run it again after poke_interval. Between pokes it holds nothing. The trade is granularity — you can’t poll every second in reschedule mode without hammering the scheduler, so it suits waits measured in minutes, which most file-arrival waits are. For the bookshop CSV, reschedule mode with a sixty-second interval is a reasonable default. But it has a caveat sharp enough to deserve its own section, so hold that thought.

Timeout is a policy, not just a number

Before we get to modes, notice how much behavior hangs off what happens when a sensor gives up. timeout is the wall-clock ceiling — measured from the task’s first poke, not from DAG start — and when it’s crossed the sensor raises AirflowSensorTimeout, which by default fails the task. A failed sensor fails the DAG run, and everything downstream is blocked. Sometimes that’s exactly right: if the nightly orders file never lands, you want a red run and a page.

But often a missing input isn’t a failure, it’s an absence — the upstream job didn’t run today because it was a holiday, and there’s nothing to load. For that you want soft_fail=True:

@task.sensor(
    poke_interval=60,
    timeout=90 * 60,
    mode="reschedule",
    soft_fail=True,
)
def wait_for_orders_csv(hook, key):
    return hook.object_exists(key)

With soft_fail, a timeout skips the task instead of failing it. The distinction propagates: a skipped sensor, under the default trigger rule, skips its downstream tasks rather than marking them failed. So soft_fail=True is how you say “if the file isn’t here, quietly do nothing today” versus “if the file isn’t here, wake me up.” Choose deliberately — a soft-failing sensor guarding a critical load can hide a genuine upstream outage behind a wall of green-with-skips.

Two more knobs shape the polling itself. exponential_backoff=True stops the sensor from polling on a fixed cadence and instead lengthens the gap between pokes over time — 60s, then longer, then longer still — which is kind to a rate-limited API you’re checking. Pair it with max_wait (a timedelta) to cap how large that gap is allowed to grow, so an all-night wait doesn’t back off to checking once an hour:

from datetime import timedelta

from airflow.providers.http.sensors.http import HttpSensor

wait_for_upstream_api = HttpSensor(
    task_id="wait_for_upstream_api",
    http_conn_id="orders_api",
    endpoint="v1/exports/{{ ds }}/status",
    response_check=lambda r: r.json().get("state") == "ready",
    poke_interval=30,
    exponential_backoff=True,
    max_wait=timedelta(minutes=5),
    timeout=2 * 60 * 60,
)

And silent_fail=True handles the case where the check itself errors — a transient connection reset while poking S3, say. Normally an exception inside a poke fails the task immediately; silent_fail logs it and treats that poke as “not yet,” so a flaky check doesn’t nuke a two-hour wait over one dropped packet. It’s the difference between “the condition is false” and “I couldn’t tell,” and you usually want the latter to be patient rather than fatal.

The reschedule-mode caveat nobody mentions

Reschedule mode sounds like a free win — release the slot, poll again later, no worker pinned. It is not free, and the way it isn’t free has starved more than one cluster.

When a reschedule-mode sensor releases its slot, the task instance does not leave the scheduler’s world. It enters the up_for_reschedule state, and it comes back every poke_interval to run one poke. Each of those pokes is a real task-instance execution: it counts against max_active_tis_per_dag, it draws from whatever pool the task is assigned to, and it’s subject to the scheduler’s parallelism limits exactly like any other task. Reschedule mode doesn’t make the sensor cost nothing — it makes it cost briefly, repeatedly instead of continuously.

That distinction bites in two ways. First, poke_interval has a floor you don’t set: the scheduler loop. The scheduler only notices that an up_for_reschedule task is due to run again when it comes around on its next loop. If your scheduler loop is running every few seconds under load, a poke_interval=5 is a fiction — you’ll get pokes spaced by the loop, not by your five seconds. Reschedule mode is honest only at minute-ish granularity; ask it for sub-minute polling and it silently rounds up to whatever the scheduler can manage.

Second, and worse, is the starvation math. Suppose you give your sensors a pool of 16 slots to keep them from crowding out real loads. Now imagine 20 DAGs, each with a reschedule-mode sensor waiting on a file, all firing at roughly the same time after midnight. Twenty sensors want to poke; the pool has 16 slots. Four sensors queue behind the pool, waiting for a slot just to ask a yes/no question. Meanwhile the sixteen that got in poke, find nothing, and reschedule — and on the next loop all twenty are contending again. You’ve built a system where tasks compete for the privilege of checking whether they can start, and if the pool is small relative to the fan-in of waiters, some of them get reschedule-starved: perpetually near the back of the queue, poking far less often than their poke_interval claims, occasionally not making it in before their timeout expires. The failure looks baffling — “my sensor timed out but the file was there for an hour” — and the cause is that it never got a slot to look.

The lesson isn’t “don’t use reschedule mode.” It’s that reschedule mode moves the cost from worker slots to scheduler churn and pool contention, and if you have many sensors waking at once, that new cost is real and needs a pool sized for the peak count of simultaneous waiters, not the average. Or — and this is where the chapter is heading — you sidestep the whole trade with deferrable operators, which don’t consume a task-instance slot to wait at all.

Deferrable is the real fix

Reschedule mode solves the pinned-slot problem by re-queuing the task over and over, and we just saw what that re-queuing costs. There’s a cleaner mechanism built for exactly this: deferrable operators, backed by triggers.

A deferrable task runs until it hits a point where it must wait, then defers — it raises out of its worker entirely, handing a small async trigger to a separate process called the triggerer. The triggerer runs hundreds or thousands of these triggers concurrently on an asyncio event loop, cheaply, because a trigger that’s waiting on a file is just an idle coroutine. When the trigger’s condition fires, it wakes the task back up on a worker to finish. During the wait, the deferred task occupies no worker slot, no pool slot, and no scheduler queue — it’s a coroutine in the triggerer costing almost nothing. That’s the crucial contrast with reschedule mode: a deferred task is not up_for_reschedule coming back to poke; it’s deferred, entirely off the scheduler’s execution path until the trigger says go.

Many built-in operators and sensors ship a deferrable path behind a deferrable=True flag:

from airflow.providers.amazon.aws.sensors.s3 import S3KeySensor

wait_for_orders = S3KeySensor(
    task_id="wait_for_orders",
    bucket_name="bookshop-dropbox",
    bucket_key="orders/{{ ds }}/orders.csv",
    deferrable=True,
)

Under the hood the task defers to a trigger that polls the bucket asynchronously and resumes your DAG the instant the orders CSV appears. The behavior your DAG sees is identical to a plain sensor; what changed is that a two-hour wait no longer costs you a worker or a pool slot or a scheduler loop’s worth of churn.

If you’d rather not repeat deferrable=True on every sensor in the project, set it once. The default_deferrable config ([operators] default_deferrable = True, or AIRFLOW__OPERATORS__DEFAULT_DEFERRABLE=True) flips the default for every operator that supports deferral, so a plain S3KeySensor(...) becomes deferrable without the flag. It’s a good project-wide default once you’ve confirmed the triggerer is running everywhere your DAGs run.

Which sensors actually honor deferrable=True? As of the current providers, the common ones do: S3KeySensor and the GCS object sensors, the SQL sensors (SqlSensor), HttpSensor, the time-based DateTimeSensor and TimeDeltaSensor, and ExternalTaskSensor. Some older or niche sensors — and, notably, any @task.sensor you write yourself — have no async path and ignore the flag; they only ever poke. When in doubt, check whether the sensor exposes a deferrable argument at all. If it doesn’t, poke versus reschedule is your only lever, and the previous section’s math applies.

There’s an older, more explicit form you’ll still see: sensors with an Async suffix, like DateTimeSensorAsync, which are hard-wired to defer with no flag to toggle:

from airflow.providers.standard.sensors.date_time import DateTimeSensorAsync

wait_until_load_window = DateTimeSensorAsync(
    task_id="wait_until_load_window",
    target_time="{{ logical_date.add(hours=1) }}",
)

The deferrable=True flag on the non-Async class is the newer, preferred spelling — same effect, one class instead of two.

Building your own deferrable sensor

Consuming deferrable=True is the easy 90%. The instructive part — and the thing that separates people who use Airflow from people who extend it — is writing a deferrable operator when no provider ships the wait you need. The bookshop’s upstream team, for instance, doesn’t drop a file; they flip a row in a “batch manifest” table to status = 'CLOSED' when the day’s orders are final. No off-the-shelf sensor knows about that table. So we build one.

A custom deferrable component has two halves that talk across the worker/triggerer boundary:

  1. A trigger: a BaseTrigger subclass with an async run() that does the actual non-blocking waiting and yields a TriggerEvent when the condition is met, plus a serialize() that lets the trigger be stored in the metadata DB and rehydrated in the triggerer.
  2. An operator (or sensor): whose execute() calls self.defer(...) to hand off the trigger, and whose execute_complete() is called when the trigger fires.

Here’s the trigger. Its whole job is to poll the manifest table asynchronously and resolve when the batch is closed:

import asyncio
from typing import Any, AsyncIterator

from airflow.triggers.base import BaseTrigger, TriggerEvent
from airflow.providers.postgres.hooks.postgres import PostgresHook


class BatchClosedTrigger(BaseTrigger):
    def __init__(self, conn_id: str, batch_date: str, poke_interval: float = 60.0):
        super().__init__()
        self.conn_id = conn_id
        self.batch_date = batch_date
        self.poke_interval = poke_interval

    def serialize(self) -> tuple[str, dict[str, Any]]:
        return (
            "bookshop.triggers.BatchClosedTrigger",
            {
                "conn_id": self.conn_id,
                "batch_date": self.batch_date,
                "poke_interval": self.poke_interval,
            },
        )

    async def run(self) -> AsyncIterator[TriggerEvent]:
        while True:
            status = await self._fetch_status()
            if status == "CLOSED":
                yield TriggerEvent({"status": "success", "batch_date": self.batch_date})
                return
            await asyncio.sleep(self.poke_interval)

    async def _fetch_status(self) -> str | None:
        hook = PostgresHook(postgres_conn_id=self.conn_id)
        sql = "SELECT status FROM batch_manifest WHERE batch_date = %s"
        # run the sync DB call in a thread so it doesn't block the event loop
        row = await asyncio.to_thread(hook.get_first, sql, parameters=(self.batch_date,))
        return row[0] if row else None

Three things in there are load-bearing. First, serialize() returns the fully-qualified path to the trigger class plus a plain-dict of constructor kwargs. When your operator defers, Airflow serializes the trigger by that tuple, writes it to the DB, and the triggerer reconstructs it by importing that path and calling __init__(**kwargs). That’s why every argument the trigger needs must be a JSON-serializable kwarg — no live hook object, no connection, just the conn_id string it can rebuild from.

Second, run() is an async generator. It loops, checks, awaits a sleep, and the moment the batch is closed it yields exactly one TriggerEvent carrying whatever payload you want to hand back to the task, then returns to end the coroutine. The event dict is arbitrary; put a status flag in it so execute_complete can distinguish success from a soft-failure signal.

Third — and this is the rule that trips everyone — the event loop must never block. The triggerer is running your run() alongside potentially thousands of others on a single asyncio loop. A synchronous hook.get_first(...) would block that loop, freezing every other trigger in the process until your DB call returns. That’s why the DB call is wrapped in asyncio.to_thread(...): it shoves the blocking psycopg call onto a thread-pool so the event loop stays free. Any real network or disk I/O in a trigger has to go through a genuinely async client or a thread offload. If you take one thing from this section: no synchronous I/O in run().

Now the sensor that uses it:

from typing import Any

from airflow.sdk.bases.sensor import BaseSensorOperator
from airflow.sdk.definitions.context import Context
from airflow.exceptions import AirflowException

from bookshop.triggers import BatchClosedTrigger


class BatchClosedSensor(BaseSensorOperator):
    def __init__(self, *, conn_id: str, batch_date: str, **kwargs):
        super().__init__(**kwargs)
        self.conn_id = conn_id
        self.batch_date = batch_date

    def execute(self, context: Context) -> None:
        self.defer(
            trigger=BatchClosedTrigger(
                conn_id=self.conn_id,
                batch_date=self.batch_date,
                poke_interval=self.poke_interval,
            ),
            method_name="execute_complete",
            timeout=self.execution_timeout,
        )

    def execute_complete(self, context: Context, event: dict[str, Any]) -> str:
        if event.get("status") != "success":
            raise AirflowException(f"Batch never closed: {event}")
        self.log.info("Batch %s closed; proceeding.", event["batch_date"])
        return event["batch_date"]

The flow reads cleanly once you see the handoff. execute() runs on a worker, immediately calls self.defer(...), and that call raises a special exception that unwinds the task off the worker — the worker slot is now free. Airflow persists the trigger, the triggerer picks it up and runs BatchClosedTrigger.run(), and when that coroutine yields its TriggerEvent, Airflow schedules the task back onto a worker and calls the method named in method_name — here execute_complete — passing the event payload. execute_complete runs the resolution logic: it inspects the event, and its return value becomes the task’s XCom, just as a normal execute return would. The timeout= on defer is the deferred-side deadline; if the trigger never fires within it, the task is failed with a timeout rather than hanging forever.

That last point is a subtlety worth naming. A classic sensor’s timeout produces AirflowSensorTimeout from the poking loop on the worker. A deferred task doesn’t poke on a worker, so its timeout is enforced differently — by the timeout you pass to self.defer, tracked while the trigger is parked in the triggerer. Same intent, different machinery, and the deferred version doesn’t burn a worker while the clock runs.

@task.sensor, revisited: hooks, keys, and PokeReturnValue

Back at the top we wrote a @task.sensor with hook and key parameters and glossed over where those come from. The decorated function is a poke: it’s called repeatedly, and Airflow doesn’t magically inject a hook — you supply the arguments the same way you’d feed any TaskFlow function. The cleanest way is a closure over the hook, or partial application:

from airflow.providers.amazon.aws.hooks.s3 import S3Hook


@task.sensor(poke_interval=60, timeout=2 * 60 * 60, mode="reschedule")
def wait_for_orders_csv(bucket: str, key: str):
    hook = S3Hook(aws_conn_id="bookshop_s3")
    return hook.check_for_key(key, bucket_name=bucket)


# in the DAG body:
wait_for_orders_csv(bucket="bookshop-dropbox", key="orders/{{ ds }}/orders.csv")

Building the hook inside the poke (rather than passing a live hook object in) matters for the same serialization reason as before, and it keeps the connection fresh across reschedules. If you want one sensor per key over a list of keys, @task.sensor supports .expand() just like any mapped task — wait_for_orders_csv.partial(bucket="bookshop-dropbox").expand(key=[...]) fans out a sensor per key, each waiting independently.

The genuinely useful feature hiding in @task.sensor is PokeReturnValue. A plain sensor returns truthy to mean “done” and nothing else — the downstream task learns that the condition was met but not what the poke saw. PokeReturnValue lets a sensor both signal completion and push an XCom in the same breath:

from airflow.sdk.bases.sensor import PokeReturnValue


@task.sensor(poke_interval=60, timeout=2 * 60 * 60, mode="reschedule")
def wait_for_manifest(bucket: str, prefix: str):
    hook = S3Hook(aws_conn_id="bookshop_s3")
    keys = hook.list_keys(bucket_name=bucket, prefix=prefix)
    if not keys:
        return PokeReturnValue(is_done=False)
    return PokeReturnValue(is_done=True, xcom_value=sorted(keys))

When is_done is True, the sensor completes and the xcom_value is pushed as the task’s return XCom, so the downstream load can pull the exact list of keys that landed without re-listing the bucket. When is_done is False, the sensor keeps waiting. It turns a sensor from a pure gate into a gate-that-hands-you-the-goods, which removes a redundant “now go find what I was waiting for” task downstream.

ExternalTaskSensor and cross-DAG waits

The single most common real-world sensor isn’t waiting on a file — it’s waiting on another DAG. The bookshop’s daily_load DAG must not build reports until the daily_ingest DAG has finished landing today’s data, and those are separate DAGs on the same schedule. ExternalTaskSensor is the canonical tool:

from airflow.providers.standard.sensors.external_task import ExternalTaskSensor

wait_for_ingest = ExternalTaskSensor(
    task_id="wait_for_ingest",
    external_dag_id="daily_ingest",
    external_task_id="finalize_load",
    allowed_states=["success"],
    failed_states=["failed", "upstream_failed"],
    poke_interval=120,
    timeout=3 * 60 * 60,
    deferrable=True,
)

Two details make or break it. First, alignment: by default ExternalTaskSensor looks for a run of the external DAG at the same logical date as the sensor’s own run. If the two DAGs share a schedule that’s automatic; if their schedules differ, you must supply execution_date_fn or execution_delta to point the sensor at the right upstream run, or it’ll wait forever for a run that doesn’t exist. Second, set failed_states — without it, the sensor treats an upstream failure as merely “not success yet” and keeps waiting until timeout, turning a fast upstream failure into a three-hour hang. Listing failed and upstream_failed in failed_states lets the sensor give up immediately when the thing it’s waiting for has clearly died.

For mode, ExternalTaskSensor is one of the sensors that ships deferrable=True, and cross-DAG waits are often long (a full upstream pipeline), so deferrable is usually the right call here — it’s precisely the “long, variable wait, possibly many at once” case the triggerer was built for. Reschedule mode works too and is fine when you don’t run the triggerer, but remember the pool math from earlier: a fleet of report DAGs all waiting on ingest via reschedule-mode ExternalTaskSensor is exactly the fan-in that starves a small sensor pool.

The prerequisite people forget

Everything deferrable in this chapter — deferrable=True, the Async sensors, your hand-built BatchClosedSensor — depends on the triggerer process actually running. In an Astro project astro dev start brings it up alongside the scheduler and workers, so locally it just works. On a cluster you manage yourself, confirm the triggerer is deployed before you scatter deferrable=True across your DAGs, or those tasks will defer into a void and never resume. The triggerer supports HA — you can run more than one, and triggers are distributed across them — so it isn’t a single point of failure, but it does have to be there. It’s the one piece of infrastructure that turns “wait cheaply” from a promise into a fact.

Final thoughts

A sensor is the rare Airflow component where the default setting is the dangerous one. poke mode reads as harmless — it’s just waiting — and that’s exactly why it slips into production and quietly eats a pool. Reschedule mode looks like the cure until you do the arithmetic and find you’ve traded a pinned worker for pool contention and scheduler churn that can starve the very sensors you were trying to make cheap. The mental model worth keeping is that waiting and occupying a resource are separable, and the whole progression — from poke to reschedule to deferrable, and finally to writing your own trigger with a non-blocking run() — is Airflow prying those two apart. Once the triggerer is running, “wait for the file” stops being a resource decision and goes back to being what it should have been all along: a line in your DAG that costs nothing until the world is ready.

Next: Dynamic Task Mapping and TaskGroups: Fan Out, Stay Tidy

Comments