[v3-3-test] Honor the API server Dag cache TTL when no size limit is set (#71814)

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``: TTL in seconds for cached Dags (default: 3600, 0 = no TTL; both 0 = no eviction)
 
 When to Use Gunicorn
 ^^^^^^^^^^^^^^^^^^^^
@@ -188,9 +188,11 @@ 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. A cached 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.
+If the TTL is shorter than that interval, even frequently requested entries can expire and reload between checks.
 
 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

@@ -700,16 +700,22 @@ 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 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.
 
 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``).
@@ -741,7 +747,7 @@ 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.
 
 

airflow-core/newsfragments/71814.bugfix.rst

@@ -0,0 +1 @@
+The API server now honors ``[api] dag_cache_ttl`` when ``dag_cache_size`` is 0, so cached serialized Dags can expire even when their count is not limited.

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

@@ -16,7 +16,6 @@
 # under the License.
 from __future__ import annotations
 
-import logging
 from typing import TYPE_CHECKING, Annotated
 
 from fastapi import Depends, HTTPException, Request, status
@@ -30,27 +29,16 @@
     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)
+    cache_ttl = 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
+        raise ValueError("[api] dag_cache_size must be greater than or equal to 0")
+    if cache_ttl < 0:
+        raise ValueError("[api] dag_cache_ttl must be greater than or equal to 0")
 
     return DBDagBag(cache_size=cache_size, cache_ttl=cache_ttl)
 

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

@@ -1680,26 +1680,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.
+
+        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. If the TTL is shorter
+        than that interval, even frequently requested entries can expire and reload between
+        checks. ``dag_cache_size`` remains the only hard ceiling on memory.
+
+        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"

airflow-core/src/airflow/models/dagbag.py

@@ -18,6 +18,7 @@
 from __future__ import annotations
 
 import hashlib
+import math
 import time
 from collections.abc import MutableMapping
 from contextlib import nullcontext
@@ -62,9 +63,9 @@ class DBDagBag:
     """
     Internal class for retrieving dags from the database.
 
-    Optionally supports LRU+TTL caching when cache_size is provided.
-    The scheduler uses this without caching, while the API server can
-    enable caching via configuration.
+    Optionally caches deserialized dags: a size limit enables LRU eviction, and a TTL enables
+    age-based eviction with or without a size limit. Callers that pass neither get a plain dict
+    that never evicts.
 
     :meta private:
     """
