Make the scheduler Dag cache size and TTL configurable

airflow-core/docs/faq.rst

@@ -728,17 +728,18 @@ this in the ``[api]`` section:
     dag_cache_size = 64    ; max cached versions (0 = no size limit)
     dag_cache_ttl = 3600   ; seconds before a cached entry expires (0 = no TTL)
 
-``dag_cache_size`` is the only hard ceiling on memory. An entry's TTL is refreshed only when the
-entry is checked against the database after ``[core] min_serialized_dag_update_interval``, not on
-every request. With a shorter TTL, even frequently requested entries can expire and reload
-between checks. Setting both options to 0 uses an unbounded dict with no eviction, matching the
-behavior before 3.2.2.
+``dag_cache_size`` is the only hard ceiling on memory. An entry's TTL is refreshed only when the entry is
+checked against the database after ``[core] min_serialized_dag_update_interval``, not on every request.
+With a shorter TTL, even frequently requested entries can expire and reload between checks. Setting both
+options to 0 disables eviction, so the API server retains every deserialized Dag it loads until restart,
+matching the behavior before 3.2.2.
 
 The cache is keyed by Dag version ID. After a Dag is updated, the API server may serve the
 previous version until the cached entry expires (controlled by ``dag_cache_ttl``).
 
-See :ref:`config:api__dag_cache_size` and :ref:`config:api__dag_cache_ttl` for the full
-configuration reference.
+See :ref:`config:api__dag_cache_size` and :ref:`config:api__dag_cache_ttl` for the full configuration
+reference. The scheduler has the same pair of options under ``[scheduler]`` (see
+:ref:`faq:scheduler-memory-growth`).
 
 **2. Gunicorn with rolling worker restarts (available since Airflow 3.2.0)**
 
@@ -767,6 +768,46 @@ See :ref:`config:api__server_type`, :ref:`config:api__worker_refresh_interval`,
     For production deployments, using both cache eviction and gunicorn worker recycling
     provides the best results.
 
+.. _faq:scheduler-memory-growth:
+
+How to prevent scheduler memory growth?
+----------------------------------------
+
+The scheduler caches deserialized Dag objects. Airflow 3.3.2 capped this cache at 512 Dag versions to
+prevent memory from growing with every version the scheduler had ever seen (see
+:ref:`faq:dag-version-inflation`). Starting in Airflow 3.4.0, you can adjust that size limit and optionally
+add time-based eviction in the ``[scheduler]`` section:
+
+.. code-block:: ini
+
+    [scheduler]
+    dag_cache_size = 512   ; max cached versions, evicting least recently used
+    dag_cache_ttl = 0      ; seconds before an idle cached entry expires (0 = no TTL)
+
+The defaults cap the cache at 512 Dag versions and evict the least recently used one beyond that, so memory
+cannot grow with the number of versions the scheduler has ever seen. The scheduler reaches the cache through
+the Dag version of each active Dag run, so its working set is the versions with runs in flight; 512 is meant
+to sit above that for a typical deployment.
+
+If the scheduler is still OOM killed, lower ``dag_cache_size``. Raise it if you have more Dag versions with
+runs in flight than the limit, since a limit below the working set evicts versions that are still being
+scheduled and costs a database fetch and a deserialization on the next loop — watch
+``scheduler.dag_bag.cache_miss`` to tell the two apart. Setting ``dag_cache_size = 0`` switches to no size
+limit, leaving eviction to ``dag_cache_ttl``, which bounds memory by the concurrently active set rather than
+outright: each re-check resets an entry's expiry, so a TTL reclaims a version only once its runs finish and
+it stops being requested. Setting both to 0 disables eviction, so the scheduler retains every deserialized
+Dag it loads until restart, matching the behavior before 3.3.2.
+
+Neither option affects how quickly the scheduler picks up a Dag change. A Dag update that creates a new
+version is seen immediately, because the new version is a different cache key. A version rewritten in place
+is re-checked against its current hash once
+:ref:`config:core__min_serialized_dag_update_interval` has elapsed since the entry was last
+validated, so that option, not ``dag_cache_ttl``, bounds how long a rewritten version can be served stale.
+The same applies to the API server.
+
+See :ref:`config:scheduler__dag_cache_size` and :ref:`config:scheduler__dag_cache_ttl` for the full
+configuration reference.
+
 
 MySQL and MySQL variant Databases
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

airflow-core/newsfragments/71816.feature.rst

@@ -0,0 +1 @@
+Added ``[scheduler] dag_cache_size`` and ``[scheduler] dag_cache_ttl`` options for tuning deserialized Dag cache eviction. The cache defaults to an LRU limit of 512 versions with no TTL. A value of 0 disables the corresponding eviction limit; setting both options to 0 disables eviction.

