Make the Dag cache configurable and per-component in metrics

airflow-core/docs/administration-and-deployment/web-stack.rst

@@ -142,8 +142,8 @@ The following configuration options are available in the ``[api]`` section:
 - ``server_type``: ``uvicorn`` (default) or ``gunicorn``
 - ``worker_refresh_interval``: Seconds between worker refresh cycles (0 = disabled, default)
 - ``worker_refresh_batch_size``: Number of workers to refresh per cycle (default: 1)
-- ``dag_cache_size``: Max cached SerializedDAG versions in the API server (default: 64, 0 = unbounded)
-- ``dag_cache_ttl``: TTL in seconds for cached DAGs (default: 3600, 0 = LRU only)
+- ``dag_cache_size``: Max cached SerializedDAG versions in the API server (default: 64, 0 = no size limit)
+- ``dag_cache_ttl``: Idle timeout in seconds for cached Dags (default: 3600, 0 = no TTL; both 0 = no eviction)
 
 When to Use Gunicorn
 ^^^^^^^^^^^^^^^^^^^^
@@ -188,9 +188,10 @@ For example, to trigger a rolling restart of the API server pods:
 
    kubectl rollout restart deployment airflow-api-server
 
-The API server also supports bounded DAG caching via ``dag_cache_size`` and
-``dag_cache_ttl``, which limits memory consumed by cached SerializedDAG objects.
-This reduces memory growth from DAG version accumulation regardless of server type.
+The API server also evicts cached SerializedDAG objects via ``dag_cache_size`` and
+``dag_cache_ttl``, which reduces memory growth from Dag version accumulation regardless of
+server type. Note that only ``dag_cache_size`` caps memory outright: each re-check resets a
+cached entry's expiry, so ``dag_cache_ttl`` reclaims only the versions that stop being requested.
 
 In many Kubernetes environments, relying solely on Kubernetes OOM kills or
 crash restarts is not recommended, as memory growth may not always trigger an

airflow-core/docs/faq.rst

@@ -717,22 +717,29 @@ The API server caches serialized Dag objects in memory. Over time, as Dag versio
 
 There are two complementary approaches:
 
-**1. Bounded DAG caching (available since Airflow 3.3.0)**
+**1. Dag cache eviction (available since Airflow 3.2.2)**
 
-The API server supports LRU+TTL caching that bounds how many serialized Dag versions are kept
-in memory. Configure this in the ``[api]`` section:
+The API server can evict cached serialized Dag versions by size, by age, or both. Configure
+this in the ``[api]`` section:
 
 .. code-block:: ini
 
     [api]
-    dag_cache_size = 64    ; max cached versions (0 = unbounded, pre-3.2 behavior)
-    dag_cache_ttl = 3600   ; seconds before a cached entry expires (0 = LRU only)
+    dag_cache_size = 64    ; max cached versions (0 = no size limit)
+    dag_cache_ttl = 3600   ; seconds before an idle cached entry expires (0 = no TTL)
+
+``dag_cache_size`` is the only hard ceiling on memory. Each re-check resets a cached entry's
+expiry, so ``dag_cache_ttl`` reclaims only the Dag versions that stop being requested for the
+whole interval; versions that keep being requested are kept. It therefore trims the versions
+that accumulate over time, but memory still tracks the set of versions in active use. Setting
+both to 0 uses an unbounded dict with no eviction, 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.
+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)**
 
@@ -758,9 +765,48 @@ See :ref:`config:api__server_type`, :ref:`config:api__worker_refresh_interval`,
 .. note::
 
     Worker recycling handles memory growth from *any* source, not just the Dag cache.
-    For production deployments, using both bounded caching and gunicorn worker recycling
+    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, so before 3.4.0 its memory grew with every Dag
+version it had ever seen (see :ref:`faq:dag-version-inflation`) until the process was restarted or
+OOM killed. The cache is now bounded by default. Tune it 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 uses
+an unbounded dict with no eviction, matching the behavior before 3.4.0.
+
+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/71813.significant.rst

@@ -0,0 +1,12 @@
+``dag_cache_ttl`` is now honored when ``dag_cache_size = 0``, which previously disabled eviction and silently ignored the TTL. An ``[api] dag_cache_size = 0`` deployment now evicts on the default ``dag_cache_ttl = 3600``; set ``[api] dag_cache_ttl = 0`` to keep the old behavior. The scheduler gained the same options, ``[scheduler] dag_cache_size`` (default 512) and ``[scheduler] dag_cache_ttl`` (default 0), so it no longer retains every Dag version it has ever seen: the cache is capped at 512 versions and evicts the least recently used one beyond that. Raise ``[scheduler] dag_cache_size`` if more Dag versions have runs in flight than the limit, since a limit below that working set evicts versions still being scheduled and costs a re-fetch on the next loop. Set it to 0 for no size limit, leaving eviction to ``[scheduler] dag_cache_ttl``, which bounds memory by the concurrently active set rather than outright because each re-check resets a cached entry's expiry.
+
+* Types of change
+
+  * [ ] Dag changes
+  * [x] Config changes
+  * [ ] API changes
+  * [ ] CLI changes
+  * [x] Behaviour changes
+  * [ ] Plugin changes
+  * [ ] Dependency changes
+  * [ ] Code interface changes

airflow-core/src/airflow/api_fastapi/common/dagbag.py

