perf(lock): avoid scanning live leases on every acquisition (#9356)

* fix(lock): allow more than 255 concurrent shared lock holders The fast-lock state packed the reader count into 8 bits, so the 256th concurrent shared holder of one key was refused. Bucket-scoped fences are read-locked by every in-flight request into the bucket, so busy buckets hit the cap. The count now has 16 bits. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> * test(lock): cover shared-reader capacity through namespace quorum Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> * perf(lock): skip lease scans until a shard deadline is due Keep a conservative earliest-expiry hint with the guard table. Healthy acquisitions and reaper passes avoid walking live leases, while due sweeps recompute the hint and release expired guards outside the registry lock. Preserve resource-specific retry counts and cover refresh, replacement, removal and cross-resource expiration. Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> --------- Co-authored-by: Rain Jiang <rain-jiang@outlook.com> Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com> Co-authored-by: heihutu <heihutu@gmail.com> Co-authored-by: zhi22915 <qiuzgang@gmail.com>
2026-11-13 13:29:46 +00:00 · 2026-11-12 23:21:21 +08:00
parent 55d9bf08a2
commit 078fc47e9c
3 changed files with 332 additions and 56 deletions
@@ -30,19 +30,92 @@ use crate::{
 const DEFAULT_GUARD_SHARD_COUNT: usize = 64;
 const EVENT_NAMESPACE_LOCK_ACQUIRE: &str = "namespace_lock_acquire";
 const EVENT_NAMESPACE_LOCK_RELEASE: &str = "namespace_lock_release";
+const EVENT_LOCK_GUARD_RECLAIMED: &str = "lock_guard_reclaimed";
 const LOG_COMPONENT_LOCK: &str = "lock";
 const LOG_SUBSYSTEM_NAMESPACE: &str = "namespace";

-type GuardMap = HashMap<LockId, LocalGuardEntry>;
 type GuardShard = Arc<GuardRegistry>;
 type GuardStorage = Arc<Vec<GuardShard>>;

+#[derive(Debug, Default)]
+struct GuardRegistryState {
+    entries: HashMap<LockId, LocalGuardEntry>,
+    // A lower bound on every registered lease deadline. Removal and refresh
+    // may leave it early, so a due sweep recomputes it from the surviving leases.
+    next_expiry: Option<Instant>,
+    #[cfg(test)]
+    expiry_scans: usize,
+}
+
+impl GuardRegistryState {
+    fn insert(&mut self, lock_id: LockId, entry: LocalGuardEntry) -> Option<LocalGuardEntry> {
+        self.next_expiry = Some(
+            self.next_expiry
+                .map_or(entry.deadline, |deadline| deadline.min(entry.deadline)),
+        );
+        self.entries.insert(lock_id, entry)
+    }
+
+    fn remove(&mut self, lock_id: &LockId) -> Option<LocalGuardEntry> {
+        let entry = self.entries.remove(lock_id);
+        if self.entries.is_empty() {
+            self.next_expiry = None;
+        }
+        entry
+    }
+
+    fn refresh(&mut self, lock_id: &LockId) -> bool {
+        let Some(entry) = self.entries.get_mut(lock_id) else {
+            return false;
+        };
+        if entry.is_expired() {
+            return false;
+        }
+        entry.refresh();
+        self.next_expiry = Some(
+            self.next_expiry
+                .map_or(entry.deadline, |deadline| deadline.min(entry.deadline)),
+        );
+        true
+    }
+
+    fn sweep_due(&self) -> bool {
+        self.next_expiry.is_some_and(|deadline| deadline <= Instant::now())
+    }
+
+    fn extract_expired(&mut self) -> Vec<LocalGuardEntry> {
+        if !self.sweep_due() {
+            return Vec::new();
+        }
+        #[cfg(test)]
+        {
+            self.expiry_scans += 1;
+        }
+        let now = Instant::now();
+        let mut next_expiry: Option<Instant> = None;
+        let expired = self
+            .entries
+            .extract_if(|_, entry| {
+                if entry.deadline <= now {
+                    true
+                } else {
+                    next_expiry = Some(next_expiry.map_or(entry.deadline, |deadline| deadline.min(entry.deadline)));
+                    false
+                }
+            })
+            .map(|(_, entry)| entry)
+            .collect();
+        self.next_expiry = next_expiry;
+        expired
+    }
+}
+
 /// Registry access never reserves the table for a task that has not resumed.
 /// Only short synchronous map operations hold the native lock; contention
 /// waits on a notification without blocking an executor thread.
 #[derive(Debug, Default)]
 struct GuardRegistry {
-    entries: RwLock<GuardMap>,
+    entries: RwLock<GuardRegistryState>,
    changed: Notify,
 }

@@ -83,7 +156,7 @@ impl GuardRegistry {
        }
    }

-    async fn read(&self) -> GuardRegistryAccess<'_, RwLockReadGuard<'_, GuardMap>> {
+    async fn read(&self) -> GuardRegistryAccess<'_, RwLockReadGuard<'_, GuardRegistryState>> {
        tokio::task::consume_budget().await;
        if let Some(guard) = self.entries.try_read() {
            return self.access(guard);
@@ -100,7 +173,7 @@ impl GuardRegistry {
        }
    }

-    async fn write(&self) -> GuardRegistryAccess<'_, RwLockWriteGuard<'_, GuardMap>> {
+    async fn write(&self) -> GuardRegistryAccess<'_, RwLockWriteGuard<'_, GuardRegistryState>> {
        tokio::task::consume_budget().await;
        if let Some(guard) = self.entries.try_write() {
            return self.access(guard);
@@ -284,8 +357,8 @@ impl LocalClient {
                            replaced_entry,
                            registration_wait,
                            mutation_started.elapsed(),
-                            guards.len(),
-                            guards.capacity(),
+                            guards.entries.len(),
+                            guards.entries.capacity(),
                        )
                    };
                    // Repeated shared requests can replace an entry. Releasing
@@ -415,8 +488,8 @@ impl LocalClient {
        self.manager.clone().unwrap_or_else(crate::get_global_lock_manager)
    }

-    /// Guard entries for one resource share a shard so expiration checks do not
-    /// need to take every shard's write lock on the acquisition hot path.
+    /// Guard entries for one resource share a shard so acquisition only checks
+    /// that shard's expiration deadline.
    fn get_resource_shard_index(&self, resource: &crate::ObjectKey) -> usize {
        resource.shard_index(self.shard_mask)
    }
@@ -429,46 +502,32 @@ impl LocalClient {

    async fn reclaim_expired_guards_for_resource(&self, resource: &crate::ObjectKey) -> usize {
        let shard = &self.guard_storage[self.get_resource_shard_index(resource)];
-        let has_expired = {
-            let guards = shard.read().await;
-            guards
-                .iter()
-                .any(|(lock_id, entry)| &lock_id.resource == resource && entry.is_expired())
-        };
-        if !has_expired {
+        if !shard.read().await.sweep_due() {
            return 0;
        }

        let expired_entries = {
            let mut guards = shard.write().await;
-            guards
-                .extract_if(|lock_id, entry| &lock_id.resource == resource && entry.is_expired())
-                .map(|(_, entry)| entry)
-                .collect::<Vec<_>>()
+            guards.extract_expired()
        };
-        Self::release_reclaimed_guards(expired_entries, Some(resource))
+        let reclaimed_for_resource = expired_entries.iter().filter(|entry| entry.guard.key() == resource).count();
+        Self::release_reclaimed_guards(expired_entries);
+        reclaimed_for_resource
    }

    async fn extract_expired_guards(storage: &GuardStorage) -> Vec<LocalGuardEntry> {
        let mut expired_entries = Vec::new();
        for shard in storage.iter() {
-            let has_expired = {
-                let guards = shard.read().await;
-                guards.values().any(LocalGuardEntry::is_expired)
-            };
-            if !has_expired {
+            if !shard.read().await.sweep_due() {
                continue;
            }
            let mut guards = shard.write().await;
-            expired_entries.extend(guards.extract_if(|_, entry| entry.is_expired()).map(|(_, entry)| entry));
+            expired_entries.extend(guards.extract_expired());
        }
        expired_entries
    }

-    fn release_reclaimed_guards(
-        entries: impl IntoIterator<Item = LocalGuardEntry>,
-        resource: Option<&crate::ObjectKey>,
-    ) -> usize {
+    fn release_reclaimed_guards(entries: impl IntoIterator<Item = LocalGuardEntry>) -> usize {
        let mut reclaimed = 0;
        for mut entry in entries {
            let _ = entry.guard.release();
@@ -476,11 +535,14 @@ impl LocalClient {
            reclaimed += 1;
        }
        if reclaimed > 0 {
-            if let Some(resource) = resource {
-                tracing::debug!(event = "lock_guard_reclaimed", resource = %resource, count = reclaimed, "expired lock guards reclaimed");
-            } else {
-                tracing::debug!(event = "lock_guard_reaper_sweep", count = reclaimed, "expired lock guards reclaimed");
-            }
+            tracing::debug!(
+                event = EVENT_LOCK_GUARD_RECLAIMED,
+                component = LOG_COMPONENT_LOCK,
+                subsystem = LOG_SUBSYSTEM_NAMESPACE,
+                state = "reclaimed",
+                count = reclaimed,
+                "Expired lock guards reclaimed"
+            );
        }
        reclaimed
    }
@@ -499,7 +561,7 @@ impl LocalClient {
                    break;
                };
                let expired_entries = Self::extract_expired_guards(&storage).await;
-                Self::release_reclaimed_guards(expired_entries, None);
+                Self::release_reclaimed_guards(expired_entries);
            }
        });
    }
@@ -585,19 +647,17 @@ impl LockClient for LocalClient {
        let shard = self.get_shard(lock_id);
        let expired_entry = {
            let mut guards = shard.write().await;
-            let Some(entry) = guards.get_mut(lock_id) else {
-                return Ok(false);
-            };
-            if entry.is_expired() {
-                guards.remove(lock_id)
-            } else {
-                entry.refresh();
+            if guards.refresh(lock_id) {
                None
+            } else if let Some(entry) = guards.remove(lock_id) {
+                Some(entry)
+            } else {
+                return Ok(false);
            }
        };

        if let Some(entry) = expired_entry {
-            Self::release_reclaimed_guards([entry], Some(&lock_id.resource));
+            Self::release_reclaimed_guards([entry]);
            Ok(false)
        } else {
            Ok(true)
@@ -611,7 +671,7 @@ impl LockClient for LocalClient {
    async fn check_status(&self, lock_id: &LockId) -> Result<Option<LockInfo>> {
        let shard = self.get_shard(lock_id);
        let guards = shard.read().await;
-        if let Some(entry) = guards.get(lock_id) {
+        if let Some(entry) = guards.entries.get(lock_id) {
            // We have an active guard for this lock
            let lock_type = match entry.guard.mode() {
                crate::LockMode::Shared => LockType::Shared,
@@ -644,8 +704,8 @@ impl LockClient for LocalClient {
        let mut leases = Vec::new();
        for shard in self.guard_storage.iter() {
            let guards = shard.read().await;
-            leases.reserve(guards.len());
-            leases.extend(guards.iter().map(|(lock_id, entry)| crate::LockLeaseInfo {
+            leases.reserve(guards.entries.len());
+            leases.extend(guards.entries.iter().map(|(lock_id, entry)| crate::LockLeaseInfo {
                resource: lock_id.resource.clone(),
                lock_type: match entry.guard.mode() {
                    crate::LockMode::Shared => LockType::Shared,
@@ -851,6 +911,139 @@ mod tests {
            .with_acquire_timeout(Duration::from_millis(80))
    }

+    #[tokio::test(start_paused = true)]
+    async fn unexpired_shared_acquisitions_and_reaper_do_not_scan_guard_entries() {
+        let client = LocalClient::with_manager(Arc::new(GlobalLockManager::new()));
+        client.reaper_started.store(true, Ordering::Release);
+        let resource = crate::ObjectKey::new("bucket", "unexpired-shared-registry");
+        let mut requests = Vec::with_capacity(512);
+        for _ in 0..512 {
+            let request = LockRequest::new(resource.clone(), LockType::Shared, "reader");
+            assert!(client.acquire_lock(&request).await.expect("register shared lease").success);
+            requests.push(request);
+        }
+        assert!(LocalClient::extract_expired_guards(&client.guard_storage).await.is_empty());
+        {
+            let registry = client.get_shard(&requests[0].lock_id).read().await;
+            assert_eq!(registry.entries.len(), 512);
+            assert_eq!(registry.expiry_scans, 0, "live lease acquisition and reaping must use the deadline hint");
+            assert!(registry.next_expiry.is_some());
+        }
+        for request in &requests {
+            assert!(client.release(&request.lock_id).await.expect("release shared lease"));
+        }
+        let registry = client.get_shard(&requests[0].lock_id).read().await;
+        assert!(registry.entries.is_empty());
+        assert_eq!(registry.next_expiry, None, "the final release must clear the deadline hint");
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn same_shard_expiration_sweep_only_counts_the_requested_resource() {
+        let manager = Arc::new(GlobalLockManager::new());
+        let client = LocalClient::with_manager(manager.clone());
+        client.reaper_started.store(true, Ordering::Release);
+        let resource = crate::ObjectKey::new("bucket", "live-sweep-trigger");
+        let shard_index = client.get_resource_shard_index(&resource);
+        let other_resource = (0..4096)
+            .map(|index| crate::ObjectKey::new("bucket", format!("expired-same-shard-{index}")))
+            .find(|candidate| client.get_resource_shard_index(candidate) == shard_index)
+            .expect("find an independent resource in the same guard shard");
+        let live = request(resource.clone(), "live", Duration::from_secs(30));
+        let expired = request(other_resource.clone(), "expired", Duration::from_secs(5));
+        assert!(client.acquire_lock(&live).await.expect("register live lease").success);
+        assert!(client.acquire_lock(&expired).await.expect("register earlier lease").success);
+        tokio::time::advance(Duration::from_secs(5)).await;
+
+        assert_eq!(client.reclaim_expired_guards_for_resource(&resource).await, 0);
+        assert!(
+            client
+                .check_status(&expired.lock_id)
+                .await
+                .expect("expired lease status")
+                .is_none()
+        );
+        assert!(client.check_status(&live.lock_id).await.expect("live lease status").is_some());
+        let replacement = manager
+            .acquire_lock(crate::ObjectLockRequest::new_write(other_resource, "replacement"))
+            .await
+            .expect("the sweep must release the other resource's expired physical guard");
+        drop(replacement);
+        assert_eq!(client.reclaim_expired_guards_for_resource(&resource).await, 0);
+        assert_eq!(client.get_shard(&live.lock_id).read().await.expiry_scans, 1);
+        assert!(client.release(&live.lock_id).await.expect("release live lease"));
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn refreshed_and_removed_earliest_leases_only_trigger_one_sweep_per_deadline() {
+        let client = LocalClient::with_manager(Arc::new(GlobalLockManager::new()));
+        client.reaper_started.store(true, Ordering::Release);
+        let resource = crate::ObjectKey::new("bucket", "refresh-expiry-hint");
+        let short = LockRequest::new(resource.clone(), LockType::Shared, "short").with_ttl(Duration::from_secs(10));
+        let long = LockRequest::new(resource.clone(), LockType::Shared, "long").with_ttl(Duration::from_secs(30));
+        assert!(client.acquire_lock(&short).await.expect("short lease").success);
+        assert!(client.acquire_lock(&long).await.expect("long lease").success);
+        tokio::time::advance(Duration::from_secs(5)).await;
+        assert!(client.refresh(&short.lock_id).await.expect("extend the earliest deadline"));
+        tokio::time::advance(Duration::from_secs(5)).await;
+        assert_eq!(client.reclaim_expired_guards_for_resource(&resource).await, 0);
+        assert_eq!(client.reclaim_expired_guards_for_resource(&resource).await, 0);
+        assert_eq!(client.get_shard(&short.lock_id).read().await.expiry_scans, 1);
+
+        assert!(
+            client
+                .release(&short.lock_id)
+                .await
+                .expect("remove the earliest surviving lease")
+        );
+        tokio::time::advance(Duration::from_secs(5)).await;
+        assert_eq!(client.reclaim_expired_guards_for_resource(&resource).await, 0);
+        assert_eq!(client.reclaim_expired_guards_for_resource(&resource).await, 0);
+        assert_eq!(client.get_shard(&long.lock_id).read().await.expiry_scans, 2);
+
+        tokio::time::advance(Duration::from_secs(15)).await;
+        assert_eq!(client.reclaim_expired_guards_for_resource(&resource).await, 1);
+        let registry = client.get_shard(&long.lock_id).read().await;
+        assert_eq!(registry.expiry_scans, 3);
+        assert!(registry.entries.is_empty());
+        assert_eq!(registry.next_expiry, None);
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn repeated_shared_id_replacement_preserves_earlier_expiration_and_reaper_cleanup() {
+        let manager = Arc::new(GlobalLockManager::new());
+        let client = LocalClient::with_manager(manager.clone());
+        client.reaper_started.store(true, Ordering::Release);
+        let resource = crate::ObjectKey::new("bucket", "replace-expiry-hint");
+        let original = LockRequest::new(resource.clone(), LockType::Shared, "owner").with_ttl(Duration::from_secs(30));
+        assert!(client.acquire_lock(&original).await.expect("register original lease").success);
+        let replacement = original.clone().with_ttl(Duration::from_secs(5));
+        assert!(
+            client
+                .acquire_lock(&replacement)
+                .await
+                .expect("replace the same grant identity")
+                .success
+        );
+        assert_eq!(client.get_shard(&original.lock_id).read().await.entries.len(), 1);
+        let backend = manager.as_fast_lock_manager().expect("enabled physical lock backend");
+        assert_eq!(backend.get_shard(&resource).active_guard_count(), 1);
+
+        tokio::time::advance(Duration::from_secs(5)).await;
+        let expired = LocalClient::extract_expired_guards(&client.guard_storage).await;
+        assert_eq!(LocalClient::release_reclaimed_guards(expired), 1);
+        assert_eq!(backend.get_shard(&resource).active_guard_count(), 0);
+        assert!(
+            client
+                .check_status(&original.lock_id)
+                .await
+                .expect("replaced lease status")
+                .is_none()
+        );
+        let registry = client.get_shard(&original.lock_id).read().await;
+        assert_eq!(registry.expiry_scans, 1);
+        assert_eq!(registry.next_expiry, None);
+    }
+
    fn hold_guard_shard(shard: GuardShard) -> (std::sync::mpsc::Sender<()>, std::thread::JoinHandle<()>) {
        let (entered_tx, entered_rx) = std::sync::mpsc::sync_channel(0);
        let (release_tx, release_rx) = std::sync::mpsc::channel();
@@ -1037,7 +1230,7 @@ mod tests {
        let access = timeout(Duration::from_millis(100), survivor)
            .await
            .expect("a cancelled waiter must not consume another waiter's unlock notification");
-        assert!(access.is_empty());
+        assert!(access.entries.is_empty());
    }

    #[tokio::test]
@@ -21,10 +21,14 @@ use crate::fast_lock::optimized_notify::OptimizedNotify;
 use crate::fast_lock::types::{LockMode, LockPriority};

 /// Optimized atomic lock state encoding in u64
-/// Bits: [63:48] reserved | [47:32] writers_waiting | [31:16] readers_waiting | [15:8] readers_count | [7:1] flags | [0] writer_flag
+/// Bits: [63:48] readers_count | [47:32] writers_waiting | [31:16] readers_waiting | [15:1] reserved | [0] writer_flag
+///
+/// The reader count is 16 bits wide because distributed bucket-scoped read locks (for example the
+/// per-bucket lifecycle and metadata-transaction fences) are held by every in-flight request into that
+/// bucket on every node, so a single key routinely has hundreds of concurrent holders.
 const WRITER_FLAG_MASK: u64 = 0x1;
-const READERS_SHIFT: u8 = 8;
-const READERS_MASK: u64 = 0xFF << READERS_SHIFT; // Support up to 255 concurrent readers
+const READERS_SHIFT: u8 = 48;
+const READERS_MASK: u64 = 0xFFFF << READERS_SHIFT;
 const READERS_WAITING_SHIFT: u8 = 16;
 const READERS_WAITING_MASK: u64 = 0xFFFF << READERS_WAITING_SHIFT;
 const WRITERS_WAITING_SHIFT: u8 = 32;
@@ -96,9 +100,7 @@ impl AtomicLockState {
                return false;
            }

-            let readers = self.readers_count(current);
-            if readers == 0xFF {
-                // Updated limit to 255
+            if self.readers_count(current) == u16::MAX {
                return false; // Too many readers
            }

@@ -303,8 +305,8 @@ impl AtomicLockState {
        self.last_accessed.store(now, Ordering::Relaxed);
    }

-    fn readers_count(&self, state: u64) -> u8 {
-        ((state & READERS_MASK) >> READERS_SHIFT) as u8
+    fn readers_count(&self, state: u64) -> u16 {
+        ((state & READERS_MASK) >> READERS_SHIFT) as u16
    }

    fn readers_waiting(&self, state: u64) -> u16 {
@@ -622,6 +624,50 @@ mod tests {
        assert!(state.is_free());
    }

+    // Bucket-scoped fences are read-locked by every in-flight request into the bucket, so one key
+    // needs far more than 255 concurrent shared holders. The old 8-bit reader count refused the
+    // 256th reader outright.
+    #[test]
+    fn shared_lock_admits_more_than_255_holders() {
+        let state = ObjectLockState::new();
+        let timeout = Duration::from_secs(30);
+        let owners: Vec<Arc<str>> = (0..1024).map(|i| Arc::from(format!("owner-{i}"))).collect();
+
+        for owner in &owners {
+            assert!(
+                state.try_acquire_shared_fast(owner, timeout),
+                "shared acquisition for {owner} must succeed"
+            );
+        }
+        assert_eq!(state.atomic_state.holder_and_waiter_counts(), (1024, 0));
+        assert!(!state.try_acquire_exclusive_fast(&Arc::from("writer"), timeout));
+
+        for owner in &owners {
+            assert!(state.release_shared(owner), "shared release for {owner} must succeed");
+        }
+        assert!(state.try_acquire_exclusive_fast(&Arc::from("writer"), timeout));
+    }
+
+    #[test]
+    fn shared_lock_refuses_readers_beyond_counter_capacity() {
+        let state = AtomicLockState::new();
+        assert!(state.inc_readers_waiting());
+
+        for _ in 0..u16::MAX {
+            assert!(state.try_acquire_shared());
+        }
+        assert!(!state.try_acquire_shared(), "the reader count must not overflow");
+        assert_eq!(state.holder_and_waiter_counts(), (u64::from(u16::MAX), 1));
+        assert_eq!(
+            state.readers_waiting_count(),
+            1,
+            "a full reader count must not spill into the waiter counters"
+        );
+
+        assert!(state.release_shared());
+        assert!(state.try_acquire_shared(), "a released slot must be reusable");
+    }
+
    #[test]
    fn test_object_lock_state() {
        let state = ObjectLockState::new();
@@ -300,6 +300,43 @@ fn create_test_object_key(bucket: &str, object: &str) -> ObjectKey {
    }
 }

+#[tokio::test(start_paused = true)]
+async fn distributed_namespace_shared_readers_exceed_255() {
+    let managers = (0..3).map(|_| Arc::new(GlobalLockManager::new())).collect::<Vec<_>>();
+    let clients = managers
+        .iter()
+        .map(|manager| Arc::new(LocalClient::with_manager(Arc::clone(manager))) as Arc<dyn LockClient>)
+        .collect();
+    let lock = NamespaceLock::with_clients("shared-reader-capacity".to_string(), clients);
+    let resource = create_test_object_key(".rustfs.sys", "bucket-targets/bucket/transaction.lock");
+    let mut guards = Vec::with_capacity(512);
+
+    for reader in 0..512 {
+        let guard = lock
+            .get_read_lock(resource.clone(), &format!("reader-{reader}"), Duration::from_secs(1))
+            .await
+            .unwrap_or_else(|error| panic!("shared reader {reader} must reach quorum: {error}"));
+        guards.push(guard);
+    }
+    assert_eq!(guards.len(), 512);
+    assert!(guards.iter().all(|guard| !guard.is_lock_lost()));
+    let contended_writer = lock
+        .lock_guard(resource.clone(), "blocked-writer", Duration::from_millis(100), Duration::from_secs(30))
+        .await
+        .expect("writer contention must preserve the acquisition result");
+    assert!(
+        contended_writer.is_none(),
+        "a writer must remain excluded while shared readers hold quorum"
+    );
+
+    drop(guards);
+    wait_until_all_managers_can_write(&managers, resource.clone()).await;
+    let _writer = lock
+        .get_write_lock(resource, "writer-after-readers", Duration::from_secs(1))
+        .await
+        .expect("all reader grants must drain before the next writer acquires quorum");
+}
+
 async fn wait_until_all_managers_can_write(managers: &[Arc<GlobalLockManager>], resource: ObjectKey) {
    let deadline = tokio::time::Instant::now() + Duration::from_secs(2);