From d108c81969f7a8a266f92afc5786cd321c96f9b5 Mon Sep 17 00:00:00 2001 From: viiccwen Date: Mon, 13 Jul 2026 15:32:24 +0800 Subject: [PATCH 1/2] Speed up repeated latest Dag lookups API requests repeatedly resolve the latest Dag and should benefit from the configured cache without sacrificing freshness. Signed-off-by: viiccwen --- airflow-core/docs/faq.rst | 5 ++- .../src/airflow/config_templates/config.yml | 13 +++--- airflow-core/src/airflow/models/dagbag.py | 18 ++++++++ airflow-core/tests/unit/models/test_dagbag.py | 44 +++++++++++++++++++ 4 files changed, 71 insertions(+), 9 deletions(-) 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..9302567186cb3 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. From 7c7d119b3adf1ff8d84d15db83defc929ad08dd4 Mon Sep 17 00:00:00 2001 From: Vic Wen Date: Tue, 14 Jul 2026 14:18:34 +0800 Subject: [PATCH 2/2] Potential fix for pull request finding Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- airflow-core/src/airflow/config_templates/config.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/airflow-core/src/airflow/config_templates/config.yml b/airflow-core/src/airflow/config_templates/config.yml index 9302567186cb3..963dbbf5c8de9 100644 --- a/airflow-core/src/airflow/config_templates/config.yml +++ b/airflow-core/src/airflow/config_templates/config.yml @@ -1670,7 +1670,7 @@ api: Set to 0 to use an unbounded dict (no eviction, matching pre-3.2 behavior). 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. + its cached deserialized Dag when the version ID and hash are unchanged. version_added: 3.3.0 type: integer example: ~