Bound the scheduler's deserialized Dag cache (#71704)

The scheduler kept every Dag version it deserialized in a mapping that never
evicted, so a long-running scheduler grew with the number of versions it had
ever seen until it was restarted or OOM killed. Deployments that redeploy Dags
frequently accumulate versions fastest and hit this soonest.

A least-recently-used cap is the only thing that bounds this outright. An idle
timeout would not: the scheduler re-checks an entry on each lookup, which
re-arms its expiry, so a timeout reclaims a version only once its runs finish
and it stops being requested, leaving memory a function of the concurrently
active set rather than a fixed ceiling.

Deliberately not configurable here, so the fix stays small enough to
cherry-pick. Cache activity currently reports under the existing
api_server.dag_bag.* metrics; a scheduler-specific namespace, along with
configuration, follows separately.

closes: #69001

Author: jason810496

airflow-core/newsfragments/71704.bugfix.rst

@@ -0,0 +1 @@
+The scheduler's Dag cache is now a bounded LRU of 512 versions, so scheduler memory no longer grows with every Dag version the process has ever seen.

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

@@ -155,6 +155,18 @@
 TASK_STUCK_IN_QUEUED_RESCHEDULE_EVENT = "stuck in queued reschedule"
 """:meta private:"""
 
+SCHEDULER_DAG_CACHE_SIZE = 512
+"""
+Max deserialized Dag versions the scheduler keeps in memory.
+
+The scheduler reaches its DagBag through the Dag version of each active Dag run, so an
+unbounded cache retains every version the process has ever seen and grows for the life of
+the process. Sized to sit above the versions-with-runs-in-flight working set of a typical
+deployment, so eviction costs a re-fetch only where that working set is genuinely larger.
+
+:meta private:
+"""
+
 # Per-tick cap on pending AssetPartitionDagRun rows the scheduler evaluates.
 # Bounds the per-tick transaction so executor heartbeats and regular scheduling
 # aren't starved; remaining APDRs drain across subsequent ticks.
@@ -370,7 +382,7 @@ def __init__(
         if log:
             self._log = log
 
-        self.scheduler_dag_bag = DBDagBag(load_op_links=False)
+        self.scheduler_dag_bag = DBDagBag(load_op_links=False, cache_size=SCHEDULER_DAG_CACHE_SIZE)
 
         # Set of (dag_id, asset_name, asset_uri) tuples for trigger policies that
         # are permanently unreachable for the rollup window's cardinality — the

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

@@ -64,7 +64,7 @@
 from airflow.executors.executor_utils import ExecutorName
 from airflow.executors.local_executor import LocalExecutor
 from airflow.jobs.job import Job, run_job
-from airflow.jobs.scheduler_job_runner import SchedulerJobRunner
+from airflow.jobs.scheduler_job_runner import SCHEDULER_DAG_CACHE_SIZE, SchedulerJobRunner
 from airflow.models.asset import (
     AssetActive,
     AssetAliasModel,
@@ -414,6 +414,15 @@ def test_executor_loaded_in_scheduler_job(self, mock_init_executors, mock_defaul
         assert scheduler_job.executor == mock_local_executor
         assert scheduler_job.executors == [mock_local_executor]
 
+    def test_scheduler_dag_bag_is_bounded(self):
+        """The scheduler's Dag cache must evict, or it retains every version it has ever seen."""
+        from cachetools import LRUCache
+
+        job_runner = SchedulerJobRunner(Job())
+
+        assert isinstance(job_runner.scheduler_dag_bag._dags, LRUCache)
+        assert job_runner.scheduler_dag_bag._dags.maxsize == SCHEDULER_DAG_CACHE_SIZE
+
     @pytest.mark.parametrize(
         "heartrate",
         [10, 5],