Task Lifecycle: Intermediate

Before the scheduler moves a task to scheduled, it reads the task's trigger rule against the final states of every task directly upstream. The default rule is all_success: each upstream must have ended in success. Under it, 1 failed upstream puts the task in upstream_failed and 1 skipped upstream puts it in skipped, so a single failure near the top of a DAG paints the column below it orange without any of those tasks starting. The other rules change what counts as enough, and some of them stop waiting for the remaining upstream the moment their condition is met.

merge_counts keeps the default rule and needs both feeds green. page_oncall carries one_failed, so the scheduler evaluates it the instant either feed fails, while the other feed may still be pulling. drop_staging carries all_done, which waits for both feeds to reach a final state of any kind and then runs. A trigger rule is set per task, as a keyword on @task or on any operator, and it looks only 1 level up: publish_report reads the state of merge_counts and never sees the feeds behind it.

A twelfth rule exists for a task type covered later in the track; the 11 above cover every ordinary task.

Play the run for the 1 November window, which appears at 00:00 on 2 November. Both feeds start at 00:01. pull_store_counts raises at 00:06 and, with 0 retries, goes to failed. The scheduler now evaluates the 3 tasks behind the feeds. page_oncall sees 1 failed upstream and starts at 00:06 while the warehouse feed is still running. drop_staging waits, because all_done needs every upstream final, and starts at 00:09 when pull_warehouse_counts succeeds. merge_counts sees a failure under all_success and is set to upstream_failed without a worker ever picking it up, and publish_report follows at the same moment, because an upstream in upstream_failed counts as failed.

A branch task returns the task id of the path to take, or a list of ids when more than 1 path should run. Airflow then sets each task directly downstream of the branch that was not returned to skipped, and the skip spreads: a task behind a skipped task under all_success is skipped as well, all the way down the DAG. The decorator form is @task.branch; BranchPythonOperator does the same from a python_callable. The function can read anything in the run's context, and most branches check logical_date: the 1st of the month rebuilds, every other day appends.

On the run for 1 November the branch returns load_full, so load_delta is marked skipped and load_full runs. On the run for 15 November it returns load_delta and load_full is the one skipped. Either way publish sees 1 succeeded upstream and 1 skipped upstream. With the default rule in place of the one above, that's not enough: a skipped upstream leaves all_success unmet, and the scheduler sets publish to skipped. The orders table is loaded and never published, and the run badge is green, because its only leaf ended in skipped and a skipped leaf counts toward success.

none_failed_min_one_success is the rule a join wants: it runs when no upstream failed and at least 1 succeeded, so the skipped path is ignored and a failed path still blocks the publish. none_failed is close, with 1 difference that matters: it also runs when every upstream was skipped, which is what happens when a branch returns an empty list to say that nothing should load this run. all_done would run publish after load_full failed, announcing a table that was never rebuilt. one_success starts the join the moment 1 path succeeds, which is fine with 2 exclusive paths and wrong once the branch can return 2 ids, because the join would begin while the second path is still loading. When the chosen path fails, every one of these rules except all_done leaves publish in upstream_failed and the run in failed, which is the outcome you want.

The branch itself touches only its direct downstream. A task wired directly behind the branch whose id the branch never returns is skipped on every run. Say someone hangs notify_which_path off pick_load_path to report the choice: the branch returns load_full or load_delta, never notify_which_path, so that cell is grey every night and nothing is reported. Put that task behind the join instead, where the join's rule decides, or return its id from the branch along with the chosen path. The branch's own cell is success on every run, including a run where it chose nothing; the choice it made is in its log, and the grid shows it only through which cells turned grey.

A short-circuit task runs a function and reads its return value as a verdict. A truthy value lets the run continue. A falsy value skips every task downstream of the short-circuit, direct or not, whatever trigger rule each carries, and the short-circuit itself ends in success. The run finishes green with most of its column grey, which is the right shape for an hour in which there was nothing to do. The decorator form is @task.short_circuit; ShortCircuitOperator takes a python_callable and does the same.

On an hour with 0 landed clicks, clicks_arrived returns False and ends in success, and aggregate_clicks and refresh_dashboard are set to skipped by the short-circuit itself, before the scheduler evaluates any rule. That reach past the direct downstream is the parameter ignore_downstream_trigger_rule, which defaults to True. Set it to False and the short-circuit skips only aggregate_clicks; refresh_dashboard is then left to its own rule, so under all_success it is skipped as well, and under none_failed it runs on the empty hour. On an hour with clicks, the function returns True, nothing is skipped, and the verdict cost 1 short task.

