Airflow Deferrable Operators: Sensors Without Worker Waste

In production, sensors are the quiet budget leak nobody watches. A FileSensor waiting for a 40-minute ETL to drop a file doesn't just wait — it holds a full worker slot for the entire 40 minutes, a slot a real task could be using. Multiply that by 50 sensors and you've effectively shrunk your cluster by 50 workers you're paying for but not using.

The old fixes were workarounds. mode="reschedule" freed the slot but hammered the scheduler with re-enqueue events. Airflow's real answer is deferrable operators: the task parks itself in the metadata DB, a lightweight triggerer process watches asynchronously, and a trigger event wakes the task exactly when the file lands.

The triggerer runs on asyncio, so one process can babysit tens of thousands of pending waits on a few hundred megabytes of RAM.

This is the upgrade that turns sensors from a tax on your cluster into a rounding error. If you're still poking in 2026, you're paying a worker salary for a task that does nothing.

1. The Poke Problem: Worker Slot Economics

An executor owns a fixed number of worker slots — parallelism on LocalExecutor, concurrency on Celery, pods on KubernetesExecutor. Every running task occupies one slot for as long as it runs. The core economics lesson: a slot is a scarce resource, and a waiting task is still a running task.

A poke-mode sensor is the worst citizen in that economy. It occupies a slot, wakes up every poke_interval seconds to check a condition, sleeps, and repeats — for the entire timeout. If the wait is 40 minutes and the file lands at minute 39, you've spent 39 minutes of worker time on a task that did one check's worth of work.

Meanwhile, real tasks sit in the queue. Queue depth grows, SLAs slip, and the knee-jerk fix is to add workers. That's buying capacity to cover a waste problem.

from datetime import datetime

from airflow import DAG
from airflow.providers.filesystem.sensors.filesystem import FileSensor

with DAG(
    dag_id="partner_feed_ingest",
    schedule="0 2 * * *",
    start_date=datetime(2026, 1, 1),
    catchup=False,
) as dag:
    wait_for_feed = FileSensor(
        task_id="wait_for_feed",
        filepath="/opt/data/partners/feed_{{ ds }}.csv",
        poke_interval=30,
        timeout=2400,  # 40 minutes of poking
        mode="poke",
    )

2. The Deferrable Model: Defer and Resume

Deferrable operators change the contract between a task and the executor. Instead of occupying a slot and poking, the task runs for a moment, then calls self.defer() — it serializes a trigger description into the metadata DB and releases the worker slot. The task instance enters the deferred state.

The trigger object knows how to wait. When the condition is met — the file appears, the API returns 200, the partition shows up — it fires a trigger event. The triggerer hands that event to the scheduler, which re-queues the task. The task resumes where it left off, exactly where execute_complete picks up.

Two things make this work: a defer call that's cheap (milliseconds, not minutes), and a resume path that's deterministic. The task doesn't re-run from the top — it resumes. That's the crucial difference from a retry, which restarts the whole task instance.

Two boundaries from the official docs. First, deferred tasks stop counting against pool slots by default — Airflow exempts them from pool limits unless you opt them back in per pool. Second, deferral is only available in class-based operators; a plain PythonOperator or a TaskFlow @task callable cannot defer, so a wait inside a callable has to be moved into an operator class or split out into a custom trigger-based operator.

from datetime import datetime

from airflow import DAG
from airflow.providers.filesystem.sensors.filesystem import FileSensor

with DAG(
    dag_id="partner_feed_ingest",
    schedule="0 2 * * *",
    start_date=datetime(2026, 1, 1),
    catchup=False,
) as dag:
    wait_for_feed = FileSensor(
        task_id="wait_for_feed",
        filepath="/opt/data/partners/feed_{{ ds }}.csv",
        poke_interval=30,
        timeout=2400,
        mode="reschedule",
        deferrable=True,
    )

3. The Triggerer Process and Its Scaling

The triggerer is the engine room of deferrable execution. It's a separate Airflow component — airflow triggerer — that runs trigger objects on an asyncio event loop. One triggerer process handles thousands of deferred triggers with a handful of threads, because waiting on file checks and API polls is I/O, not CPU.

Scaling the triggerer is boring in the best way. Default capacity is 1,000 concurrent triggers per process; run multiple triggerer replicas for headroom and high availability, exactly like you run multiple schedulers. If a triggerer dies, its triggers survive in the metadata DB and get picked up when it restarts.

The operational trap is the opposite of scaling: not running one at all. Teams deploy webserver, scheduler, and workers, forget the triggerer, and then wonder why deferred tasks never resume. The UI shows deferred; the triggerer log is empty; nothing moves.

HA is real, though: triggerers heartbeat to the metadata DB, and when one dies or is partitioned away, the surviving triggerers re-adopt its triggers after a ~30-second grace period — no manual restart needed. Airflow also de-duplicates events fired while a trigger briefly runs on two hosts during a partition, so double-execution is invisible to your operators.

4. Converting Sensors with deferrable=True

Most of the sensor catalog is deferrable-ready. FileSensor, ExternalTaskSensor, S3KeySensor, TimeSensor, and the cloud provider sensors accept deferrable=True and internally swap their poke loop for a trigger. In Airflow 3 the trigger machinery is the default path for new sensor types.

