From 8d82fb154b877b426995b5d6453e380db5ba01eb Mon Sep 17 00:00:00 2001 From: LIU ZHE YOU Date: Wed, 19 Aug 2026 05:48:03 +0000 Subject: [PATCH 1/5] Honor the API server Dag cache TTL when no size limit is set `[api] dag_cache_size = 0` reads as "no size limit", but it selected a mapping that never evicted at all and silently ignored `[api] dag_cache_ttl`. Age-based eviction could therefore only be enabled by also accepting a size cap, which is backwards for a deployment that wants to bound staleness rather than count. A TTL now applies with or without a size limit, and only setting both options to 0 disables eviction. Deployments on `dag_cache_size = 0` start evicting on the default hour-long TTL; setting `dag_cache_ttl = 0` restores the old behaviour. The options were also documented as added in 3.3.0. They shipped in 3.2.2. --- .../web-stack.rst | 11 ++--- airflow-core/docs/faq.rst | 18 ++++++--- .../src/airflow/api_fastapi/common/dagbag.py | 15 ++----- .../src/airflow/config_templates/config.yml | 33 ++++++++++----- airflow-core/src/airflow/models/dagbag.py | 31 ++++++++------ .../unit/api_fastapi/common/test_dagbag.py | 28 ++++++------- airflow-core/tests/unit/models/test_dagbag.py | 40 ++++++++----------- 7 files changed, 93 insertions(+), 83 deletions(-) diff --git a/airflow-core/docs/administration-and-deployment/web-stack.rst b/airflow-core/docs/administration-and-deployment/web-stack.rst index f043789d5e553..55146ce9a7316 100644 --- a/airflow-core/docs/administration-and-deployment/web-stack.rst +++ b/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 diff --git a/airflow-core/docs/faq.rst b/airflow-core/docs/faq.rst index ce89ecac393ce..59a3fa44475c0 100644 --- a/airflow-core/docs/faq.rst +++ b/airflow-core/docs/faq.rst @@ -717,16 +717,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 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``). @@ -758,7 +764,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. diff --git a/airflow-core/src/airflow/api_fastapi/common/dagbag.py b/airflow-core/src/airflow/api_fastapi/common/dagbag.py index d87aca49a524a..551ef77d04f6a 100644 --- a/airflow-core/src/airflow/api_fastapi/common/dagbag.py +++ b/airflow-core/src/airflow/api_fastapi/common/dagbag.py @@ -36,21 +36,14 @@ 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") + log.warning("dag_cache_size must be >= 0, using 0 (no size limit)") cache_size = 0 - if cache_ttl_config < 0: + if cache_ttl < 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 + cache_ttl = 0 return DBDagBag(cache_size=cache_size, cache_ttl=cache_ttl) diff --git a/airflow-core/src/airflow/config_templates/config.yml b/airflow-core/src/airflow/config_templates/config.yml index 6d2438fd21362..10247f3ca21b9 100644 --- a/airflow-core/src/airflow/config_templates/config.yml +++ b/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" diff --git a/airflow-core/src/airflow/models/dagbag.py b/airflow-core/src/airflow/models/dagbag.py index 5c0556974ff3b..d8d1d71849892 100644 --- a/airflow-core/src/airflow/models/dagbag.py +++ b/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,8 +80,9 @@ 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. """ self.load_op_links = load_op_links self._dags: MutableMapping[UUID | str, _CacheEntry] = {} @@ -88,17 +90,20 @@ def __init__( 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 = max(cache_size or 0, 0) + ttl = max(cache_ttl or 0, 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: diff --git a/airflow-core/tests/unit/api_fastapi/common/test_dagbag.py b/airflow-core/tests/unit/api_fastapi/common/test_dagbag.py index 48c6f706ba7e2..c2924536cbe9e 100644 --- a/airflow-core/tests/unit/api_fastapi/common/test_dagbag.py +++ b/airflow-core/tests/unit/api_fastapi/common/test_dagbag.py @@ -16,6 +16,7 @@ # under the License. from __future__ import annotations +import math from unittest import mock import pytest @@ -24,6 +25,7 @@ from airflow.api_fastapi.app import purge_cached_app 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 +91,22 @@ 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"), + pytest.param("-1", "3600", TTLCache, math.inf, id="negative_size_clamped"), ], ) - @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 - ): + def test_create_dag_bag_cache_modes(self, cache_size, cache_ttl, expected_dags_type, expected_maxsize): 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) + with conf_vars({("api", "dag_cache_size"): cache_size, ("api", "dag_cache_ttl"): cache_ttl}): + dag_bag = create_dag_bag() - dag_bag = create_dag_bag() - assert dag_bag._use_cache is expected_use_cache 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 diff --git a/airflow-core/tests/unit/models/test_dagbag.py b/airflow-core/tests/unit/models/test_dagbag.py index 79668d4fe54f4..10a4c8b4c80b8 100644 --- a/airflow-core/tests/unit/models/test_dagbag.py +++ b/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,22 @@ 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(-1, -1, dict, None, id="negatives_clamped_to_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 def test_clear_cache_with_caching(self): """Test clear_cache() with caching enabled.""" From 5f6c258782ad01bd398a57728e98da3e762512cd Mon Sep 17 00:00:00 2001 From: LIU ZHE YOU Date: Wed, 19 Aug 2026 09:31:54 +0000 Subject: [PATCH 2/5] Reject negative API Dag cache values and clarify TTL behavior Negative cache configuration should fail fast, and operator guidance must reflect that TTL refreshes happen only at revalidation boundaries. --- .../web-stack.rst | 8 +++--- airflow-core/docs/faq.rst | 12 ++++----- airflow-core/newsfragments/71814.bugfix.rst | 1 + .../src/airflow/api_fastapi/common/dagbag.py | 9 ++----- .../src/airflow/config_templates/config.yml | 8 +++--- .../unit/api_fastapi/common/test_dagbag.py | 26 ++++++++++++++++--- 6 files changed, 41 insertions(+), 23 deletions(-) create mode 100644 airflow-core/newsfragments/71814.bugfix.rst diff --git a/airflow-core/docs/administration-and-deployment/web-stack.rst b/airflow-core/docs/administration-and-deployment/web-stack.rst index 55146ce9a7316..f1a9c8eaf78c5 100644 --- a/airflow-core/docs/administration-and-deployment/web-stack.rst +++ b/airflow-core/docs/administration-and-deployment/web-stack.rst @@ -143,7 +143,7 @@ The following configuration options are available in the ``[api]`` section: - ``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 = no size limit) -- ``dag_cache_ttl``: Idle timeout in seconds for cached Dags (default: 3600, 0 = no TTL; both 0 = no eviction) +- ``dag_cache_ttl``: TTL in seconds for cached Dags (default: 3600, 0 = no TTL; both 0 = no eviction) When to Use Gunicorn ^^^^^^^^^^^^^^^^^^^^ @@ -190,8 +190,10 @@ For example, to trigger a rolling restart of the API server pods: 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. +server type. Note that only ``dag_cache_size`` caps memory outright. A cached entry's TTL is +refreshed when the entry is revalidated 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 revalidations. In many Kubernetes environments, relying solely on Kubernetes OOM kills or crash restarts is not recommended, as memory growth may not always trigger an diff --git a/airflow-core/docs/faq.rst b/airflow-core/docs/faq.rst index 59a3fa44475c0..93292ead8e291 100644 --- a/airflow-core/docs/faq.rst +++ b/airflow-core/docs/faq.rst @@ -726,13 +726,13 @@ this in the ``[api]`` section: [api] 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_ttl = 3600 ; seconds before a 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. +``dag_cache_size`` is the only hard ceiling on memory. An entry's TTL is refreshed only when the +entry is revalidated after ``[core] min_serialized_dag_update_interval``, not on every request. +With a shorter TTL, even frequently requested entries can expire and reload between +revalidations. 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``). diff --git a/airflow-core/newsfragments/71814.bugfix.rst b/airflow-core/newsfragments/71814.bugfix.rst new file mode 100644 index 0000000000000..220e3141d3d11 --- /dev/null +++ b/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. diff --git a/airflow-core/src/airflow/api_fastapi/common/dagbag.py b/airflow-core/src/airflow/api_fastapi/common/dagbag.py index 551ef77d04f6a..374a4a2adfe5b 100644 --- a/airflow-core/src/airflow/api_fastapi/common/dagbag.py +++ b/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,8 +29,6 @@ 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.""" @@ -39,11 +36,9 @@ def create_dag_bag() -> DBDagBag: cache_ttl = conf.getint("api", "dag_cache_ttl", fallback=3600) if cache_size < 0: - log.warning("dag_cache_size must be >= 0, using 0 (no size limit)") - cache_size = 0 + raise ValueError("dag_cache_size must be greater than or equal to 0") if cache_ttl < 0: - log.warning("dag_cache_ttl must be >= 0, disabling TTL") - cache_ttl = 0 + raise ValueError("dag_cache_ttl must be greater than or equal to 0") return DBDagBag(cache_size=cache_size, cache_ttl=cache_ttl) diff --git a/airflow-core/src/airflow/config_templates/config.yml b/airflow-core/src/airflow/config_templates/config.yml index 10247f3ca21b9..9059d378158e7 100644 --- a/airflow-core/src/airflow/config_templates/config.yml +++ b/airflow-core/src/airflow/config_templates/config.yml @@ -1734,10 +1734,10 @@ api: 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. + An entry's TTL is refreshed only when the entry is revalidated 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 + revalidations. ``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 diff --git a/airflow-core/tests/unit/api_fastapi/common/test_dagbag.py b/airflow-core/tests/unit/api_fastapi/common/test_dagbag.py index c2924536cbe9e..4b20025a8227e 100644 --- a/airflow-core/tests/unit/api_fastapi/common/test_dagbag.py +++ b/airflow-core/tests/unit/api_fastapi/common/test_dagbag.py @@ -23,6 +23,7 @@ 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 @@ -97,12 +98,9 @@ class TestCreateDagBag: 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"), - pytest.param("-1", "3600", TTLCache, math.inf, id="negative_size_clamped"), ], ) def test_create_dag_bag_cache_modes(self, cache_size, cache_ttl, expected_dags_type, expected_maxsize): - from airflow.api_fastapi.common.dagbag import create_dag_bag - with conf_vars({("api", "dag_cache_size"): cache_size, ("api", "dag_cache_ttl"): cache_ttl}): dag_bag = create_dag_bag() @@ -110,3 +108,25 @@ def test_create_dag_bag_cache_modes(self, cache_size, cache_ttl, expected_dags_t 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", + "dag_cache_size must be greater than or equal to 0", + id="negative_size", + ), + pytest.param( + "64", + "-1", + "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=expected_message): + create_dag_bag() From 00ea7fe0bc95c161d39e3e7c0a27ba97b81b71e3 Mon Sep 17 00:00:00 2001 From: LIU ZHE YOU Date: Wed, 19 Aug 2026 09:55:25 +0000 Subject: [PATCH 3/5] Reject negative cache values for every DBDagBag caller Validating at the shared boundary prevents non-API callers from silently clamping invalid cache settings. --- .../src/airflow/api_fastapi/common/dagbag.py | 5 ----- airflow-core/src/airflow/models/dagbag.py | 10 +++++++-- .../unit/api_fastapi/common/test_dagbag.py | 22 ------------------- airflow-core/tests/unit/models/test_dagbag.py | 22 ++++++++++++++++++- 4 files changed, 29 insertions(+), 30 deletions(-) diff --git a/airflow-core/src/airflow/api_fastapi/common/dagbag.py b/airflow-core/src/airflow/api_fastapi/common/dagbag.py index 374a4a2adfe5b..94815ed096b35 100644 --- a/airflow-core/src/airflow/api_fastapi/common/dagbag.py +++ b/airflow-core/src/airflow/api_fastapi/common/dagbag.py @@ -35,11 +35,6 @@ def create_dag_bag() -> DBDagBag: cache_size = conf.getint("api", "dag_cache_size", fallback=64) cache_ttl = conf.getint("api", "dag_cache_ttl", fallback=3600) - if cache_size < 0: - raise ValueError("dag_cache_size must be greater than or equal to 0") - if cache_ttl < 0: - raise ValueError("dag_cache_ttl must be greater than or equal to 0") - return DBDagBag(cache_size=cache_size, cache_ttl=cache_ttl) diff --git a/airflow-core/src/airflow/models/dagbag.py b/airflow-core/src/airflow/models/dagbag.py index d8d1d71849892..e54df53ceee8b 100644 --- a/airflow-core/src/airflow/models/dagbag.py +++ b/airflow-core/src/airflow/models/dagbag.py @@ -83,7 +83,13 @@ def __init__( :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. """ + 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 @@ -92,8 +98,8 @@ def __init__( # 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 = max(cache_size or 0, 0) - ttl = max(cache_ttl or 0, 0) + 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 diff --git a/airflow-core/tests/unit/api_fastapi/common/test_dagbag.py b/airflow-core/tests/unit/api_fastapi/common/test_dagbag.py index 4b20025a8227e..cc31f6288a78f 100644 --- a/airflow-core/tests/unit/api_fastapi/common/test_dagbag.py +++ b/airflow-core/tests/unit/api_fastapi/common/test_dagbag.py @@ -108,25 +108,3 @@ def test_create_dag_bag_cache_modes(self, cache_size, cache_ttl, expected_dags_t 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", - "dag_cache_size must be greater than or equal to 0", - id="negative_size", - ), - pytest.param( - "64", - "-1", - "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=expected_message): - create_dag_bag() diff --git a/airflow-core/tests/unit/models/test_dagbag.py b/airflow-core/tests/unit/models/test_dagbag.py index 10a4c8b4c80b8..8dbc35ea385be 100644 --- a/airflow-core/tests/unit/models/test_dagbag.py +++ b/airflow-core/tests/unit/models/test_dagbag.py @@ -251,7 +251,6 @@ class TestDBDagBagCache: ("cache_size", "cache_ttl", "expected_type", "expected_maxsize"), [ pytest.param(None, None, dict, None, id="neither_plain_dict"), - pytest.param(-1, -1, dict, None, id="negatives_clamped_to_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"), @@ -264,6 +263,27 @@ def test_cache_selection(self, cache_size, cache_ttl, expected_type, expected_ma 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.""" dag_bag = DBDagBag(cache_size=10, cache_ttl=60) From b483b79b1b23dc900bfdd27ae9756323ab3dd364 Mon Sep 17 00:00:00 2001 From: LIU ZHE YOU Date: Wed, 19 Aug 2026 11:05:58 +0000 Subject: [PATCH 4/5] Clarify cache validation context and docs wording Context-specific errors help operators identify invalid API settings, while defensive constructor checks protect other callers and documentation remains compatible with the spellchecker. --- .../web-stack.rst | 6 ++--- airflow-core/docs/faq.rst | 6 ++--- .../src/airflow/api_fastapi/common/dagbag.py | 5 ++++ .../src/airflow/config_templates/config.yml | 4 ++-- airflow-core/src/airflow/models/dagbag.py | 1 + .../unit/api_fastapi/common/test_dagbag.py | 23 +++++++++++++++++++ 6 files changed, 37 insertions(+), 8 deletions(-) diff --git a/airflow-core/docs/administration-and-deployment/web-stack.rst b/airflow-core/docs/administration-and-deployment/web-stack.rst index f1a9c8eaf78c5..f358664f55029 100644 --- a/airflow-core/docs/administration-and-deployment/web-stack.rst +++ b/airflow-core/docs/administration-and-deployment/web-stack.rst @@ -191,9 +191,9 @@ For example, to trigger a rolling restart of the API server pods: 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 when the entry is revalidated 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 revalidations. +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 diff --git a/airflow-core/docs/faq.rst b/airflow-core/docs/faq.rst index 93292ead8e291..46c783317db92 100644 --- a/airflow-core/docs/faq.rst +++ b/airflow-core/docs/faq.rst @@ -729,9 +729,9 @@ this in the ``[api]`` section: 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 revalidated after ``[core] min_serialized_dag_update_interval``, not on every request. -With a shorter TTL, even frequently requested entries can expire and reload between -revalidations. Setting both options to 0 uses an unbounded dict with no eviction, matching 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 diff --git a/airflow-core/src/airflow/api_fastapi/common/dagbag.py b/airflow-core/src/airflow/api_fastapi/common/dagbag.py index 94815ed096b35..85ce253fcb8f2 100644 --- a/airflow-core/src/airflow/api_fastapi/common/dagbag.py +++ b/airflow-core/src/airflow/api_fastapi/common/dagbag.py @@ -35,6 +35,11 @@ def create_dag_bag() -> DBDagBag: cache_size = conf.getint("api", "dag_cache_size", fallback=64) cache_ttl = conf.getint("api", "dag_cache_ttl", fallback=3600) + if cache_size < 0: + 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) diff --git a/airflow-core/src/airflow/config_templates/config.yml b/airflow-core/src/airflow/config_templates/config.yml index 9059d378158e7..49ae0c1f3090d 100644 --- a/airflow-core/src/airflow/config_templates/config.yml +++ b/airflow-core/src/airflow/config_templates/config.yml @@ -1734,10 +1734,10 @@ api: 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 revalidated after + 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 - revalidations. ``dag_cache_size`` remains the only hard ceiling on memory. + 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 diff --git a/airflow-core/src/airflow/models/dagbag.py b/airflow-core/src/airflow/models/dagbag.py index e54df53ceee8b..dfa336f2fe0a5 100644 --- a/airflow-core/src/airflow/models/dagbag.py +++ b/airflow-core/src/airflow/models/dagbag.py @@ -85,6 +85,7 @@ def __init__( 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: diff --git a/airflow-core/tests/unit/api_fastapi/common/test_dagbag.py b/airflow-core/tests/unit/api_fastapi/common/test_dagbag.py index cc31f6288a78f..d56e96a24a3da 100644 --- a/airflow-core/tests/unit/api_fastapi/common/test_dagbag.py +++ b/airflow-core/tests/unit/api_fastapi/common/test_dagbag.py @@ -17,6 +17,7 @@ from __future__ import annotations import math +import re from unittest import mock import pytest @@ -108,3 +109,25 @@ def test_create_dag_bag_cache_modes(self, cache_size, cache_ttl, expected_dags_t 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() From a21020f68b2b79906237d1a18142fb4fc52e543e Mon Sep 17 00:00:00 2001 From: LIU ZHE YOU Date: Wed, 19 Aug 2026 12:20:06 +0000 Subject: [PATCH 5/5] Clarify API cache TTL documentation layout --- .../docs/administration-and-deployment/web-stack.rst | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/airflow-core/docs/administration-and-deployment/web-stack.rst b/airflow-core/docs/administration-and-deployment/web-stack.rst index f358664f55029..ed8264efc4be4 100644 --- a/airflow-core/docs/administration-and-deployment/web-stack.rst +++ b/airflow-core/docs/administration-and-deployment/web-stack.rst @@ -191,9 +191,8 @@ For example, to trigger a rolling restart of the API server pods: 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. +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