@@ -16,43 +16,24 @@
 # under the License.
 from __future__ import annotations
 
-import logging
 from typing import TYPE_CHECKING, Annotated
 
 from fastapi import Depends, HTTPException, Request, status
 from sqlalchemy.orm import Session
 
 from airflow.configuration import conf
-from airflow.models.dagbag import DBDagBag
+from airflow.models.dagbag import DBDagBag, dag_cache_conf
 from airflow.models.serialized_dag import SerializedDagModel
 
 if TYPE_CHECKING:
     from airflow.models.dagrun import DagRun
     from airflow.serialization.definitions.dag import SerializedDAG
 
-log = logging.getLogger(__name__)
-
 
 def create_dag_bag() -> DBDagBag:
-    """Create DagBag with configurable LRU+TTL caching for API server usage."""
-    cache_size = conf.getint("api", "dag_cache_size", fallback=64)
-    cache_ttl_config = conf.getint("api", "dag_cache_ttl", fallback=3600)
-
-    if cache_size < 0:
-        log.warning("dag_cache_size must be >= 0, using unbounded dict")
-        cache_size = 0
-    if cache_ttl_config < 0:
-        log.warning("dag_cache_ttl must be >= 0, disabling TTL")
-        cache_ttl_config = 0
-
-    # Use unbounded dict (no eviction) if cache_size is 0
-    if cache_size <= 0:
-        return DBDagBag(cache_size=0)
-
-    # Disable TTL if cache_ttl is 0
-    cache_ttl: int | None = cache_ttl_config if cache_ttl_config > 0 else None
-
-    return DBDagBag(cache_size=cache_size, cache_ttl=cache_ttl)
+    """Build the API server's DagBag from the ``[api]`` cache options."""
+    cache_size, cache_ttl = dag_cache_conf("api", size_fallback=64, ttl_fallback=3600)
+    return DBDagBag(cache_size=cache_size, cache_ttl=cache_ttl, stats_prefix="api_server.dag_bag")
 
 
 def dag_bag_from_app(request: Request) -> DBDagBag:

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

@@ -1714,26 +1714,37 @@ api:
       default: "False"
     dag_cache_size:
       description: |
-        Size of the LRU cache for SerializedDAG objects in the API server.
-        Set to 0 to use an unbounded dict (no eviction, matching pre-3.2 behavior).
+        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.
+
         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
         version, but the deserialized result is cached for subsequent
         version-specific lookups.
-      version_added: 3.3.0
+      version_added: 3.2.2
       type: integer
       example: ~
       default: "64"
     dag_cache_ttl:
       description: |
-        Time-to-live (seconds) for cached SerializedDAG objects in the API server.
-        After this time, cached DAGs will be re-fetched from the database on next access.
-        Set to 0 to disable TTL (cache entries will only be evicted by LRU policy).
-
-        Note: After a DAG is updated, the API server may serve the previous version
-        until the cached entry expires. Lower values reduce staleness but increase
-        database load.
-      version_added: 3.3.0
+        Seconds a deserialized SerializedDAG stays in the API server's cache, applied whether or
+        not ``dag_cache_size`` sets a size limit. After this time the Dag is re-fetched from the
+        database on next access. Set to 0 to disable TTL, leaving eviction to the
+        ``dag_cache_size`` LRU policy.
+
+        Note that each re-check resets an entry's expiry, so this only reclaims Dag versions
+        that stop being requested for the whole interval. Versions that keep being requested are
+        kept, so memory tracks the set of Dag versions in active use and ``dag_cache_size``
+        remains the only hard ceiling.
+
+        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.2.2
       type: integer
       example: ~
       default: "3600"
@@ -2653,6 +2664,42 @@ 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``. Set both this and ``dag_cache_ttl`` to 0 to use an unbounded dict
+        with no eviction, matching the behavior before 3.4.0; the cache then grows with the number
+        of Dag versions the scheduler has ever seen, which on a long-running scheduler is
+        unbounded.
+      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

@@ -99,7 +99,7 @@
 )
 from airflow.models.dag import DagModel
 from airflow.models.dag_version import DagVersion, _resolve_version_data
-from airflow.models.dagbag import DBDagBag
+from airflow.models.dagbag import DBDagBag, dag_cache_conf
 from airflow.models.dagbundle import DagBundleModel
 from airflow.models.dagrun import DagRun
 from airflow.models.dagwarning import DagWarning, DagWarningType
@@ -298,6 +298,25 @@ def _get_current_dr_task_concurrency(states: Iterable[TaskInstanceState]) -> Sub
     )
 
 
+def _create_scheduler_dag_bag() -> DBDagBag:
+    """
+    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. 512 is meant 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.
+    """
+    cache_size, cache_ttl = dag_cache_conf("scheduler", size_fallback=512, ttl_fallback=0)
+    return DBDagBag(
+        load_op_links=False,
+        cache_size=cache_size,
+        cache_ttl=cache_ttl,
+        stats_prefix="scheduler.dag_bag",
+    )
+
+
 class SchedulerJobRunner(BaseJobRunner, LoggingMixin):
     """
     SchedulerJobRunner runs for a specific time interval and schedules jobs that are ready to run.
@@ -370,7 +389,7 @@ def __init__(
         if log:
             self._log = log
 
-        self.scheduler_dag_bag = DBDagBag(load_op_links=False)
+        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