What is sensor deadlock?

Sensors in poke mode occupy a worker slot for their whole wait. If enough sensors are waiting, they consume every slot in the pool and the tasks they are waiting for can never run — so the sensors wait forever. Reschedule mode and deferrable operators both prevent this.

What is the difference between poke and reschedule mode?

Poke keeps the task running and sleeps between checks, holding a worker slot the whole time. Reschedule marks the task up_for_reschedule between checks and releases the slot, at the cost of task startup overhead on each poke. Use reschedule when the interval is over about a minute.

Do deferrable operators need extra infrastructure?

They need the triggerer process running — airflow triggerer, which standalone starts for you. It runs an asyncio event loop that can watch thousands of conditions in one process, which is why deferred tasks cost almost nothing while waiting.

Can I set a timeout on a sensor?

Yes, and you should. timeout is the maximum wait in seconds before the sensor fails; without it a sensor can wait until the DAG run itself times out. Pair it with soft_fail=True if a missing input should skip downstream tasks rather than fail them.

Sensors and Deferrable Operators

A large share of orchestration is waiting — for a file, a partition, another DAG. How you wait decides whether your cluster stays usable.

A poke sensor

from datetime import datetime, timedelta
from airflow import DAG
from airflow.sensors.filesystem import FileSensor
from airflow.operators.bash import BashOperator

with DAG(
    dag_id="wait_poke",
    start_date=datetime(2026, 2, 1),
    schedule=None,
    catchup=False,
) as dag:

    wait = FileSensor(
        task_id="wait_for_file",
        filepath="/tmp/incoming/data.csv",
        fs_conn_id="fs_default",
        poke_interval=10,          # check every 10 seconds
        timeout=600,               # give up after 10 minutes
        mode="poke",               # holds a worker slot the whole time
    )

    process = BashOperator(
        task_id="process",
        bash_command="wc -l < /tmp/incoming/data.csv",
    )

    wait >> process

Run it with the file absent:

mkdir -p /tmp/incoming
airflow tasks test wait_poke wait_for_file 2026-02-14

[2026-02-14 13:02:11] {base.py:250} INFO - Poking for file /tmp/incoming/data.csv
[2026-02-14 13:02:11] {base.py:288} INFO - Sensor is not ready. Waiting 10 seconds.
[2026-02-14 13:02:21] {base.py:250} INFO - Poking for file /tmp/incoming/data.csv
[2026-02-14 13:02:21] {base.py:288} INFO - Sensor is not ready. Waiting 10 seconds.

Create the file in another terminal:

printf 'a,b\n1,2\n3,4\n' > /tmp/incoming/data.csv

[2026-02-14 13:02:31] {base.py:250} INFO - Poking for file /tmp/incoming/data.csv
[2026-02-14 13:02:31] {base.py:281} INFO - Success criteria met. Exiting.
[2026-02-14 13:02:31] {taskinstance.py:1400} INFO - Marking task as SUCCESS.

The slot problem

mode="poke" keeps the task running between checks. It sleeps, but it occupies a worker slot for the entire wait. With poke_interval=10 and a two-hour wait, that is one slot held for two hours to do 720 filesystem checks.

Scale it up and the failure mode appears:

with DAG(dag_id="deadlock_demo", start_date=datetime(2026, 2, 1),
         schedule=None, catchup=False, max_active_tasks=16) as dag:

    for i in range(16):
        FileSensor(
            task_id=f"wait_{i}",
            filepath=f"/tmp/incoming/part_{i}.csv",
            poke_interval=30,
            timeout=3600,
            mode="poke",
        ) >> BashOperator(task_id=f"load_{i}", bash_command=f"echo loading {i}")

airflow dags trigger deadlock_demo && sleep 20
airflow tasks states-for-dag-run deadlock_demo manual__2026-02-14T13:10:00+00:00

deadlock_demo | wait_0  | ... | running
deadlock_demo | wait_1  | ... | running
...
deadlock_demo | wait_15 | ... | running
deadlock_demo | load_0  | ... | None

All 16 slots consumed by sensors. If the thing being waited for is produced by another Airflow task, that task can never start — sensor deadlock. The pool is full of tasks waiting for work that needs the pool.

Reschedule mode

    wait = FileSensor(
        task_id="wait_for_file",
        filepath="/tmp/incoming/data.csv",
        poke_interval=60,
        timeout=3600,
        mode="reschedule",         # release the slot between checks
    )

[2026-02-14 13:20:02] {base.py:250} INFO - Poking for file /tmp/incoming/data.csv
[2026-02-14 13:20:02] {taskinstance.py:1206} INFO - Marking task as UP_FOR_RESCHEDULE.
[2026-02-14 13:21:02] {base.py:250} INFO - Poking for file /tmp/incoming/data.csv
[2026-02-14 13:21:02] {taskinstance.py:1206} INFO - Marking task as UP_FOR_RESCHEDULE.

UP_FOR_RESCHEDULE instead of staying running — the slot is free between checks. The cost is full task startup on each poke, so it is only worth it when poke_interval is comfortably longer than startup time, roughly a minute or more.

Deferrable operators

Better still: hand the waiting to a separate process that does nothing but wait.

from datetime import datetime, timedelta
from airflow import DAG
from airflow.sensors.date_time import DateTimeSensorAsync
from airflow.sensors.filesystem import FileSensor
from airflow.operators.bash import BashOperator

