Airflow downstream task skipped or upstream_failed when it should have run: trigger rules, branching and short-circuit

#Symptom

A task you expected to run never did. In the grid view it is pink (skipped) or orange (upstream_failed), and its log is empty because it never started. Common cases:

• The "join" task after a branch is always skipped.
• A "send report" or "cleanup" task is skipped every time an earlier check returns nothing.
• A "notify on failure" or "drop temp table" task shows upstream_failed exactly when you needed it.
• Everything after a ShortCircuitOperator is skipped, including tasks in a different part of the DAG.

None of these are scheduler bugs. Each task has a trigger rule that decides, from the states of its direct upstream tasks, whether it runs. The default is all_success, and skipped is not success.

#Checks

#1. Read the states of every task in the run

Find the run ID, then list the task states:

airflow dags list-runs <dag_id> | head -20     # Airflow 3.x
airflow dags list-runs -d <dag_id> | head -20  # Airflow 2.x
airflow tasks states-for-dag-run <dag_id> <run_id>

Work backwards from the task that did not run. Look only at its direct upstream tasks; trigger rules never look further back.

#2. Check the task's trigger rule

In the UI, open the task instance and check Task Instance Details (field trigger_rule). In code, search for it:

grep -n "trigger_rule" dags/<your_dag>.py

No match means the task uses all_success.

#3. Check for branching and short-circuit operators upstream

Look for BranchPythonOperator, @task.branch, ShortCircuitOperator, @task.short_circuit, and any code that raises AirflowSkipException. Any of these can create skipped states that then flow downstream.

#What each state means

#The trigger rules you actually need

Set the rule with a string so it works the same across versions:

Warning

all_done and one_failed tasks change how the DAG run state is computed. If the final leaf task is an all_done cleanup that succeeds, the run can be marked success even though real work failed, and failure alerts on the DAG run never fire. Keep a real-work task as a leaf, or use setup/teardown tasks (Airflow 2.7+), which are excluded from the run's outcome by default.

#Fixes

#Join after a branch is always skipped

The branch callable returns one path; the other path is skipped. The join task under all_success sees a skipped parent and skips itself.

from airflow.sdk import DAG, task          # Airflow 3.x
# from airflow.decorators import task      # Airflow 2.x
# from airflow import DAG
import pendulum

with DAG("branch_join_example", start_date=pendulum.datetime(2026, 11, 1, tz="UTC"),
         schedule="@daily", catchup=False):

    @task.branch
    def choose(**context):
        return "full_load" if context["logical_date"].day == 1 else "incremental_load"

    @task
    def full_load(): ...

    @task
    def incremental_load(): ...

    @task(trigger_rule="none_failed_min_one_success")
    def publish(): ...

    choose() >> [full_load(), incremental_load()] >> publish()

none_failed_min_one_success runs publish when one path succeeded and the other was skipped, but still stops it if the chosen path failed.

Tip

Don't use one_success for a join. It fires the moment the first parent succeeds, which is fine with exactly one path taken, but it also runs publish when another parent failed.

#Short-circuit skipped more than you wanted

By default, a short-circuit that returns a falsy value skips all downstream tasks, ignoring their trigger rules. To skip only the direct downstream tasks and let later tasks apply their own trigger rules, set ignore_downstream_trigger_rules=False:

@task.short_circuit(ignore_downstream_trigger_rules=False)
def has_new_files() -> bool:
    return len(list_new_files()) > 0

@task
def load(): ...

@task(trigger_rule="none_failed")
def update_audit_table(): ...    # runs even when there were no files

has_new_files() >> load() >> update_audit_table()

The classic operator takes the same parameter: ShortCircuitOperator(task_id=..., python_callable=..., ignore_downstream_trigger_rules=False). Import paths vary by version: in Airflow 3 it lives in the standard provider (airflow.providers.standard.operators.python), in Airflow 2 in airflow.operators.python.

#Cleanup task shows upstream_failed

Give the cleanup all_done so it runs whatever happened before it:

@task(trigger_rule="all_done")
def drop_staging_table(): ...

On Airflow 2.7+ a teardown task is cleaner. It runs after its setup and the work between them, and its own result does not decide the DAG run state:

create = create_cluster()
work = run_job()
delete = delete_cluster()
create >> work >> delete.as_teardown(setups=create)

#Failure notification never fires

A notify task under all_success can only run when nothing failed. Use one_failed, and wire it to the tasks you want to watch:

@task(trigger_rule="one_failed")
def page_on_call(): ...

[extract(), transform(), load()] >> page_on_call()

For most teams, on_failure_callback on the DAG or task is more reliable than a notify task, because it fires without changing the graph. See the failure alerts checklist for callbacks and notifiers.

#Re-running after the fix

Changing a trigger rule does not re-evaluate tasks already marked skipped or upstream_failed. Clear them to rerun with the new rule:

# Clear one task and everything downstream of it for a date range
airflow tasks clear <dag_id> -t "^publish$" -s 2026-11-10 -e 2026-11-13 --downstream --yes

Caution

airflow tasks clear without -t clears every task in every matching run, and those tasks run again. On non-idempotent loads that means duplicate rows. Always pass a task regex and a tight date range, and drop --yes the first time so you see the list before confirming.

#Test the logic before deploying

Run the DAG once locally, with no scheduler, to see the resulting states:

airflow dags test branch_join_example 2026-11-13

The output lists each task's final state, so you can check that the join ran and the unused branch was skipped.

#Prevent it next time

• Any task with more than one parent, or below a branch or short-circuit, needs a trigger rule chosen on purpose. Make that a code review check.
• Use none_failed_min_one_success for joins after branches, all_done or teardown tasks for cleanup, and callbacks for alerts.
• Don't make an all_done task the only leaf of a DAG, or failed runs will show as successful.
• Add a unit test that loads the DagBag and checks dag.get_task("publish").trigger_rule for your critical join tasks.
• Run airflow dags test in CI for DAGs with branching, so a skipped join shows up before production.