Sensors and Modes

A sensor is an operator whose poke() asks "is the condition true yet?" every poke_interval seconds until it is, or until timeout (default a week) fails the task. In poke mode the task sleeps between checks and holds a worker slot throughout. In reschedule mode each check is a short attempt: a False answer ends it as up_for_reschedule, frees the slot, and the scheduler starts it again later. booknest_wait runs FileSensor three ways against the same missing export file:

EXPORT = "incoming/orders_{{ ds }}.jsonl"    # relative to the connection's path
...
    wait = dict(filepath=EXPORT, fs_conn_id="booknest_landing", poke_interval=15, timeout=600)
    FileSensor(task_id="wait_poke", mode="poke", **wait)              # holds a slot
    FileSensor(task_id="wait_reschedule", mode="reschedule", **wait)  # frees it between pokes
    FileSensor(task_id="wait_deferrable", deferrable=True, **wait)    # triggerer waits

A first try failed all three: the default fs_default connection does not exist, because airflow db migrate (the only initializer since Airflow 3 129 removed db init) creates no default connections. A minute into the second run, run7/w1.sh listed the states and the LocalExecutor's busy workers, then dropped the file:

 wait_deferrable | deferred          |          1
 wait_poke       | running           |          1
 wait_reschedule | up_for_reschedule |          1
   36     7  airflow worker -- LocalExecutor: 01a0f9da-c748-7809-b70e-bdad82ba775e
--- 23:45:24, file dropped
 wait_poke       | success |          1 | 23:44:23   | 23:45:33 | 70.4
 wait_reschedule | success |          1 | 23:45:39   | 23:45:39 |  0.1

Only wait_poke held a worker, for 70 seconds; wait_reschedule left four task_reschedule rows and its last attempt took 0.1 seconds. Eight poking sensors would stall this stack's eight slots, so poke only for waits of seconds and reschedule (or defer) for longer ones.