Conversion is a two-line change: set deferrable=True and keep a sane poke_interval (it becomes the trigger's poll rate) and timeout. Then verify the resume path, not just the defer path. Run the DAG with a file that arrives late and confirm the task completes after the file lands.

The trap: a sensor that defers but whose trigger can't serialize. Triggers must be reconstructible from constructor args — a trigger holding an open socket or a live client object fails serialization at defer time and the task errors immediately.

One config lever to know: [operators] default_deferrable flips the default for every operator and sensor that supports both modes, so deferrable=True stops being a per-task decision on big fleets.

5. Cost Math: 40 Minutes of Poke vs Milliseconds of Defer

Put numbers on it. A poke-mode sensor with a 40-minute timeout burns 2,400 worker-seconds of capacity, whether or not the file arrives. The same task deferrable holds the slot for the milliseconds it takes to serialize the trigger — call it 4 orders of magnitude cheaper, and the worker is free to run real work the whole wait.

Reschedule mode sits between the two: the slot is freed, but every poke re-enqueues the task through the scheduler and writes state transitions to the metadata DB. With a 30-second poke_interval, a 40-minute wait generates ~80 re-enqueues and scheduler wakeups. Deferrable mode generates exactly two: one defer, one resume.

That's why the triggerer wins at scale. Ten thousand deferred waits across a fleet cost one triggerer process ~10MB of RAM each thousand, versus ten thousand poke cycles hammering workers and DB. The economics invert completely as the wait count grows.

One caveat from the docs: at the opposite end of the spectrum — many short waits inside a single task — Airflow 3.2+'s async Python operators multiplex concurrent I/O within one worker slot and beat repeated defer/resume cycles, whose stop-and-restart overhead adds up when deferrals are frequent. Rule of thumb: few, long waits → defer; many, short waits → async operator.

# One 40-minute wait, per mode
# poke:      1 worker slot x 2,400 s = 2,400 worker-seconds
# reschedule: ~80 re-enqueues, 0 worker-seconds, 80 scheduler events
# deferrable: ~40 ms of triggerer time, 0 worker-seconds
#
# 50 such waits on an 8-slot cluster:
# poke -> 50 slots wanted, 8 exist: the other 42 queue for 40 min
# deferrable -> 0 slots wanted; all 8 workers stay busy on real work

6. Writing a Custom Deferrable Operator

When no provider trigger exists for your wait — a bespoke API, an internal system, a multi-condition check — you write your own. A custom deferrable operator is two classes: the operator that defers, and the trigger that waits. The trigger's run() is an async generator: check, sleep, check, and yield TriggerEvent the moment the condition is met.

The operator keeps `execute()` minimal: do any pre-check, then self.defer(trigger=..., method_name="execute_complete"). Airflow serializes the trigger, parks the task, and later calls execute_complete with the event payload on resume. Keep everything in serialize() — class path plus constructor kwargs — or deferral breaks in distributed setups.

Two production additions from the docs. on_kill() on the trigger is the safe place to cancel external work (a BigQuery job, a Databricks run): Airflow calls it when a user clears, fails, or marks the task done, but never on triggerer restart or redistribution — so nothing in-flight gets cancelled during a rolling deploy. And since 2.10, start_from_trigger (with start_trigger_args) and end_from_trigger let the task skip the worker entirely, deferring straight to the triggerer and even finishing there without renting a worker slot for the resume.

import asyncio

from airflow.sensors.base import BaseSensorOperator
from airflow.triggers.base import BaseTrigger, TriggerEvent


class PartitionReadyTrigger(BaseTrigger):
    def __init__(self, partition_path: str, poke_interval: float = 60.0):
        super().__init__()
        self.partition_path = partition_path
        self.poke_interval = poke_interval

    def serialize(self):
        return (
            "market_plugins.triggers.PartitionReadyTrigger",
            {"partition_path": self.partition_path, "poke_interval": self.poke_interval},
        )

    async def run(self):
        while True:
            if await self._partition_exists():
                yield TriggerEvent({"status": "ready"})
                return
            await asyncio.sleep(self.poke_interval)


class WaitForPartitionOperator(BaseSensorOperator):
    def __init__(self, partition_path: str, poke_interval: float = 60.0, **kwargs):
        super().__init__(**kwargs)
        self.partition_path = partition_path
        self.poke_interval = poke_interval

    def execute(self, context):
        self.defer(
            trigger=PartitionReadyTrigger(self.partition_path, self.poke_interval),
            method_name="execute_complete",
        )

    def execute_complete(self, context, event=None):
        if event is None:
            raise AirflowException("Partition never became ready")
        self.log.info("Partition ready: %s", self.partition_path)

7. Timeouts and Resume Semantics

The timeout clock doesn't pause for politeness. A task with timeout=2400 that defers at minute 0 and gets no trigger event fails at minute 40 — the wait was supposed to be inside the timeout, not outside it. Size task timeouts to the full business wait plus a buffer, or split long waits across multiple tasks.

Failure paths matter as much as success. A trigger that never fires eventually times the task out; the DAG fails, retries apply from the start of the task instance, and the retry defers again. A trigger that raises — bad credentials, deleted path — fails the task immediately. soft_fail=True converts timeout failures into skips for cases where a missing file is a legitimate outcome.

One more trap: clearing a task that's currently deferred. The old trigger keeps running until it fires or dies, so clear the DAG run, not just the task, and expect the trigger's garbage collection to lag by a poll interval.