@@ -79,26 +80,37 @@ def __init__(
         Initialize DBDagBag.
 
         :param load_op_links: Should the extra operator link be loaded when de-serializing the DAG?
-        :param cache_size: Size of LRU cache. If None or 0, uses unbounded dict (no eviction).
-        :param cache_ttl: Time-to-live for cache entries in seconds. If None or 0, no TTL (LRU only).
+        :param cache_size: Max cached entries. 0 or None means no size limit.
+        :param cache_ttl: Seconds until a cached entry expires, applied with or without a size limit.
+            0 or None disables TTL. With neither a size limit nor a TTL the cache never evicts.
+        :raises ValueError: If ``cache_size`` or ``cache_ttl`` is negative.
         """
+        # Callers should reject negative values with their own context; validate again defensively.
+        if cache_size is not None and cache_size < 0:
+            raise ValueError("cache_size must be greater than or equal to 0")
+        if cache_ttl is not None and cache_ttl < 0:
+            raise ValueError("cache_ttl must be greater than or equal to 0")
+
         self.load_op_links = load_op_links
         self._dags: MutableMapping[UUID | str, _CacheEntry] = {}
         self._use_cache = False
 
         self._revalidation_interval = conf.getint("core", "min_serialized_dag_update_interval")
 
-        # Initialize bounded cache if cache_size is provided and > 0
-        if cache_size and cache_size > 0:
-            if cache_ttl and cache_ttl > 0:
-                self._dags = TTLCache(maxsize=cache_size, ttl=cache_ttl)
-            else:
-                self._dags = LRUCache(maxsize=cache_size)
+        # A TTL applies with or without a size limit: an uncapped TTLCache is what lets
+        # ``dag_cache_size = 0`` mean "no size limit" rather than "no eviction at all".
+        size = cache_size or 0
+        ttl = cache_ttl or 0
+        if ttl > 0:
+            self._dags = TTLCache(maxsize=size or math.inf, ttl=ttl)
+            self._use_cache = True
+        elif size > 0:
+            self._dags = LRUCache(maxsize=size)
             self._use_cache = True
 
         # Lock required for bounded caches: cachetools caches are NOT thread-safe
-        # (LRU reordering and TTL cleanup mutate internal linked lists).
-        # nullcontext for unbounded dict avoids lock overhead in the scheduler path.
+        # (LRU reordering and TTL cleanup mutate internal linked lists). A plain dict needs no
+        # lock, so it uses nullcontext.
         self._lock: RLock | nullcontext = RLock() if self._use_cache else nullcontext()
 
     def _read_dag(self, serdag: SerializedDagModel) -> SerializedDAG | None:

airflow-core/tests/unit/api_fastapi/common/test_dagbag.py

@@ -16,14 +16,18 @@
 # under the License.
 from __future__ import annotations
 
+import math
+import re
 from unittest import mock
 
 import pytest
 from cachetools import LRUCache, TTLCache
 
 from airflow.api_fastapi.app import purge_cached_app
+from airflow.api_fastapi.common.dagbag import create_dag_bag
 from airflow.sdk import BaseOperator
 
+from tests_common.test_utils.config import conf_vars
 from tests_common.test_utils.db import clear_db_dags, clear_db_runs, clear_db_serialized_dags
 
 pytestmark = pytest.mark.db_test
@@ -89,24 +93,41 @@ class TestCreateDagBag:
     """Tests for create_dag_bag() function."""
 
     @pytest.mark.parametrize(
-        ("cache_size", "cache_ttl", "expected_use_cache", "expected_dags_type"),
+        ("cache_size", "cache_ttl", "expected_dags_type", "expected_maxsize"),
         [
-            pytest.param(64, 3600, True, TTLCache, id="default_ttl_cache"),
-            pytest.param(0, 3600, False, dict, id="size_zero_unbounded"),
-            pytest.param(64, 0, True, LRUCache, id="ttl_zero_lru_only"),
+            pytest.param("64", "3600", TTLCache, 64, id="default_ttl_cache"),
+            pytest.param("0", "3600", TTLCache, math.inf, id="size_zero_ttl_only"),
+            pytest.param("64", "0", LRUCache, 64, id="ttl_zero_lru_only"),
+            pytest.param("0", "0", dict, None, id="both_zero_no_eviction"),
         ],
     )
-    @mock.patch("airflow.api_fastapi.common.dagbag.conf")
-    def test_create_dag_bag_cache_modes(
-        self, mock_conf, cache_size, cache_ttl, expected_use_cache, expected_dags_type
-    ):
-        from airflow.api_fastapi.common.dagbag import create_dag_bag
-
-        mock_conf.getint.side_effect = lambda section, key, fallback: {
-            "dag_cache_size": cache_size,
-            "dag_cache_ttl": cache_ttl,
-        }.get(key, fallback)
-
-        dag_bag = create_dag_bag()
-        assert dag_bag._use_cache is expected_use_cache
+    def test_create_dag_bag_cache_modes(self, cache_size, cache_ttl, expected_dags_type, expected_maxsize):
+        with conf_vars({("api", "dag_cache_size"): cache_size, ("api", "dag_cache_ttl"): cache_ttl}):
+            dag_bag = create_dag_bag()
+
         assert isinstance(dag_bag._dags, expected_dags_type)
+        assert dag_bag._use_cache is (expected_dags_type is not dict)
+        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",
+                "3600",
+                "[api] dag_cache_size must be greater than or equal to 0",
+                id="negative_size",
+            ),
+            pytest.param(
+                "64",
+                "-1",
+                "[api] dag_cache_ttl must be greater than or equal to 0",
+                id="negative_ttl",
+            ),
+        ],
+    )
+    def test_create_dag_bag_rejects_negative_config(self, cache_size, cache_ttl, expected_message):
+        with conf_vars({("api", "dag_cache_size"): cache_size, ("api", "dag_cache_ttl"): cache_ttl}):
+            with pytest.raises(ValueError, match=re.escape(expected_message)):
+                create_dag_bag()

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

@@ -16,6 +16,7 @@
 # under the License.
 from __future__ import annotations
 
+import math
 import time
 from concurrent.futures import ThreadPoolExecutor
 from unittest.mock import MagicMock, patch
@@ -246,29 +247,42 @@ def make_lazy(task_ids):
 class TestDBDagBagCache:
     """Tests for DBDagBag optional caching behavior."""
 
-    def test_no_caching_by_default(self):
-        """Test that DBDagBag uses a simple dict without caching by default."""
-        dag_bag = DBDagBag()
-        assert dag_bag._use_cache is False
-        assert isinstance(dag_bag._dags, dict)
-
-    def test_lru_cache_enabled_with_cache_size(self):
-        """Test that LRU cache is enabled when cache_size is provided."""
-        dag_bag = DBDagBag(cache_size=10)
-        assert dag_bag._use_cache is True
-        assert isinstance(dag_bag._dags, LRUCache)
-
-    def test_ttl_cache_enabled_with_cache_size_and_ttl(self):
-        """Test that TTL cache is enabled when both cache_size and cache_ttl are provided."""
-        dag_bag = DBDagBag(cache_size=10, cache_ttl=60)
-        assert dag_bag._use_cache is True
-        assert isinstance(dag_bag._dags, TTLCache)
-
-    def test_zero_cache_size_uses_unbounded_dict(self):
-        """Test that cache_size=0 uses unbounded dict (same as no caching)."""
-        dag_bag = DBDagBag(cache_size=0, cache_ttl=60)
-        assert dag_bag._use_cache is False
-        assert isinstance(dag_bag._dags, dict)
+    @pytest.mark.parametrize(
+        ("cache_size", "cache_ttl", "expected_type", "expected_maxsize"),
+        [
+            pytest.param(None, None, dict, None, id="neither_plain_dict"),
+            pytest.param(10, None, LRUCache, 10, id="size_only_lru"),
+            pytest.param(10, 60, TTLCache, 10, id="size_and_ttl_bounded_ttl"),
+            pytest.param(0, 60, TTLCache, math.inf, id="ttl_only_uncapped"),
+        ],
+    )
+    def test_cache_selection(self, cache_size, cache_ttl, expected_type, expected_maxsize):
+        dag_bag = DBDagBag(cache_size=cache_size, cache_ttl=cache_ttl)
+        assert isinstance(dag_bag._dags, expected_type)
+        assert dag_bag._use_cache is (expected_type is not dict)
+        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,
+                None,
+                "cache_size must be greater than or equal to 0",
+                id="negative_size",
+            ),
+            pytest.param(
+                None,
+                -1,
+                "cache_ttl must be greater than or equal to 0",
+                id="negative_ttl",
+            ),
+        ],
+    )
+    def test_rejects_negative_cache_configuration(self, cache_size, cache_ttl, expected_message):
+        with pytest.raises(ValueError, match=expected_message):
+            DBDagBag(cache_size=cache_size, cache_ttl=cache_ttl)
 
     def test_clear_cache_with_caching(self):
         """Test clear_cache() with caching enabled."""