Speed up repeated latest Dag lookups

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.

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: ~

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)
 
 

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.