diff --git a/airflow-core/docs/faq.rst b/airflow-core/docs/faq.rst index 6860fc72f9177..71bea0d2bb49a 100644 --- a/airflow-core/docs/faq.rst +++ b/airflow-core/docs/faq.rst @@ -728,8 +728,9 @@ in memory. Configure this in the ``[api]`` section: 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) -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``). +The cache is keyed by Dag version ID. Cached entries are periodically checked against the +database according to ``[core] min_serialized_dag_update_interval``. The ``dag_cache_ttl`` setting +controls eviction and memory usage independently of this freshness check. See :ref:`config:api__dag_cache_size` and :ref:`config:api__dag_cache_ttl` for the full configuration reference. diff --git a/airflow-core/src/airflow/config_templates/config.yml b/airflow-core/src/airflow/config_templates/config.yml index 545cded21d71d..963dbbf5c8de9 100644 --- a/airflow-core/src/airflow/config_templates/config.yml +++ b/airflow-core/src/airflow/config_templates/config.yml @@ -1668,10 +1668,9 @@ api: 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). - 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. + The cache is keyed by Dag version ID. Lookups by Dag ID + (e.g., viewing a Dag's details) fetch the latest serialized row, but reuse + its cached deserialized Dag when the version ID and hash are unchanged. version_added: 3.3.0 type: integer example: ~ @@ -1682,9 +1681,9 @@ api: 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. + Cached entries are periodically checked against the database according to + ``[core] min_serialized_dag_update_interval``. This limits staleness independently + of the TTL, which controls eviction and memory usage. version_added: 3.3.0 type: integer example: ~ diff --git a/airflow-core/src/airflow/models/dagbag.py b/airflow-core/src/airflow/models/dagbag.py index c4bd8eceea102..1a7dd0021bf03 100644 --- a/airflow-core/src/airflow/models/dagbag.py +++ b/airflow-core/src/airflow/models/dagbag.py @@ -240,6 +240,24 @@ def get_latest_version_of_dag(self, dag_id: str, *, session: Session) -> Seriali if not (serdag := SerializedDagModel.get(dag_id, session=session)): return None + + with self._lock: + cached = self._dags.get(serdag.dag_version_id) + if cached is not None: + # The latest-row query already loaded dag_hash, so no revalidation query is needed. + if cached.dag_hash == serdag.dag_hash: + self._dags[serdag.dag_version_id] = cached._replace(last_validated=time.monotonic()) + else: + # A serialized Dag can be updated in place without changing its version ID. + self._dags.pop(serdag.dag_version_id, None) + cached = None + + if cached is not None: + if self._use_cache: + stats.incr("api_server.dag_bag.cache_hit") + return cached.dag + if self._use_cache: + stats.incr("api_server.dag_bag.cache_miss") return self._read_dag(serdag) diff --git a/airflow-core/tests/unit/models/test_dagbag.py b/airflow-core/tests/unit/models/test_dagbag.py index 79668d4fe54f4..70158730045ff 100644 --- a/airflow-core/tests/unit/models/test_dagbag.py +++ b/airflow-core/tests/unit/models/test_dagbag.py @@ -182,6 +182,50 @@ def test_get_dag_returns_none_when_not_found(self): assert result is None + @patch("airflow.models.dagbag.stats") + @patch.object(SerializedDagModel, "get", autospec=True) + def test_get_latest_version_of_dag_returns_cached_dag(self, mock_get, mock_stats): + """Reuse the cached deserialized Dag when the version and hash match.""" + dag_bag = DBDagBag(cache_size=10, cache_ttl=60) + cached_dag = MagicMock(spec=SerializedDAG) + dag_bag._dags["v1"] = _CacheEntry(cached_dag, "hash1", 0.0) + mock_serdag = MagicMock(spec=SerializedDagModel) + # A different Dag proves the result came from the cache without deserialization. + mock_serdag.dag = MagicMock(spec=SerializedDAG) + mock_serdag.dag_version_id = "v1" + mock_serdag.dag_hash = "hash1" + mock_get.return_value = mock_serdag + + result = dag_bag.get_latest_version_of_dag("test_dag", session=self.session) + + assert result is cached_dag + mock_get.assert_called_once_with("test_dag", session=self.session) + self.session.get.assert_not_called() + assert dag_bag._dags["v1"].last_validated > 0.0 + mock_stats.incr.assert_called_once_with("api_server.dag_bag.cache_hit") + + @pytest.mark.parametrize("cached_hash", [None, "old_hash"], ids=["not-cached", "hash-changed"]) + @patch("airflow.models.dagbag.stats") + @patch.object(SerializedDagModel, "get", autospec=True) + def test_get_latest_version_of_dag_emits_cache_miss(self, mock_get, mock_stats, cached_hash): + """Emit a miss when the latest Dag is absent from the cache or its hash changed.""" + dag_bag = DBDagBag(cache_size=10, cache_ttl=60) + fresh_dag = MagicMock(spec=SerializedDAG) + if cached_hash is not None: + dag_bag._dags["v1"] = _CacheEntry(MagicMock(spec=SerializedDAG), cached_hash, 0.0) + mock_serdag = MagicMock(spec=SerializedDagModel) + mock_serdag.dag = fresh_dag + mock_serdag.dag_version_id = "v1" + mock_serdag.dag_hash = "hash1" + mock_get.return_value = mock_serdag + + result = dag_bag.get_latest_version_of_dag("test_dag", session=self.session) + + assert result is fresh_dag + mock_get.assert_called_once_with("test_dag", session=self.session) + assert dag_bag._dags["v1"].dag is fresh_dag + mock_stats.incr.assert_called_once_with("api_server.dag_bag.cache_miss") + def test_get_dag_reflects_in_place_version_update_end_to_end(self): """End-to-end regression: an in-place version update must be re-read, not served stale.