This is an automated email from the ASF dual-hosted git repository.
jason810496 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new f63d32129fe Honor the API server Dag cache TTL when no size limit is
set (#71814)
f63d32129fe is described below
commit f63d32129fe90321d4c81e96559785032a6db8f3
Author: Jason(Zhe-You) Liu <[email protected]>
AuthorDate: Wed Aug 19 23:32:10 2026 +0800
Honor the API server Dag cache TTL when no size limit is set (#71814)
* 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.
* 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.
* Reject negative cache values for every DBDagBag caller
Validating at the shared boundary prevents non-API callers from silently
clamping invalid cache settings.
* 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.
* Clarify API cache TTL documentation layout
---
.../administration-and-deployment/web-stack.rst | 12 +++--
airflow-core/docs/faq.rst | 18 ++++---
airflow-core/newsfragments/71814.bugfix.rst | 1 +
.../src/airflow/api_fastapi/common/dagbag.py | 20 ++------
.../src/airflow/config_templates/config.yml | 33 ++++++++----
airflow-core/src/airflow/models/dagbag.py | 38 +++++++++-----
.../tests/unit/api_fastapi/common/test_dagbag.py | 55 ++++++++++++++------
airflow-core/tests/unit/models/test_dagbag.py | 60 +++++++++++++---------
8 files changed, 146 insertions(+), 91 deletions(-)
diff --git a/airflow-core/docs/administration-and-deployment/web-stack.rst
b/airflow-core/docs/administration-and-deployment/web-stack.rst
index f043789d5e5..ed8264efc4b 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``: 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
diff --git a/airflow-core/docs/faq.rst b/airflow-core/docs/faq.rst
index ce89ecac393..46c783317db 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 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``).
@@ -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/newsfragments/71814.bugfix.rst
b/airflow-core/newsfragments/71814.bugfix.rst
new file mode 100644
index 00000000000..220e3141d3d
--- /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 d87aca49a52..85ce253fcb8 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,27 +29,16 @@ 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)
+ 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)
diff --git a/airflow-core/src/airflow/config_templates/config.yml
b/airflow-core/src/airflow/config_templates/config.yml
index 6d2438fd213..49ae0c1f309 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.
+
+ 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"
diff --git a/airflow-core/src/airflow/models/dagbag.py
b/airflow-core/src/airflow/models/dagbag.py
index 5c0556974ff..dfa336f2fe0 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,26 +80,37 @@ class DBDagBag:
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:
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 48c6f706ba7..d56e96a24a3 100644
--- a/airflow-core/tests/unit/api_fastapi/common/test_dagbag.py
+++ b/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()
diff --git a/airflow-core/tests/unit/models/test_dagbag.py
b/airflow-core/tests/unit/models/test_dagbag.py
index 79668d4fe54..8dbc35ea385 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,42 @@ class TestDBDagBag:
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."""
