Proposal : Native Sequential and Fail-Fast Execution for Dynamic Mapped Tasks

[Ananyaas]

Summary

I would like to propose a feature for the Airflow 3.x roadmap to natively support Sequential and Fail-Fast execution within the Dynamic Task Mapping framework (.expand()).

As Airflow 2.x moves past its April 2026 EOL, defining core task orchestration boundaries for Airflow 3.x is highly relevant. Currently, Dynamic Task Mapping operates on a strict parallel MapReduce assumption where all mapped indices (map_index=0, 1, 2...) are independent. However, production workflows frequently require runtime-generated data chunks to execute chronologically, where an early failure must immediately halt the remaining sequence.

The Architectural Problem

While concurrency controls like max_active_tis_per_dag=1 can force serialized execution, they fail to create state-based dependencies between individual sibling indices:

1. No Cascading Failures: If map_index=0 fails, Airflow still triggers map_index=1 and map_index=2. In data loops where downstream steps assume the integrity of upstream modifications, this causes resource waste and data corruption.
2. Global Fail-Stop Limitations: The global fail_stop=True flag stops the entire DAG. This is too destructive when you only want to break execution for a specific mapped task array while letting independent branches finish normally.

Proposed Airflow 3.x API Design

I propose introducing two parameters to the execution mapper: sequential: bool = False and fail_fast: bool = False.

@task
def process_data_chunk(chunk_id):
    # Core processing logic
    pass

# Proposed configuration for Airflow 3.x
process_data_chunk.expand(
    chunk_id=runtime_generated_list, 
    sequential=True, 
    fail_fast=True
)

Strategic Technical Implementation

Because this modifies core scheduling assumptions, I want to ensure it integrates seamlessly with Airflow 3.x's scheduler architecture:

1. Gated Execution (airflow/models/dagrun.py):
   Inside _get_ready_tis(), if sequential=True is enabled, the scheduler gates the readiness of map_index = N until map_index = N-1 reaches a terminal SUCCESS state.

2. State Cascading (airflow/ti_deps/deps/trigger_rule_dep.py):
   If fail_fast=True is active and map_index = N switches to FAILED, the scheduling loop automatically cascades all subsequent un-run sibling indices (N+1 to max index) directly to UPSTREAM_FAILED.

Seeking Community Feedback

I am prepared to lead this implementation and draft an Airflow Improvement Proposal (AIP) if required. Before taking this to the devlist, I would love initial feedback:

• Does this feature align with the scheduling goals of Airflow 3.x?
• Are there edge cases regarding task retries or clearing mapped task states that we should plan for?

[raphaelauv]

#62922