fix: start the snapshot cache TTL after the fetch completes

sqlmesh/core/state_sync/cache.py

@@ -65,7 +65,12 @@ def get_snapshots(
                 existing[snapshot_id] = snapshot
 
         if missing:
-            existing.update(self.state_sync.get_snapshots(missing))
+            fetched = self.state_sync.get_snapshots(missing)
+            # The fetch can take longer than the TTL, so the TTL of what it returned starts now.
+            expire_at = now_timestamp() + self.ttl * 1000
+            for snapshot_id in missing.difference(fetched):
+                self.snapshot_cache[snapshot_id] = (False, expire_at)
+            existing.update(fetched)
 
         for snapshot_id, snapshot in existing.items():
             cached = self._from_cache(snapshot_id, now)

tests/core/state_sync/test_state_sync.py

@@ -3555,6 +3555,32 @@ def test_max_interval_end_per_model_ensure_finalized_snapshots(
     assert state_sync.max_interval_end_per_model(environment_name, set()) == {}
 
 
+def test_cache_ttl_starts_after_slow_fetch(mocker, make_snapshot):
+    snapshot = make_snapshot(SqlModel(name="a", query=parse_one("select 1")))
+    missing = make_snapshot(SqlModel(name="b", query=parse_one("select 2")))
+
+    start = to_timestamp("2023-01-01 00:00:00")
+    clock = [start]
+    mocker.patch("sqlmesh.core.state_sync.cache.now_timestamp", side_effect=lambda: clock[0])
+
+    def slow_get_snapshots(snapshot_ids):
+        clock[0] += 11 * 1000
+        return {snapshot.snapshot_id: snapshot}
+
+    state_sync = mocker.MagicMock()
+    state_sync.get_snapshots.side_effect = slow_get_snapshots
+    cache = CachingStateSync(state_sync, ttl=10)
+
+    expected = {snapshot.snapshot_id: snapshot}
+    assert cache.get_snapshots([snapshot, missing]) == expected
+    assert cache.get_snapshots([snapshot, missing]) == expected
+    assert state_sync.get_snapshots.call_count == 1
+
+    clock[0] += 11 * 1000
+    assert cache.get_snapshots([snapshot, missing]) == expected
+    assert state_sync.get_snapshots.call_count == 2
+
+
 def test_get_snapshots(mocker):
     mock = mocker.MagicMock()
     cache = CachingStateSync(mock)