Fix: Update updated_ts for all demoted snapshtos and not just the ones representing removed models (#1595)

Author: izeigerman

sqlmesh/core/state_sync/common.py

@@ -133,6 +133,11 @@ def promote(self, environment: Environment, no_gaps: bool = False) -> PromotionR
                 table_info.name: table_info
                 for table_info in existing_environment.promoted_snapshots
             }
+
+            demoted_snapshots = set(existing_environment.snapshots) - set(environment.snapshots)
+            for demoted_snapshot in self._get_snapshots(demoted_snapshots).values():
+                # Update the updated_at attribute.
+                self._update_snapshot(demoted_snapshot)
         else:
             existing_table_infos = {}
 
@@ -156,10 +161,6 @@ def promote(self, environment: Environment, no_gaps: bool = False) -> PromotionR
             if environment_suffix_target_changed
             else [existing_table_infos[name] for name in missing_models]
         )
-        if removed and not environment_suffix_target_changed:
-            for removed_snapshot in self._get_snapshots(removed).values():
-                # Update the updated_at attribute.
-                self._update_snapshot(removed_snapshot)
 
         return PromotionResult(
             added=sorted(table_infos),

tests/core/test_state_sync.py

@@ -411,6 +411,16 @@ def test_promote_snapshots(state_sync: EngineAdapterStateSync, make_snapshot: t.
     )
     snapshot_a.categorize_as(SnapshotChangeCategory.BREAKING)
 
+    snapshot_b_old = make_snapshot(
+        SqlModel(
+            name="b",
+            kind=FullKind(),
+            query=parse_one("select 2 from a"),
+        ),
+        nodes={"a": snapshot_a.model},
+    )
+    snapshot_b_old.categorize_as(SnapshotChangeCategory.BREAKING)
+
     snapshot_b = make_snapshot(
         SqlModel(
             name="b",
@@ -435,28 +445,29 @@ def test_promote_snapshots(state_sync: EngineAdapterStateSync, make_snapshot: t.
     ):
         promote_snapshots(state_sync, [snapshot_a], "prod")
 
-    state_sync.push_snapshots([snapshot_a, snapshot_b, snapshot_c])
+    state_sync.push_snapshots([snapshot_a, snapshot_b_old, snapshot_b, snapshot_c])
 
-    promotion_result = promote_snapshots(state_sync, [snapshot_a, snapshot_b], "prod")
+    promotion_result = promote_snapshots(state_sync, [snapshot_a, snapshot_b_old], "prod")
 
-    assert set(promotion_result.added) == set([snapshot_a.table_info, snapshot_b.table_info])
+    assert set(promotion_result.added) == set([snapshot_a.table_info, snapshot_b_old.table_info])
     assert not promotion_result.removed
     assert not promotion_result.removed_environment_naming_info
     promotion_result = promote_snapshots(
         state_sync,
-        [snapshot_a, snapshot_b, snapshot_c],
+        [snapshot_a, snapshot_b_old, snapshot_c],
         "prod",
     )
     assert set(promotion_result.added) == set(
         [
             snapshot_a.table_info,
-            snapshot_b.table_info,
+            snapshot_b_old.table_info,
             snapshot_c.table_info,
         ]
     )
     assert not promotion_result.removed
     assert not promotion_result.removed_environment_naming_info
 
+    prev_snapshot_b_old_updated_ts = snapshot_b_old.updated_ts
     prev_snapshot_c_updated_ts = snapshot_c.updated_ts
 
     promotion_result = promote_snapshots(
@@ -472,6 +483,10 @@ def test_promote_snapshots(state_sync: EngineAdapterStateSync, make_snapshot: t.
         state_sync.get_snapshots([snapshot_c])[snapshot_c.snapshot_id].updated_ts
         > prev_snapshot_c_updated_ts
     )
+    assert (
+        state_sync.get_snapshots([snapshot_b_old])[snapshot_b_old.snapshot_id].updated_ts
+        > prev_snapshot_b_old_updated_ts
+    )
 
     snapshot_d = make_snapshot(
         SqlModel(