fix: address code review findings in invalidate/cleanup-snapshots

- C1/M4: eliminate TOCTOU race in delete_snapshots_for_environment by
  calling state_sync.delete_snapshots(batch.expired_snapshot_ids) directly
  instead of re-querying via delete_expired_snapshots, so physical drops
  and state removal operate on the same snapshot ID set
- M1: remove always-truthy `if target_conditions:` guard in
  get_expired_snapshots (snapshot_id_filter always yields >= 1 condition)
- M2: when cleanup_snapshots=True and the environment does not exist, log
  a warning and return early instead of printing a misleading success message
- m1: unconditionally initialize target_snapshot_ids before the
  cleanup_snapshots block to prevent potential UnboundLocalError
- n1: enforce `sync = sync or cleanup_snapshots` explicitly so the
  implication is in code, not just docs; update docstring and CLI help
  to say "cleanup runs synchronously" instead of "Implies --sync"

Author: mday-io

sqlmesh/cli/main.py

@@ -623,7 +623,7 @@ def run(ctx: click.Context, environment: t.Optional[str] = None, **kwargs: t.Any
 @click.option(
     "--cleanup-snapshots",
     is_flag=True,
-    help="After invalidating, immediately delete physical snapshot tables that are exclusively owned by this environment (not referenced by any other environment). Implies --sync.",
+    help="After invalidating, immediately delete physical snapshot tables that are exclusively owned by this environment (not referenced by any other environment). Cleanup runs synchronously regardless of --sync.",
 )
 @click.pass_context
 @error_handler

sqlmesh/core/context.py

@@ -99,6 +99,7 @@
     Snapshot,
     SnapshotEvaluator,
     SnapshotFingerprint,
+    SnapshotId,
     missing_intervals,
     to_table_mapping,
 )
@@ -1849,18 +1850,24 @@ def invalidate_environment(
             sync: If True, the call blocks until the environment is deleted. Otherwise, the environment will
                 be deleted asynchronously by the janitor process.
             cleanup_snapshots: If True, immediately deletes physical snapshot tables that are exclusively
-                owned by this environment (not referenced by any other environment). Implies sync=True.
+                owned by this environment (not referenced by any other environment). Cleanup runs
+                synchronously regardless of --sync.
         """
         name = Environment.sanitize_name(name)
+        sync = sync or cleanup_snapshots
 
+        target_snapshot_ids: t.Set[SnapshotId] = set()
         if cleanup_snapshots:
             # Capture snapshot IDs before invalidation so we can scope the cleanup afterwards.
             env = self.state_sync.get_environment(name)
-            target_snapshot_ids = {s.snapshot_id for s in env.snapshots} if env else set()
+            if env is None:
+                logger.warning("Environment '%s' does not exist; skipping snapshot cleanup.", name)
+                return
+            target_snapshot_ids = {s.snapshot_id for s in env.snapshots}
 
         self.state_sync.invalidate_environment(name)
 
-        if sync or cleanup_snapshots:
+        if sync:
             self._cleanup_environments(name=name)
             if cleanup_snapshots and target_snapshot_ids:
                 failures = delete_snapshots_for_environment(

sqlmesh/core/janitor.py

@@ -251,11 +251,7 @@ def delete_snapshots_for_environment(
 
     if cleanup_succeeded or force_delete:
         try:
-            state_sync.delete_expired_snapshots(
-                batch_range=ExpiredBatchRange.all_batch_range(),
-                ignore_ttl=True,
-                target_snapshot_ids=target_snapshot_ids,
-            )
+            state_sync.delete_snapshots(batch.expired_snapshot_ids)
             logger.info(
                 "Cleaned up %s snapshots from invalidated environment",
                 len(batch.expired_snapshot_ids),

sqlmesh/core/state_sync/db/snapshot.py

@@ -189,8 +189,7 @@ def get_expired_snapshots(
                     batch_size=self.SNAPSHOT_BATCH_SIZE,
                 )
             )
-            if target_conditions:
-                expired_query = expired_query.where(exp.or_(*target_conditions))
+            expired_query = expired_query.where(exp.or_(*target_conditions))
 
         expired_query = expired_query.where(batch_range.where_filter)