Keep queued asset events when a Dag is at max_active_runs

[Vamsi-klu]

Keep queued asset events when a Dag is at max_active_runs

closes: #56050

This is leftover batch consume after the cap lifts, not one run per event.

What I did

Reporter timeline: event 1 → run 1; events 2, 3, 4 while at max_active_runs=1 stay in ADRQ; after run 1 finishes, run 2 consumes {2,3,4}; event 5 waits and becomes its own run.

dags_needing_dagruns no longer ANDs the cached exceeds_max_non_backfill flag onto asset leftovers. Asset create sets that flag after create_dagrun. Live QUEUED+RUNNING exclusion still defers a Dag at cap and does not delete ADRQ.

Why I did

Events 3 and 4 sat in the queue until a later event unstuck them. That used to be cutoff-and-wipe. The wipe is already gone. The leftover hole is the stale cache: asset create never refreshed exceeds_max_non_backfill, so a Dag with queued events stayed hidden after the live cap lifted.

How I did

The cached flag is timetable-only:

or_(
    and_(
        cls.exceeds_max_non_backfill == expression.false(),
        cls.next_dagrun_create_after <= func.now(),
    ),
    cls.dag_id.in_(asset_triggered_dag_ids),
)

After a successful asset create_dagrun and consumed_asset_events.extend, call _set_exceeds_max_active_runs. Do not set it when no run is created. No session.commit() in that path.

QUEUED counts toward the cap, not only RUNNING. A stuck MANUAL run counts too. Catchup True/False does not change leftover consume. Partitioned same-key follow-up: events 2/3/4 land on run 2, not on the already-created run 1.

What's the impact

Asset leftovers stay visible once the live cap lifts, even if the cache still says the Dag is full. The next run consumes the leftover batch. This does not implement one run per event.

What's the testing

airflow-core/tests/unit/jobs/test_scheduler_job.py

• test_asset_events_queued_while_at_max_active_runs_are_all_consumed
• test_asset_events_wait_when_max_active_runs_is_two
• test_already_consumed_adrq_after_max_active_runs_lift_does_not_create_empty_run
• test_partitioned_same_key_events_during_max_active_runs_land_on_next_run

airflow-core/tests/unit/models/test_dag.py

• test_dags_needing_dagruns_assets_stale_flag_does_not_hide_adrq
• test_dags_needing_dagruns_assets_retains_adrq_at_max_active_runs
• test_dags_needing_dagruns_assets_max_active_runs_two

uv run --project airflow-core pytest \
  airflow-core/tests/unit/jobs/test_scheduler_job.py \
  airflow-core/tests/unit/models/test_dag.py \
  -k 'max_active_runs or dags_needing_dagruns_assets' -q

---

Was generative AI tooling used to co-author this PR?

• Yes (Grok 4.6)

Generated-by: Grok 4.6 following the guidelines

[Vamsi-klu]

cc @Lee-W @ashb @XD-DENG leftover ADRQ at max_active_runs. This keeps batch consume. It is not the one-run-per-event redesign.

---

Drafted-by: Cursor Grok 4.6; reviewed by @Vamsi-klu before posting