This is an automated email from the ASF dual-hosted git repository.
jason810496 pushed a commit to branch v3-3-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/v3-3-test by this push:
new b32636456ee [v3-3-test] Bound the scheduler's deserialized Dag cache
(#71704) (#71821)
b32636456ee is described below
commit b32636456ee11dd79eb79ebe1430b1419247c05d
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Wed Aug 19 16:50:58 2026 +0800
[v3-3-test] Bound the scheduler's deserialized Dag cache (#71704) (#71821)
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)
closes: #69001
Co-authored-by: Jason(Zhe-You) Liu
<[email protected]>
---
airflow-core/newsfragments/71704.bugfix.rst | 1 +
airflow-core/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(-)
diff --git a/airflow-core/newsfragments/71704.bugfix.rst
b/airflow-core/newsfragments/71704.bugfix.rst
new file mode 100644
index 00000000000..5cd2aed7ec0
--- /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 a6343c4ae6c..d03b1d456f4 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 @@ DM = DagModel
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 @@ class SchedulerJobRunner(BaseJobRunner, LoggingMixin):
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 72ed2065bdb..419b459bc36 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_loader import ExecutorLoader
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 @@ class TestSchedulerJob:
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],