airflow-core/src/airflow/config_templates/config.yml

@@ -1714,10 +1714,10 @@ api:
       default: "False"
     dag_cache_size:
       description: |
-        Max number of deserialized SerializedDAG objects the API server keeps in memory.
-        Set to 0 for no size limit, leaving eviction to ``dag_cache_ttl``. Set both this and
-        ``dag_cache_ttl`` to 0 to use an unbounded dict with no eviction, matching the behavior
-        before 3.2.2.
+        Max number of deserialized SerializedDAG objects the API server keeps in memory. Set to 0 for no
+        size limit, leaving eviction to ``dag_cache_ttl``. Setting both this and ``dag_cache_ttl`` to 0
+        disables eviction, so the API server retains every deserialized Dag it loads until restart, matching
+        the behavior before 3.2.2.
 
         The cache is keyed by Dag version ID, so lookups by Dag ID
         (e.g., viewing a Dag's details) always query the database for the latest
@@ -2664,6 +2664,40 @@ scheduler:
       type: integer
       example: ~
       default: "60"
+    dag_cache_size:
+      description: |
+        Max number of deserialized SerializedDAG objects the scheduler keeps in memory, keyed by Dag version
+        ID. Once full, the least recently used version is evicted, so this is a hard ceiling on how much the
+        cache can grow no matter how many Dag versions accumulate.
+
+        Raise it if the scheduler has more Dag versions with runs in flight than this; a limit below the
+        working set evicts versions that are still being scheduled, costing a database fetch and a
+        deserialization on the next loop. Set to 0 for no size limit, leaving eviction to ``dag_cache_ttl``.
+        Setting both this and ``dag_cache_ttl`` to 0 disables eviction, so the scheduler retains every
+        deserialized Dag it loads until restart, matching the behavior before 3.3.2.
+      version_added: 3.4.0
+      type: integer
+      example: ~
+      default: "512"
+    dag_cache_ttl:
+      description: |
+        Seconds a deserialized SerializedDAG stays in the scheduler's cache. Defaults to 0, which disables
+        TTL and leaves eviction to the ``dag_cache_size`` LRU policy.
+
+        Note that each re-check resets an entry's expiry, so a TTL only reclaims Dag versions that stop being
+        requested for the whole interval. Versions still referenced by active Dag runs are kept, so
+        ``dag_cache_size`` remains the only hard ceiling and a TTL alone (``dag_cache_size = 0``) bounds
+        memory by the concurrently active set rather than outright.
+
+        Note: this does not govern staleness. A Dag update that creates a new version is picked up
+        immediately, because the new version is a different cache key. A version rewritten in place is
+        re-checked against its current ``dag_hash`` once ``[core] min_serialized_dag_update_interval`` has
+        elapsed since the entry was last validated, so that option bounds how long a rewritten version can be
+        served stale.
+      version_added: 3.4.0
+      type: integer
+      example: ~
+      default: "0"
     pool_metrics_interval:
       description: |
         How often (in seconds) should pool usage stats be sent to StatsD (if statsd_on is enabled)

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

@@ -156,17 +156,6 @@
 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
@@ -301,6 +290,30 @@ def _is_parent_process() -> bool:
     return multiprocessing.current_process().name == "MainProcess"
 
 
+def _create_scheduler_dag_bag() -> CachedDBDagBag:
+    """
+    Build the scheduler's DagBag from the ``[scheduler]`` cache options.
+
+    Defaults to an LRU cache of 512 versions with no TTL. A size limit is the only hard ceiling:
+    each re-check resets an entry's expiry, so a TTL reclaims a version only once its runs finish
+    and it stops being requested, bounding memory by the concurrently active set rather than
+    outright. With both limits disabled, retain loaded Dags without eviction for the process
+    lifetime.
+    """
+    cache_size = conf.getint("scheduler", "dag_cache_size", fallback=512)
+    cache_ttl = conf.getint("scheduler", "dag_cache_ttl", fallback=0)
+    if cache_size < 0:
+        raise ValueError("[scheduler] dag_cache_size must be greater than or equal to 0")
+    if cache_ttl < 0:
+        raise ValueError("[scheduler] dag_cache_ttl must be greater than or equal to 0")
+    return CachedDBDagBag(
+        load_op_links=False,
+        cache_size=cache_size,
+        cache_ttl=cache_ttl,
+        stats_prefix="scheduler.dag_bag",
+    )
+
+
 def _get_current_dr_task_concurrency(states: Iterable[TaskInstanceState]) -> Subquery:
     """Get the dag_run IDs and how many tasks are in the provided states for each one."""
     return (
@@ -383,12 +396,7 @@ def __init__(
         if log:
             self._log = log
 
-        self.scheduler_dag_bag = CachedDBDagBag(
-            load_op_links=False,
-            cache_size=SCHEDULER_DAG_CACHE_SIZE,
-            cache_ttl=0,
-            stats_prefix="scheduler.dag_bag",
-        )
+        self.scheduler_dag_bag = _create_scheduler_dag_bag()
 
         # 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 SCHEDULER_DAG_CACHE_SIZE, SchedulerJobRunner
+from airflow.jobs.scheduler_job_runner import SchedulerJobRunner
 from airflow.models.asset import (
     AssetActive,
     AssetAliasModel,
@@ -423,7 +423,7 @@ def test_scheduler_dag_bag_is_bounded(self):
 
         assert isinstance(job_runner.scheduler_dag_bag, CachedDBDagBag)
         assert isinstance(job_runner.scheduler_dag_bag._dags, LRUCache)
-        assert job_runner.scheduler_dag_bag._dags.maxsize == SCHEDULER_DAG_CACHE_SIZE
+        assert job_runner.scheduler_dag_bag._dags.maxsize == 512
         # Reported separately from the API server's cache, not folded into it.
         assert job_runner.scheduler_dag_bag._stats_prefix == "scheduler.dag_bag"
 

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

@@ -17,6 +17,7 @@
 from __future__ import annotations
 
 import math
+import re
 import time
 from concurrent.futures import ThreadPoolExecutor
 from unittest.mock import MagicMock, patch
@@ -36,6 +37,7 @@
 from airflow.utils.session import create_session
 
 from tests_common.test_utils import db
+from tests_common.test_utils.config import conf_vars
 
 pytestmark = pytest.mark.db_test
 
@@ -316,6 +318,61 @@ def test_api_server_reports_under_its_own_namespace(self):
 
         assert create_dag_bag()._stats_prefix == "api_server.dag_bag"
 
+    @pytest.mark.parametrize(
+        ("cache_size", "cache_ttl", "expected_bag_type", "expected_cache_type", "expected_maxsize"),
+        [
+            pytest.param(None, None, CachedDBDagBag, LRUCache, 512, id="defaults_bounded_lru"),
+            pytest.param("1024", "0", CachedDBDagBag, LRUCache, 1024, id="size_only_lru"),
+            pytest.param("1024", "3600", CachedDBDagBag, TTLCache, 1024, id="size_and_ttl"),
+            pytest.param("0", "3600", CachedDBDagBag, TTLCache, math.inf, id="ttl_only_uncapped"),
+            pytest.param("0", "0", CachedDBDagBag, dict, None, id="both_zero_no_eviction"),
+        ],
+    )
+    def test_scheduler_cache_follows_its_config(
+        self, cache_size, cache_ttl, expected_bag_type, expected_cache_type, expected_maxsize
+    ):
+        from airflow.jobs.scheduler_job_runner import _create_scheduler_dag_bag
+
+        overrides = {}
+        if cache_size is not None:
+            overrides[("scheduler", "dag_cache_size")] = cache_size
+        if cache_ttl is not None:
+            overrides[("scheduler", "dag_cache_ttl")] = cache_ttl
+
+        with conf_vars(overrides):
+            dag_bag = _create_scheduler_dag_bag()
+
+        assert type(dag_bag) is expected_bag_type
+        assert isinstance(dag_bag._dags, expected_cache_type)
+        if expected_maxsize is not None:
+            assert dag_bag._dags.maxsize == expected_maxsize
+
+    @pytest.mark.parametrize(
+        ("cache_size", "cache_ttl", "expected_message"),
+        [
+            pytest.param(
+                "-1",
+                "0",
+                "[scheduler] dag_cache_size must be greater than or equal to 0",
+                id="negative_size",
+            ),
+            pytest.param(
+                "512",
+                "-1",
+                "[scheduler] dag_cache_ttl must be greater than or equal to 0",
+                id="negative_ttl",
+            ),
+        ],
+    )
+    def test_scheduler_cache_rejects_negative_config(self, cache_size, cache_ttl, expected_message):
+        from airflow.jobs.scheduler_job_runner import _create_scheduler_dag_bag
+
+        with conf_vars(
+            {("scheduler", "dag_cache_size"): cache_size, ("scheduler", "dag_cache_ttl"): cache_ttl}
+        ):
+            with pytest.raises(ValueError, match=re.escape(expected_message)):
+                _create_scheduler_dag_bag()
+
     def test_cached_bag_requires_non_empty_stats_prefix(self):
         """A cache with no namespace to report under must fail at wiring time, not mid-request."""
         with pytest.raises(ValueError, match="requires a stats_prefix"):