An early return inside aggregate_clicks would give the same result for less code, and a worse grid. With the separate check, the empty hour is visible as grey cells, the 2 tasks behind it never take a worker, and a task wired behind them with ignore_downstream_trigger_rule=False and none_failed can still post the message that nothing landed.

LatestOnlyOperator answers 1 question when its task runs: is this run the newest one the schedule has produced? It takes the run's window, asks the timetable for the window after it, and checks whether the current clock time falls inside that next window. If it does, this run is the latest and the operator lets every downstream task proceed. If the clock has moved past that next window, a newer run exists or is due, and the operator skips its direct downstream; from there the skip cascades through trigger rules in the usual way, so a task behind the gate under none_failed would still run.

Say the DAG was paused for 3 days and unpaused at 00:10 on 2 November. Catch-up creates the 3 missing runs, and each reaches latest_only by 00:15. The operator works the same arithmetic in all 3, against the same clock:

build_price_history ran in all 3, so 3 days of history were written; refresh_price_cache ran once, from the run whose next window holds the clock. The 31 October run lost by 15 minutes: its next window closed at 2 November 00:00, and the check happened after that. A run you trigger by hand is exempt: the operator reads the run's type, sees manual, and lets everything proceed, because a person asked for that run. A run created by a backfill carries the backfill run type instead, so the check applies and old dates skip the refresh, which is the reason to place the operator ahead of anything that only makes sense once.

A callback is a Python function that Airflow calls with the task's context at a fixed moment in a try's life. A task has 5 of them: on_execute_callback as a try starts, on_success_callback when a try ends in success, on_retry_callback when a try fails and a retry remains, on_failure_callback when the last allowed try fails and the task reaches failed, and on_skipped_callback when the task's own code raises AirflowSkipException. Each accepts 1 function or a list of them, and each runs in the task's own process on the worker, right after the try ends. A task that never ran a process fires none of them.

context['task_instance'] is the instance that ran, carrying its dag_id, task_id and try_number, and context['exception'] is the exception the try raised, so the page reads payments_rollup.pull_ledger failed on try 3: ledger API 503. default_args hands every task in the DAG the same retries and the same callbacks, so a task added next month is paged without anyone remembering to wire it. The on_failure_callback passed to @dag itself is a different hook: it fires once when the run is marked failed, receives the run through context['dag_run'] rather than a task, and is executed by the dag processor rather than by a worker, so it fires even when the failure was an upstream_failed leaf that never ran.

The task made 3 tries and fired 3 callbacks, and the run fired 1. note_retry fires twice because 2 retries were allowed, page_oncall fires once when the third try exhausts them, and summarize_run fires when the run is marked failed. rollup was set to upstream_failed by the scheduler and, since no try of it ever ran, fires nothing, which is the behavior you want: 1 page for the root cause rather than 1 per orange cell. A task the scheduler skips through a trigger rule or a branch fires nothing for the same reason.

A callback is the right place for a side effect that needs the failure's details: a page with the task id, a message in a chat channel, a row in an incidents table. It is the wrong place for work the pipeline depends on, because an exception inside the callback is written to the task's log and otherwise ignored, so the task's state stays where it was and nothing downstream notices. A cleanup that must happen belongs in an all_done task, where its own cell shows whether it ran.

A task that never returns stays in running. Its process holds the worker it was given, it keeps reporting that it is alive, and the scheduler has no reason to interfere: as far as Airflow can see, a 9-hour try is a long try. The run stays in running with it, and the log stops after the last line printed. execution_timeout on the task and dagrun_timeout on the DAG put a clock on that.

execution_timeout is a timedelta measured per try from the moment the try starts running. When it elapses, the task runner raises AirflowTaskTimeout inside the task's process, and the try ends as a failure. Retries and callbacks treat it like any other failure: with retries=1, the first timed-out try goes to up_for_retry and on_retry_callback fires if one is set; after retry_delay a second try starts with its own fresh 15 minutes; if that one times out as well, the task reaches failed and on_failure_callback fires. write_sessions becomes upstream_failed and the run is marked failed at 01:32, 32 minutes after it began, rather than never.

dagrun_timeout is measured from the run's start, the moment it moved to running, which for a catch-up run is long after its logical date. When a run has been running longer than that, the scheduler marks it failed, sets every unfinished task in it to skipped, and fires the DAG's on_failure_callback; a task that was mid-try sees its state change on its next heartbeat and its process is stopped. The parameter is documented as enforced for scheduled runs, so a run you trigger by hand keeps going. In hourly_sessions the 50-minute limit sits behind the task limits as a backstop: the worst case for pull_sessions alone is 32 minutes, write_sessions adds at most 10, so a run that meets the backstop has something wrong that the task limits did not cover.