with DAG(
    dag_id="wait_deferred",
    start_date=datetime(2026, 2, 1),
    schedule=None,
    catchup=False,
) as dag:

    wait = FileSensor(
        task_id="wait_for_file",
        filepath="/tmp/incoming/data.csv",
        poke_interval=30,
        timeout=3600,
        deferrable=True,           # hand off to the triggerer
    )

    process = BashOperator(task_id="process", bash_command="wc -l < /tmp/incoming/data.csv")
    wait >> process

The triggerer must be running:

airflow triggerer

[2026-02-14 13:30:01] {triggerer_job_runner.py:107} INFO - Starting the triggerer
[2026-02-14 13:30:01] {triggerer_job_runner.py:404} INFO - 0 triggers currently running

Trigger the DAG:

[2026-02-14 13:31:04] {taskinstance.py:1520} INFO - Pausing task as DEFERRED.
  dag_id=wait_deferred, task_id=wait_for_file
[2026-02-14 13:31:04] {triggerer_job_runner.py:404} INFO - 1 triggers currently running

airflow tasks states-for-dag-run wait_deferred manual__2026-02-14T13:31:00+00:00

wait_deferred | wait_for_file | ... | deferred

State deferred, not running. The worker slot was released the moment the task deferred; the triggerer’s asyncio loop is watching the condition. One triggerer handles thousands of these in a single process, because each is a coroutine rather than a task slot.

When the file appears:

[2026-02-14 13:33:12] {triggerer_job_runner.py:404} INFO - 0 triggers currently running
[2026-02-14 13:33:13] {taskinstance.py:1157} INFO - Resuming after deferral
[2026-02-14 13:33:13] {taskinstance.py:1400} INFO - Marking task as SUCCESS.

The task resumes on a worker only to finish.

Comparing the three

Mode	Slot while waiting	Startup cost per check	Use when
`poke`	held	none	waits under a minute
`reschedule`	released	full task startup	waits of minutes to hours
`deferrable=True`	released	none	any wait, if a triggerer runs

Deferrable is the right default in Airflow 2.x. The only reason not to use it is an operator that does not support it yet.

Waiting on another DAG

from airflow.sensors.external_task import ExternalTaskSensor

    upstream_done = ExternalTaskSensor(
        task_id="wait_for_ingest",
        external_dag_id="ingest_raw",
        external_task_id="finalise",
        allowed_states=["success"],
        failed_states=["failed", "skipped"],
        execution_delta=timedelta(hours=1),   # upstream runs an hour earlier
        poke_interval=60,
        timeout=7200,
        mode="reschedule",
    )

execution_delta is the part people get wrong. The sensor looks for a run of the upstream DAG at this_logical_date - execution_delta. If the two DAGs have different schedules, an incorrect delta means the sensor waits for a run that will never exist:

[2026-02-14 13:40:02] {external_task.py:206} INFO - Poking for tasks ['finalise'] in dag ingest_raw
  on 2026-02-14T12:00:00+00:00 ...
[2026-02-14 14:40:02] {taskinstance.py:1938} ERROR - Sensor has timed out after 7200 seconds.

Prefer Datasets where you can — they express the dependency directly instead of inferring it from timestamps:

from airflow.datasets import Dataset

orders = Dataset("s3://lake/orders/")

# Producer DAG
@task(outlets=[orders])
def write_orders(): ...

# Consumer DAG — no sensor, no schedule guessing
with DAG(dag_id="consume", schedule=[orders], start_date=datetime(2026, 2, 1)) as dag:
    ...

[2026-02-14 13:45:11] {dataset_manager.py:52} INFO - Dataset s3://lake/orders/ updated,
  triggering 1 downstream DAG run

The consumer runs when the data is actually produced, whatever time that is.

Timeouts and soft fail

    wait = FileSensor(
        task_id="wait_optional",
        filepath="/tmp/incoming/optional.csv",
        poke_interval=30,
        timeout=300,
        soft_fail=True,            # skip instead of fail on timeout
        deferrable=True,
    )

[2026-02-14 13:52:41] {base.py:298} INFO - Sensor has timed out; soft_fail is set,
  marking task as SKIPPED.

wait_optional | skipped
downstream    | skipped

soft_fail turns a missing optional input into a skip rather than a page at 3am. Without it, the sensor fails and everything downstream is upstream_failed.

Always set timeout. A sensor with no timeout waits until the DAG run’s own timeout, which may be never.

Practice

1. Run 16 poke sensors with max_active_tasks=16 and try to start another task.

new_task | None

It never starts — every slot is held by a sensor. Switch them to deferrable=True and all 16 show deferred while the slots stay free. This is the concrete argument for deferrable operators.

2. Compare log volume for a one-hour wait in poke vs reschedule mode.

poke (10s interval):       360 poke lines in one task log
reschedule (60s interval):  60 poke lines across 60 task attempts

Reschedule creates a separate task instance per check, so the history is spread across attempts. Deferrable produces almost nothing — the triggerer logs once when it starts watching and once when it fires.

3. Set ExternalTaskSensor with the wrong execution_delta.

INFO - Poking for tasks ['finalise'] in dag ingest_raw on 2026-02-14T12:00:00+00:00 ...
ERROR - Sensor has timed out after 7200 seconds.

It waits for a run at a timestamp that does not exist. The log line naming the timestamp is the diagnostic — compare it to the upstream DAG’s actual run IDs. Datasets avoid the whole class of problem.

4. Use soft_fail=True and let a sensor time out.

wait_optional | skipped
downstream    | skipped

Skipped rather than failed, and the skip propagates. Note the DAG run itself is marked success — so a genuinely required input must not use soft_fail, or a silent no-op run looks like a healthy one.

Next: choosing between paths, and controlling when a task runs after a skip.