Dependencies, Blocked Tasks, and Sensors
Trace why a task did not run by walking the scheduler's dependency edges to whatever blocked it, and read sensor waits as first-class tasks that can time out.
The graph is a table of edges
A DAG looks like a picture and is stored as rows. When the scheduler serializes your pipeline it writes one row per edge: a dag_id, an upstream task, and a downstream task. That is the entire graph. There is no separate "shape" object, which is exactly why every graph question about a pipeline is answerable in SQL.
Stage 1 of 4: Two roots, both green. extract_orders is state success, and the sensor wait_for_payments is state success because its partition landed.
Failure propagates, it does not multiply
A task runs when every one of its upstreams succeeded. When one upstream fails, the scheduler does not attempt the downstream at all: it marks it upstream_failed and moves on, and the same happens to that task's downstreams, all the way to the leaves. The set of tasks reachable from the failure is its blast radius, and that is what your stakeholders actually care about. "clean_orders failed" means nothing to the finance team. "the CRM export and the finance metrics for 2026-03-14 are stale" means everything.
One hop is a self join: join the edge table to the task instances twice, once for each end of the edge, and keep the edges whose upstream end failed. The full closure is a recursive CTE over the same edge table, which is the same technique Level 4 used on an org chart, pointed at operational metadata instead.
Sensors are tasks that wait
A sensor is an ordinary task whose job is to wait for something outside the graph: a partition landing in the lake, a file arriving in a bucket, an upstream table's row count crossing a threshold. It pokes on an interval, and it has a timeout. If the thing never shows up, the sensor gives up and fails, and everything downstream of it goes upstream_failed exactly like any other failure.
That timeout is a design decision, not a detail. Set it too short and a slow but healthy upstream pages you. Set it longer than the schedule interval and yesterday's sensor is still holding a worker slot when today's run starts, which is how a pool starves and every DAG on the cluster stalls behind one waiting task.
Common mistake: paging on the upstream_failed count. Those tasks did not break, they were never attempted, and they will clear themselves the moment the real failure is fixed and the run is cleared. Alert on failed, report on upstream_failed.
Interview nuance: "this task failed, what is stale?" is a graph question, and answering it with a recursive walk of the dependency edges is the answer that separates a candidate who has operated a pipeline from one who has only drawn one. Say blast radius out loud, and name the two ways a task can fail to run: it failed, or something it waited on did.
On a real platform this differs. Airflow 3 adds event-driven scheduling and Assets, which invert the framing: instead of "run task B after task A", you declare "this asset is produced from those assets" and the scheduler derives the edges. Dagster made that the core model from the start. The stored shape is still a graph, and this query still works; what changes is who wrote the edges.
CREATE TABLE task_dependencies (
dag_id TEXT,
upstream_task_id TEXT,
downstream_task_id TEXT -- one row per edge; this is the whole graph, nothing else is stored
);
INSERT INTO task_dependencies (dag_id, upstream_task_id, downstream_task_id) VALUES
('daily_sales', 'extract_orders', 'clean_orders'),
('daily_sales', 'clean_orders', 'load_facts'),
('daily_sales', 'clean_orders', 'archive_orders'),
('daily_sales', 'wait_for_payments', 'load_facts'),
('daily_sales', 'load_facts', 'publish_metrics'),
('daily_sales', 'load_facts', 'export_crm'),
('daily_sales', 'load_facts', 'notify_sales'),
('daily_sales', 'publish_metrics', 'notify_finance'),
('daily_sales', 'export_crm', 'notify_sales'),
('hourly_events', 'wait_for_partition', 'ingest_events'),
('hourly_events', 'ingest_events', 'rollup_events'),
('hourly_events', 'rollup_events', 'refresh_cache'),
('weekly_digest', 'build_digest', 'send_digest');
CREATE TABLE task_instances (
run_id TEXT,
dag_id TEXT, -- a real task_instance is keyed by (dag_id, run_id, task_id)
task_id TEXT,
state TEXT,
try_number INTEGER
);
INSERT INTO task_instances (run_id, dag_id, task_id, state, try_number) VALUES
('ds_2026_04_01', 'daily_sales', 'extract_orders', 'success', 1),
('ds_2026_04_01', 'daily_sales', 'wait_for_payments', 'success', 1),
('ds_2026_04_01', 'daily_sales', 'clean_orders', 'success', 1),
('ds_2026_04_01', 'daily_sales', 'load_facts', 'success', 1),
('ds_2026_04_01', 'daily_sales', 'publish_metrics', 'success', 1),
('ds_2026_04_01', 'daily_sales', 'export_crm', 'success', 1),
('ds_2026_04_01', 'daily_sales', 'notify_finance', 'success', 1),
('ds_2026_04_01', 'daily_sales', 'notify_sales', 'success', 1),
('ds_2026_04_02', 'daily_sales', 'extract_orders', 'success', 1),
('ds_2026_04_02', 'daily_sales', 'wait_for_payments', 'success', 1),
('ds_2026_04_02', 'daily_sales', 'clean_orders', 'failed', 2),
('ds_2026_04_02', 'daily_sales', 'load_facts', 'upstream_failed', 1),
('ds_2026_04_02', 'daily_sales', 'publish_metrics', 'upstream_failed', 1),
('ds_2026_04_02', 'daily_sales', 'export_crm', 'upstream_failed', 1),
('ds_2026_04_02', 'daily_sales', 'notify_finance', 'upstream_failed', 1),
('ds_2026_04_02', 'daily_sales', 'notify_sales', 'upstream_failed', 1),
('he_2026_04_02_07', 'hourly_events', 'wait_for_partition', 'failed', 1),
('he_2026_04_02_07', 'hourly_events', 'ingest_events', 'upstream_failed', 1),
('he_2026_04_02_07', 'hourly_events', 'rollup_events', 'upstream_failed', 1),
('he_2026_04_02_07', 'hourly_events', 'refresh_cache', 'upstream_failed', 1),
('wd_2026_03_30', 'weekly_digest', 'build_digest', 'failed', 2),
('wd_2026_03_30', 'weekly_digest', 'send_digest', 'upstream_failed', 1);-- One hop of propagation: every failed task and what it directly froze.
SELECT up.run_id,
up.task_id AS failed_task,
dn.task_id AS blocked_task,
dn.state AS blocked_state
FROM task_instances up
JOIN task_dependencies td
ON td.dag_id = up.dag_id AND td.upstream_task_id = up.task_id
JOIN task_instances dn
ON dn.dag_id = up.dag_id AND dn.run_id = up.run_id AND dn.task_id = td.downstream_task_id
WHERE up.state = 'failed'
ORDER BY up.run_id, blocked_task;Apply
Your turn
The task this lesson builds to.
Write a query that returns every task instance blocked by a directly failed upstream, as (task_id, upstream_task_id, upstream_state), ordered by task_id, over task_dependencies(dag_id, upstream_task_id, downstream_task_id) and task_instances(run_id, dag_id, task_id, state, try_number).
Join the edge table to task_instances twice, once for the upstream end of the edge and once for the downstream end, matching on the same dag_id and the same run_id. Keep only the edges whose upstream instance is in state 'failed'. Alias the columns exactly.
3 hints and 1 automated check are waiting in the workspace.
Practice
Make it stick
A second problem on the same idea, plus 3 bonus drills.
Write a query that returns the full downstream closure of the failed clean_orders task in the daily_sales DAG, as (task_id, depth), nearest first, over task_dependencies.
depth is the number of edges from clean_orders to that task, so its direct downstreams are depth 1. A task reachable by more than one path appears once, at its shortest depth. Alias the columns exactly, ordered by depth ascending and then task_id.
3 hints and 1 automated check are waiting in the workspace.