fix(lock): reject stale lease snapshots (#6249)

crates/lock/src/fast_lock/manager.rs

@@ -100,7 +100,7 @@ impl FastObjectLockManager {
             Ok(()) => {
                 let guard = FastLockGuard::new(request.key, request.mode, request.owner, shard.clone());
                 // Register guard to prevent premature cleanup
-                shard.register_guard(guard.guard_id());
+                shard.register_guard_with_info(guard.guard_id(), guard.key(), guard.mode(), guard.owner());
                 Ok(guard)
             }
             Err(err) => Err(err),
@@ -223,7 +223,7 @@ impl FastObjectLockManager {
 
                 if acquired {
                     let guard = FastLockGuard::new(key.clone(), mode, owner.clone(), shard.clone());
-                    shard.register_guard(guard.guard_id());
+                    shard.register_guard_with_info(guard.guard_id(), guard.key(), guard.mode(), guard.owner());
                     all_successful.push(key);
                     guards.push(guard);
                 }
@@ -252,7 +252,7 @@ impl FastObjectLockManager {
                 match shard.acquire_lock(request).await {
                     Ok(()) => {
                         let guard = FastLockGuard::new(request.key.clone(), request.mode, request.owner.clone(), shard.clone());
-                        shard.register_guard(guard.guard_id());
+                        shard.register_guard_with_info(guard.guard_id(), guard.key(), guard.mode(), guard.owner());
                         acquired_guards.push(guard);
                     }
                     Err(err) => {
@@ -310,6 +310,15 @@ impl FastObjectLockManager {
         infos
     }
 
+    /// Enumerate held locks with holder counts and stable holder identities.
+    pub fn list_locks_with_holder_generations(&self) -> Vec<(crate::fast_lock::types::ObjectLockInfo, u32, Option<Vec<u64>>)> {
+        let mut infos = Vec::new();
+        for shard in &self.shards {
+            infos.extend(shard.list_locks_with_holder_generations());
+        }
+        infos
+    }
+
     /// Force-release every holder of the lock on `key`.
     ///
     /// Returns the number of owners released (0 if the resource was not locked).
@@ -556,15 +565,15 @@ mod tests {
         let write_key = ObjectKey::new("bucket", "write-object");
         let read_key = ObjectKey::new("bucket", "read-object");
 
-        let _write_guard = manager
+        let write_guard = manager
             .acquire_write_lock(write_key.clone(), "writer")
             .await
             .expect("write lock should acquire");
-        let _read_guard = manager
+        let read_guard = manager
             .acquire_read_lock(read_key.clone(), "reader")
             .await
             .expect("read lock should acquire");
-        let _second_read_guard = manager
+        let second_read_guard = manager
             .acquire_read_lock(read_key.clone(), "reader")
             .await
             .expect("second read lock should acquire");
@@ -593,6 +602,30 @@ mod tests {
             .expect("write holder count listed");
         assert_eq!(*write_holder_count, 1);
 
+        let generations = manager.list_locks_with_holder_generations();
+        let (_, _, read_generations) = generations
+            .iter()
+            .find(|(info, _, _)| info.key == read_key)
+            .expect("read holder generations listed");
+        let mut expected_read_generations = vec![read_guard.guard_id(), second_read_guard.guard_id()];
+        expected_read_generations.sort_unstable();
+        assert_eq!(read_generations.as_ref(), Some(&expected_read_generations));
+
+        let (_, _, write_generations) = generations
+            .iter()
+            .find(|(info, _, _)| info.key == write_key)
+            .expect("write holder generation listed");
+        assert_eq!(write_generations.as_ref(), Some(&vec![write_guard.guard_id()]));
+
+        drop(read_guard);
+        let remaining = manager.list_locks_with_holder_generations();
+        let (_, remaining_count, remaining_generations) = remaining
+            .iter()
+            .find(|(info, _, _)| info.key == read_key)
+            .expect("remaining read holder generation listed");
+        assert_eq!(*remaining_count, 1);
+        assert_eq!(remaining_generations.as_ref(), Some(&vec![second_read_guard.guard_id()]));
+
         manager.shutdown().await;
     }
 

crates/lock/src/fast_lock/shard.rs

@@ -24,7 +24,20 @@ use crate::fast_lock::{
     state::ObjectLockState,
     types::{LockMode, LockResult, ObjectKey, ObjectLockRequest},
 };
-use std::collections::HashSet;
+
+#[derive(Debug)]
+struct ActiveGuardInfo {
+    key: ObjectKey,
+    mode: LockMode,
+    owner: Arc<str>,
+}
+
+#[derive(Debug, PartialEq, Eq, Hash)]
+struct GuardHolderKey {
+    key: ObjectKey,
+    mode: LockMode,
+    owner: Arc<str>,
+}
 
 /// Lock shard to reduce global contention
 #[derive(Debug)]
@@ -38,7 +51,7 @@ pub struct LockShard {
     /// Shard ID for debugging
     _shard_id: usize,
     /// Active guard IDs to prevent cleanup of locks with live guards
-    active_guards: parking_lot::Mutex<HashSet<u64>>,
+    active_guards: parking_lot::Mutex<HashMap<u64, Option<ActiveGuardInfo>>>,
 }
 
 /// Cancellation-safe waiter counter ticket.
@@ -84,7 +97,7 @@ impl LockShard {
             object_pool: ObjectStatePool::new(),
             metrics: ShardMetrics::new(),
             _shard_id: shard_id,
-            active_guards: parking_lot::Mutex::new(HashSet::new()),
+            active_guards: parking_lot::Mutex::new(HashMap::new()),
         }
     }
 
@@ -327,7 +340,7 @@ impl LockShard {
         // First, try to remove the guard from active set
         let guard_was_active = {
             let mut guards = self.active_guards.lock();
-            guards.remove(&guard_id)
+            guards.remove(&guard_id).is_some()
         };
 
         // If guard was not active, this is a double-release attempt
@@ -375,8 +388,19 @@ impl LockShard {
 
     /// Register a guard to prevent premature cleanup
     pub fn register_guard(&self, guard_id: u64) {
+        self.active_guards.lock().insert(guard_id, None);
+    }
+
+    pub(crate) fn register_guard_with_info(&self, guard_id: u64, key: &ObjectKey, mode: LockMode, owner: &Arc<str>) {
         let mut guards = self.active_guards.lock();
-        guards.insert(guard_id);
+        guards.insert(
+            guard_id,
+            Some(ActiveGuardInfo {
+                key: key.clone(),
+                mode,
+                owner: owner.clone(),
+            }),
+        );
     }
 
     /// Unregister a guard (called when guard is dropped)
@@ -396,7 +420,7 @@ impl LockShard {
     #[cfg(test)]
     pub fn is_guard_active(&self, guard_id: u64) -> bool {
         let guards = self.active_guards.lock();
-        guards.contains(&guard_id)
+        guards.contains_key(&guard_id)
     }
 
     /// Calculate adaptive timeout based on current system load and request priority
@@ -602,6 +626,50 @@ impl LockShard {
         infos
     }
 
+    pub(crate) fn list_locks_with_holder_generations(
+        &self,
+    ) -> Vec<(crate::fast_lock::types::ObjectLockInfo, u32, Option<Vec<u64>>)> {
+        // Snapshot lock state before guard registrations. Acquires register after
+        // mutating state, while releases unregister before mutating state, so a
+        // concurrent transition can only make the cohort mismatch and fall back.
+        let infos = self.list_locks_with_holder_counts();
+        let guards = self.active_guards.lock();
+        let mut guard_ids_by_holder: HashMap<GuardHolderKey, Vec<u64>> = HashMap::with_capacity(guards.len());
+        for (&guard_id, guard) in guards
+            .iter()
+            .filter_map(|(guard_id, guard)| guard.as_ref().map(|guard| (guard_id, guard)))
+        {
+            let key = GuardHolderKey {
+                key: guard.key.clone(),
+                mode: guard.mode,
+                owner: guard.owner.clone(),
+            };
+            guard_ids_by_holder
+                .entry(key)
+                .and_modify(|guard_ids| guard_ids.push(guard_id))
+                .or_insert_with(|| vec![guard_id]);
+        }
+        drop(guards);
+        for guard_ids in guard_ids_by_holder.values_mut() {
+            guard_ids.sort_unstable();
+        }
+
+        infos
+            .into_iter()
+            .map(|(info, holder_count)| {
+                let key = GuardHolderKey {
+                    key: info.key.clone(),
+                    mode: info.mode,
+                    owner: info.owner.clone(),
+                };
+                let generation = guard_ids_by_holder
+                    .remove(&key)
+                    .filter(|guard_ids| u32::try_from(guard_ids.len()).ok() == Some(holder_count));
+                (info, holder_count, generation)
+            })
+            .collect()
+    }
+
     /// Force-release every holder of a lock on `key`, regardless of owner.
     ///
     /// Returns the number of owners that were released. Used by the admin

crates/lock/src/fast_lock/types.rs

@@ -257,7 +257,7 @@ impl std::fmt::Display for ObjectKey {
 }
 
 /// Lock type for object operations
-#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
 pub enum LockMode {
     /// Shared lock for read operations
     Shared,