From 0e845112da179d298082f85cdcdfb6c1335af8a7 Mon Sep 17 00:00:00 2001
From: "Jason(Zhe-You) Liu" <68415893+jason810496@users.noreply.github.com>
Date: Wed, 19 Aug 2026 15:18:02 +0800
Subject: [PATCH] [v3-3-test] 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.
(cherry picked from commit 29dd99d0034c3cff0fe37882f6b210ffca902d29)

Co-authored-by: Jason(Zhe-You) Liu <68415893+jason810496@users.noreply.github.com>
closes: #69001
---
 airflow-core/newsfragments/71704.bugfix.rst        |  1 +
 .../src/airflow/jobs/scheduler_job_runner.py       | 14 +++++++++++++-
 airflow-core/tests/unit/jobs/test_scheduler_job.py | 11 ++++++++++-
 3 files changed, 24 insertions(+), 2 deletions(-)
 create mode 100644 airflow-core/newsfragments/71704.bugfix.rst

diff --git a/airflow-core/newsfragments/71704.bugfix.rst b/airflow-core/newsfragments/71704.bugfix.rst
new file mode 100644
index 0000000000000..5cd2aed7ec0dd
--- /dev/null
+++ b/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.
diff --git a/airflow-core/src/airflow/jobs/scheduler_job_runner.py b/airflow-core/src/airflow/jobs/scheduler_job_runner.py
index a6343c4ae6c71..d03b1d456f466 100644
--- a/airflow-core/src/airflow/jobs/scheduler_job_runner.py
+++ b/airflow-core/src/airflow/jobs/scheduler_job_runner.py
@@ -154,6 +154,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.
@@ -348,7 +360,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
diff --git a/airflow-core/tests/unit/jobs/test_scheduler_job.py b/airflow-core/tests/unit/jobs/test_scheduler_job.py
index 72ed2065bdbff..419b459bc3619 100644
--- a/airflow-core/tests/unit/jobs/test_scheduler_job.py
+++ b/airflow-core/tests/unit/jobs/test_scheduler_job.py
@@ -57,7 +57,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,
@@ -410,6 +410,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],