Stop leftover asset events hiding behind a stale max_active_runs cache

The cached exceeds_max_non_backfill flag was applied to every Dag,
including asset-triggered ones, and asset create never refreshed it.
Leftovers then waited for an unrelated later event.

Co-authored-by: Cursor <cursoragent@cursor.com>

Author: Vamsi-klu

airflow-core/newsfragments/56050.bugfix.rst

@@ -0,0 +1 @@
+Asset-triggered Dags no longer skip leftover queued events when a stale ``max_active_runs`` cache still says the Dag is at capacity.

airflow-core/src/airflow/jobs/scheduler_job_runner.py

@@ -2752,6 +2752,7 @@ def _create_dag_runs_asset_triggered(
                 )
                 stats.incr("asset.triggered_dagruns", tags=prune_dict({"team_name": team_name}))
                 dag_run.consumed_asset_events.extend(asset_events)
+                self._set_exceeds_max_active_runs(dag_model=dag_model, session=session)
                 self.log.info(
                     "Created asset-triggered DagRun for '%s': run_id=%s, consumed %d asset events",
                     dag.dag_id,

airflow-core/src/airflow/models/dag.py

@@ -36,6 +36,7 @@
     Integer,
     String,
     Text,
+    and_,
     case,
     func,
     inspect as sa_inspect,
@@ -776,9 +777,13 @@ def dag_ready(dag_id: str, cond: SerializedAssetBase, statuses: dict[UKey, bool]
                 cls.is_paused == expression.false(),
                 cls.is_stale == expression.false(),
                 cls.has_import_errors == expression.false(),
-                cls.exceeds_max_non_backfill == expression.false(),
+                # The cached flag is timetable-only. Asset leftovers stay selectable
+                # here; the live QUEUED+RUNNING count above already deferred Dags at cap.
                 or_(
-                    cls.next_dagrun_create_after <= func.now(),
+                    and_(
+                        cls.exceeds_max_non_backfill == expression.false(),
+                        cls.next_dagrun_create_after <= func.now(),
+                    ),
                     cls.dag_id.in_(asset_triggered_dag_ids),
                 ),
             )

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

@@ -5940,10 +5940,12 @@ def _make_event(timestamp):
     def test_asset_events_queued_while_at_max_active_runs_are_all_consumed(
         self, catchup, terminal_state, session, dag_maker
     ):
-        """Events 2, 3, 4 that land while run 1 holds ``max_active_runs=1`` must all be
-        consumed by run 2 after the cap lifts. Catchup only changes whether
-        pre-subscription events join the first run; it must not change this
-        follow-up consume. A FAILED run 1 lifts the cap the same way SUCCESS does.
+        """Events queued while a Dag is at max_active_runs land on the next run.
+
+        Events 2, 3, 4 that land while run 1 holds ``max_active_runs=1`` must all be
+        consumed by run 2 after the cap lifts. A later event 5 is its own run.
+        Catchup only changes whether pre-subscription events join the first run.
+        A FAILED run 1 lifts the cap the same way SUCCESS does.
         """
         asset = Asset(uri="test://asset-max-active-runs-queue", name="mar_queue_asset", group="test_group")
         with dag_maker(
@@ -6024,6 +6026,21 @@ def _adrq_event_ids():
         assert run2.run_after == timezone.coerce_datetime(event4.timestamp)
         assert _adrq_event_ids() == set()
 
+        event5 = _queue_event(base + timedelta(minutes=10))
+        _tick()
+        runs = _runs()
+        assert len(runs) == 2
+        assert _adrq_event_ids() == {event5.id}
+
+        run2 = session.merge(run2)
+        run2.state = terminal_state
+        session.flush()
+        _tick()
+        runs = _runs()
+        assert len(runs) == 3
+        assert {e.id for e in runs[2].consumed_asset_events} == {event5.id}
+        assert _adrq_event_ids() == set()
+
     @pytest.mark.need_serialized_dag
     def test_asset_events_wait_when_max_active_runs_is_two(self, session, dag_maker):
         """A third queued event is not consumed until one of two active runs finishes."""

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

@@ -2612,6 +2612,40 @@ def test_dags_needing_dagruns_assets(self, dag_maker, session):
         dag_models = query.all()
         assert dag_models == [dag_model]
 
+    def test_dags_needing_dagruns_assets_stale_flag_does_not_hide_adrq(self, dag_maker, session):
+        """A stale exceeds_max_non_backfill flag must not hide asset leftovers.
+
+        The cached flag is timetable-only. With zero active runs and queued
+        events, the Dag stays selectable even when the flag is still True.
+        """
+        asset = Asset(uri="test://asset-stale-flag", group="test-group")
+        with dag_maker(
+            session=session,
+            dag_id="stale_flag_asset_consumer",
+            max_active_runs=1,
+            schedule=[asset],
+            start_date=pendulum.now().add(days=-2),
+        ) as dag:
+            EmptyOperator(task_id="dummy")
+
+        dag_model = session.scalar(select(DagModel).where(DagModel.dag_id == dag.dag_id))
+        asset_model: AssetModel = dag_model.schedule_assets[0]
+        for _ in range(3):
+            event = AssetEvent(asset_id=asset_model.id, timestamp=timezone.utcnow())
+            session.add(event)
+            session.flush()
+            session.add(
+                AssetDagRunQueue(
+                    asset_id=asset_model.id, target_dag_id=dag_model.dag_id, asset_event_id=event.id
+                )
+            )
+        dag_model.exceeds_max_non_backfill = True
+        session.flush()
+
+        query, triggered_date_by_dag = DagModel.dags_needing_dagruns(session)
+        assert query.all() == [dag_model]
+        assert dag_model.dag_id in triggered_date_by_dag
+
     @pytest.mark.parametrize("active_state", [DagRunState.QUEUED, DagRunState.RUNNING])
     @pytest.mark.parametrize(
         "run_type",