Merge branch 'main' into cxymds/fix-1853-lock-lease-reaper

crates/lock/src/client/local.rs

@@ -15,8 +15,10 @@
 use std::collections::HashMap;
 use std::hash::{Hash, Hasher};
 use std::sync::Arc;
+use std::sync::atomic::{AtomicBool, Ordering};
 use std::time::{Duration, SystemTime};
 use tokio::sync::RwLock;
+use tokio::time::Instant;
 
 use crate::{
     FastLockGuard, GlobalLockManager, LockClient, LockId, LockInfo, LockManager, LockMetadata, LockPriority, LockRequest,
@@ -26,43 +28,51 @@ use crate::{
 /// Default shard count for guard storage (must be power of 2)
 const DEFAULT_GUARD_SHARD_COUNT: usize = 64;
 
+type GuardShard = Arc<RwLock<HashMap<LockId, LocalGuardEntry>>>;
+type GuardStorage = Arc<Vec<GuardShard>>;
+
 /// Local lock client using FastLock with sharded guard storage for better concurrency
 #[derive(Debug)]
 pub struct LocalClient {
     /// Sharded guard storage to reduce lock contention
-    guard_storage: Vec<Arc<RwLock<HashMap<LockId, LocalGuardEntry>>>>,
+    guard_storage: GuardStorage,
     /// Mask for fast shard index calculation (shard_count - 1)
     shard_mask: usize,
     /// Optional lock manager (if None, uses global singleton)
     manager: Option<Arc<GlobalLockManager>>,
+    reaper_started: AtomicBool,
+    reaper_interval: Duration,
 }
 
 #[derive(Debug)]
 struct LocalGuardEntry {
     guard: FastLockGuard,
     expires_at: SystemTime,
+    deadline: Instant,
     ttl: Duration,
-    /// Owner recorded at acquire time; used only for reclaim diagnostics (#899).
-    owner: String,
 }
 
 impl LocalGuardEntry {
-    fn new(guard: FastLockGuard, ttl: Duration, owner: String) -> Self {
+    fn new(guard: FastLockGuard, ttl: Duration) -> Self {
         let now = SystemTime::now();
+        let monotonic_now = Instant::now();
         Self {
             guard,
-            expires_at: now + ttl,
+            expires_at: now.checked_add(ttl).unwrap_or(now),
+            deadline: monotonic_now.checked_add(ttl).unwrap_or(monotonic_now),
             ttl,
-            owner,
         }
     }
 
     fn is_expired(&self) -> bool {
-        self.expires_at <= SystemTime::now()
+        self.deadline <= Instant::now()
     }
 
     fn refresh(&mut self) {
-        self.expires_at = SystemTime::now() + self.ttl;
+        let now = SystemTime::now();
+        let monotonic_now = Instant::now();
+        self.expires_at = now.checked_add(self.ttl).unwrap_or(now);
+        self.deadline = monotonic_now.checked_add(self.ttl).unwrap_or(monotonic_now);
     }
 }
 
@@ -77,26 +87,38 @@ impl LocalClient {
     pub fn with_shard_count(shard_count: usize) -> Self {
         assert!(shard_count.is_power_of_two(), "Shard count must be power of 2");
 
-        let guard_storage: Vec<Arc<RwLock<HashMap<LockId, LocalGuardEntry>>>> =
-            (0..shard_count).map(|_| Arc::new(RwLock::new(HashMap::new()))).collect();
+        let guard_storage: Vec<GuardShard> = (0..shard_count).map(|_| Arc::new(RwLock::new(HashMap::new()))).collect();
 
+        Self::with_storage(Arc::new(guard_storage), None, crate::fast_lock::CLEANUP_INTERVAL)
+    }
+
+    fn with_storage(guard_storage: GuardStorage, manager: Option<Arc<GlobalLockManager>>, reaper_interval: Duration) -> Self {
+        let shard_count = guard_storage.len();
+        debug_assert!(shard_count.is_power_of_two());
         Self {
             guard_storage,
             shard_mask: shard_count - 1,
-            manager: None,
+            manager,
+            reaper_started: AtomicBool::new(false),
+            reaper_interval,
         }
     }
 
     /// Create new local client with a specific lock manager
     /// This allows simulating multi-node environments where each node has its own lock backend
     pub fn with_manager(manager: Arc<GlobalLockManager>) -> Self {
-        Self {
-            guard_storage: (0..DEFAULT_GUARD_SHARD_COUNT)
-                .map(|_| Arc::new(RwLock::new(HashMap::new())))
-                .collect(),
-            shard_mask: DEFAULT_GUARD_SHARD_COUNT - 1,
-            manager: Some(manager),
-        }
+        let guard_storage = (0..DEFAULT_GUARD_SHARD_COUNT)
+            .map(|_| Arc::new(RwLock::new(HashMap::new())))
+            .collect();
+        Self::with_storage(Arc::new(guard_storage), Some(manager), crate::fast_lock::CLEANUP_INTERVAL)
+    }
+
+    #[cfg(test)]
+    pub(crate) fn with_manager_and_reaper_interval(manager: Arc<GlobalLockManager>, reaper_interval: Duration) -> Self {
+        let guard_storage = (0..DEFAULT_GUARD_SHARD_COUNT)
+            .map(|_| Arc::new(RwLock::new(HashMap::new())))
+            .collect();
+        Self::with_storage(Arc::new(guard_storage), Some(manager), reaper_interval)
     }
 
     /// Get the lock manager (injected manager if available, otherwise global singleton)
@@ -118,52 +140,61 @@ impl LocalClient {
     }
 
     async fn reclaim_expired_guards_for_resource(&self, resource: &crate::ObjectKey) -> usize {
-        let mut reclaimed = 0usize;
+        let expired_entries = Self::extract_expired_guards(&self.guard_storage, Some(resource)).await;
+        Self::release_reclaimed_guards(expired_entries, Some(resource))
+    }
 
-        for shard in &self.guard_storage {
-            let expired_entries = {
-                let mut guards = shard.write().await;
-                let mut retained = HashMap::with_capacity(guards.len());
-                let mut expired_entries = Vec::new();
+    async fn extract_expired_guards(storage: &GuardStorage, resource: Option<&crate::ObjectKey>) -> Vec<LocalGuardEntry> {
+        let mut expired_entries = Vec::new();
+        for shard in storage.iter() {
+            let mut guards = shard.write().await;
+            expired_entries.extend(
+                guards
+                    .extract_if(|lock_id, entry| {
+                        resource.is_none_or(|resource| &lock_id.resource == resource) && entry.is_expired()
+                    })
+                    .map(|(_, entry)| entry),
+            );
+        }
+        expired_entries
+    }
 
-                for (lock_id, entry) in std::mem::take(&mut *guards) {
-                    if &lock_id.resource == resource && entry.is_expired() {
-                        expired_entries.push(entry);
-                    } else {
-                        retained.insert(lock_id, entry);
-                    }
-                }
-
-                *guards = retained;
-                expired_entries
-            };
-
-            for mut entry in expired_entries {
-                // An expired entry whose owner never refreshed it (a dead coordinator, #698) is
-                // reclaimed so a live contender can re-form quorum. With guard heartbeats in place
-                // (#899) a live owner keeps its entry from expiring, so reaching here means the
-                // lease genuinely lapsed. Surface it for observability; the reclaim decision itself
-                // is unchanged.
-                let since_last_refresh = entry
-                    .expires_at
-                    .checked_sub(entry.ttl)
-                    .and_then(|last_refresh| SystemTime::now().duration_since(last_refresh).ok())
-                    .unwrap_or(entry.ttl);
-                tracing::warn!(
-                    owner = %entry.owner,
-                    resource = %resource,
-                    ttl_ms = entry.ttl.as_millis() as u64,
-                    since_last_refresh_ms = since_last_refresh.as_millis() as u64,
-                    "reclaiming expired lock guard whose lease was not refreshed"
-                );
+    fn release_reclaimed_guards(mut entries: Vec<LocalGuardEntry>, resource: Option<&crate::ObjectKey>) -> usize {
+        let reclaimed = entries.len();
+        for entry in &mut entries {
+            let _ = entry.guard.release();
+        }
+        if reclaimed > 0 {
+            for _ in 0..reclaimed {
                 rustfs_io_metrics::record_lock_reclaimed();
-                let _ = entry.guard.release();
-                reclaimed = reclaimed.saturating_add(1);
+            }
+            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");
             }
         }
-
         reclaimed
     }
+
+    fn ensure_reaper(&self) {
+        if self.reaper_started.swap(true, Ordering::AcqRel) {
+            return;
+        }
+        let storage = Arc::downgrade(&self.guard_storage);
+        let interval = self.reaper_interval;
+        tokio::spawn(async move {
+            let mut ticker = tokio::time::interval(interval);
+            loop {
+                ticker.tick().await;
+                let Some(storage) = storage.upgrade() else {
+                    break;
+                };
+                let expired_entries = Self::extract_expired_guards(&storage, None).await;
+                Self::release_reclaimed_guards(expired_entries, None);
+            }
+        });
+    }
 }
 
 impl Default for LocalClient {
@@ -175,28 +206,36 @@ impl Default for LocalClient {
 #[async_trait::async_trait]
 impl LockClient for LocalClient {
     async fn acquire_lock(&self, request: &LockRequest) -> Result<LockResponse> {
+        self.ensure_reaper();
         let lock_manager = self.get_lock_manager();
         let reclaimed_before_acquire = self.reclaim_expired_guards_for_resource(&request.resource).await;
+        let acquire_deadline = Instant::now()
+            .checked_add(request.acquire_timeout)
+            .unwrap_or_else(Instant::now);
 
-        let build_lock_request = || match request.lock_type {
+        let build_lock_request = |acquire_timeout| match request.lock_type {
             LockType::Exclusive => crate::ObjectLockRequest::new_write(request.resource.clone(), request.owner.clone())
-                .with_acquire_timeout(request.acquire_timeout),
+                .with_acquire_timeout(acquire_timeout),
             LockType::Shared => crate::ObjectLockRequest::new_read(request.resource.clone(), request.owner.clone())
-                .with_acquire_timeout(request.acquire_timeout),
+                .with_acquire_timeout(acquire_timeout),
         };
 
         let mut retried_after_reclaim = reclaimed_before_acquire > 0;
         loop {
-            match lock_manager.acquire_lock(build_lock_request()).await {
+            let remaining = acquire_deadline.saturating_duration_since(Instant::now());
+            if remaining.is_zero() {
+                return Ok(LockResponse::failure("Lock acquisition timeout", request.acquire_timeout));
+            }
+            match lock_manager.acquire_lock(build_lock_request(remaining)).await {
                 Ok(guard) => {
                     let lock_id = request.lock_id.clone();
                     let acquired_at = SystemTime::now();
-                    let expires_at = acquired_at + request.ttl;
+                    let expires_at = acquired_at.checked_add(request.ttl).unwrap_or(acquired_at);
 
                     {
                         let shard = self.get_shard(&lock_id);
                         let mut guards = shard.write().await;
-                        guards.insert(lock_id.clone(), LocalGuardEntry::new(guard, request.ttl, request.owner.clone()));
+                        guards.insert(lock_id.clone(), LocalGuardEntry::new(guard, request.ttl));
                     }
 
                     let lock_info = LockInfo {
@@ -317,3 +356,111 @@ impl LockClient for LocalClient {
         true
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::{GlobalLockManager, LockClient, LockRequest, LockType};
+
+    fn request(resource: crate::ObjectKey, owner: &str, ttl: Duration) -> LockRequest {
+        LockRequest::new(resource, LockType::Exclusive, owner)
+            .with_ttl(ttl)
+            .with_acquire_timeout(Duration::from_millis(80))
+    }
+
+    async fn wait_until_reaped(client: &LocalClient, lock_id: &LockId) {
+        for _ in 0..80 {
+            if client.check_status(lock_id).await.unwrap().is_none() {
+                return;
+            }
+            tokio::time::sleep(Duration::from_millis(5)).await;
+        }
+        panic!("lock guard was not reaped before test deadline");
+    }
+
+    #[tokio::test(flavor = "current_thread")]
+    async fn expired_guard_is_reaped_without_resource_reacquire() {
+        let manager = Arc::new(GlobalLockManager::new());
+        let client = LocalClient::with_manager_and_reaper_interval(manager.clone(), Duration::from_millis(5));
+        let request = request(crate::ObjectKey::new("bucket", "unique-chunk"), "owner-a", Duration::from_millis(10));
+        let lock_id = request.lock_id.clone();
+
+        assert!(client.acquire_lock(&request).await.unwrap().success);
+        assert!(client.check_status(&lock_id).await.unwrap().is_some());
+        tokio::time::sleep(Duration::from_millis(15)).await;
+        wait_until_reaped(&client, &lock_id).await;
+
+        let direct = manager
+            .acquire_lock(crate::ObjectLockRequest::new_write(request.resource.clone(), "owner-b"))
+            .await;
+        assert!(direct.is_ok());
+    }
+
+    #[tokio::test(flavor = "current_thread")]
+    async fn sibling_client_cannot_reclaim_but_owner_reaper_releases_shared_lock() {
+        let manager = Arc::new(GlobalLockManager::new());
+        let owner = LocalClient::with_manager_and_reaper_interval(manager.clone(), Duration::from_millis(5));
+        let contender = LocalClient::with_manager_and_reaper_interval(manager, Duration::from_millis(5));
+        let request_a = request(crate::ObjectKey::new("bucket", "shared-resource"), "owner-a", Duration::from_millis(10));
+        assert!(owner.acquire_lock(&request_a).await.unwrap().success);
+
+        let request_b = request(request_a.resource.clone(), "owner-b", Duration::from_millis(20))
+            .with_acquire_timeout(Duration::from_millis(5));
+        assert!(!contender.acquire_lock(&request_b).await.unwrap().success);
+
+        tokio::time::sleep(Duration::from_millis(25)).await;
+        assert!(owner.check_status(&request_a.lock_id).await.unwrap().is_none());
+        assert!(contender.acquire_lock(&request_b).await.unwrap().success);
+    }
+
+    #[tokio::test(flavor = "current_thread")]
+    async fn refresh_wins_before_deadline_and_reaper_wins_after_deadline() {
+        let manager = Arc::new(GlobalLockManager::new());
+        let client = LocalClient::with_manager_and_reaper_interval(manager, Duration::from_millis(5));
+        let request = request(crate::ObjectKey::new("bucket", "refresh-race"), "owner-a", Duration::from_millis(25));
+        let lock_id = request.lock_id.clone();
+        assert!(client.acquire_lock(&request).await.unwrap().success);
+
+        tokio::time::sleep(Duration::from_millis(10)).await;
+        assert!(client.refresh(&lock_id).await.unwrap());
+        tokio::time::sleep(Duration::from_millis(15)).await;
+        assert!(client.check_status(&lock_id).await.unwrap().is_some());
+        wait_until_reaped(&client, &lock_id).await;
+    }
+
+    #[tokio::test(flavor = "current_thread")]
+    async fn zero_ttl_is_reaped_and_oversized_ttl_does_not_panic() {
+        let manager = Arc::new(GlobalLockManager::new());
+        let client = LocalClient::with_manager_and_reaper_interval(manager, Duration::from_millis(5));
+
+        let zero = request(crate::ObjectKey::new("bucket", "zero-ttl"), "owner-zero", Duration::ZERO);
+        let zero_id = zero.lock_id.clone();
+        assert!(client.acquire_lock(&zero).await.unwrap().success);
+        wait_until_reaped(&client, &zero_id).await;
+
+        let huge = request(crate::ObjectKey::new("bucket", "huge-ttl"), "owner-huge", Duration::MAX);
+        let huge_id = huge.lock_id.clone();
+        assert!(client.acquire_lock(&huge).await.unwrap().success);
+        wait_until_reaped(&client, &huge_id).await;
+    }
+
+    #[tokio::test(flavor = "current_thread")]
+    async fn acquire_retry_preserves_total_deadline() {
+        let manager = Arc::new(GlobalLockManager::new());
+        let client = LocalClient::with_manager_and_reaper_interval(manager, Duration::from_secs(60));
+        let first = request(crate::ObjectKey::new("bucket", "deadline-budget"), "owner-a", Duration::from_millis(10));
+        assert!(client.acquire_lock(&first).await.unwrap().success);
+
+        let second =
+            request(first.resource.clone(), "owner-b", Duration::from_millis(30)).with_acquire_timeout(Duration::from_millis(60));
+        let started = Instant::now();
+        let response = client.acquire_lock(&second).await.unwrap();
+        assert!(!response.success, "the first attempt consumed the caller's acquire budget");
+        assert!(
+            started.elapsed() < Duration::from_millis(100),
+            "reclaim retry must not double the acquire budget"
+        );
+        let recovered = client.acquire_lock(&second).await.unwrap();
+        assert!(recovered.success, "the reclaimed guard must be available to the next request");
+    }
+}

crates/lock/src/namespace/tests.rs

@@ -840,6 +840,117 @@ async fn test_namespace_lock_distributed_reclaims_expired_same_resource_after_fa
     );
 }
 
+#[tokio::test]
+async fn four_node_failed_release_converges_without_replica_repair() {
+    let managers = (0..4).map(|_| Arc::new(GlobalLockManager::new())).collect::<Vec<_>>();
+    let flaky_clients = managers
+        .iter()
+        .map(|manager| {
+            Arc::new(FlakyReleaseClient {
+                inner: LocalClient::with_manager_and_reaper_interval(manager.clone(), Duration::from_millis(5)),
+                failed_releases_remaining: AtomicUsize::new(usize::MAX),
+                release_attempts: AtomicUsize::new(0),
+            })
+        })
+        .collect::<Vec<_>>();
+    let clients = flaky_clients
+        .iter()
+        .map(|client| client.clone() as Arc<dyn LockClient>)
+        .collect::<Vec<_>>();
+    let lock = NamespaceLock::Distributed(DistributedLock::new("four-node-expired-lease".to_string(), clients, 3));
+    let resource = create_test_object_key("bucket", "object-four-node-expired");
+    let request = LockRequest::new(resource.clone(), LockType::Exclusive, "owner-a")
+        .with_acquire_timeout(Duration::from_millis(300))
+        .with_ttl(Duration::from_millis(40));
+
+    let mut guard = lock
+        .acquire_guard(&request)
+        .await
+        .expect("initial acquire should not error")
+        .expect("initial acquire should reach quorum");
+    assert!(guard.release(), "release should be acknowledged while RPC cleanup is pending");
+
+    for _ in 0..40 {
+        if flaky_clients.iter().all(|client| client.release_attempts() >= 3) {
+            break;
+        }
+        tokio::time::sleep(Duration::from_millis(5)).await;
+    }
+
+    let deadline = tokio::time::Instant::now() + Duration::from_secs(2);
+    loop {
+        let all_reaped =
+            futures::future::join_all(flaky_clients.iter().map(|client| client.inner.check_status(&request.lock_id)))
+                .await
+                .into_iter()
+                .all(|status| status.expect("status should not error").is_none());
+        if all_reaped {
+            break;
+        }
+        assert!(tokio::time::Instant::now() < deadline, "all four local lease entries must converge");
+        tokio::time::sleep(Duration::from_millis(10)).await;
+    }
+
+    for suffix in ["chunk-0", "chunk-1", ".rustfs.sys/multipart/upload-0"] {
+        for client in &flaky_clients {
+            let orphan = LockRequest::new(create_test_object_key("bucket", suffix), LockType::Exclusive, "orphan")
+                .with_ttl(Duration::from_millis(25));
+            assert!(client.inner.acquire_lock(&orphan).await.expect("orphan acquire").success);
+        }
+    }
+    tokio::time::sleep(Duration::from_millis(80)).await;
+
+    let recovered = lock
+        .acquire_guard(
+            &LockRequest::new(resource, LockType::Exclusive, "owner-b")
+                .with_acquire_timeout(Duration::from_millis(300))
+                .with_ttl(Duration::from_millis(40)),
+        )
+        .await
+        .expect("recovery acquire should not error")
+        .expect("four-node quorum should recover after local reapers run");
+    drop(recovered);
+}
+
+#[tokio::test]
+async fn four_node_stale_quorum_contention_respects_acquire_deadline() {
+    let managers = (0..4).map(|_| Arc::new(GlobalLockManager::new())).collect::<Vec<_>>();
+    let node_clients = managers
+        .iter()
+        .map(|manager| Arc::new(LocalClient::with_manager_and_reaper_interval(manager.clone(), Duration::from_millis(5))))
+        .collect::<Vec<_>>();
+    let resource = create_test_object_key("bucket", "stale-quorum");
+    let stale = LockRequest::new(resource.clone(), LockType::Exclusive, "stale-owner").with_ttl(Duration::from_millis(180));
+    for client in &node_clients {
+        assert!(client.acquire_lock(&stale).await.expect("stale acquire").success);
+    }
+
+    let clients = node_clients
+        .iter()
+        .map(|client| client.clone() as Arc<dyn LockClient>)
+        .collect::<Vec<_>>();
+    let lock = NamespaceLock::Distributed(DistributedLock::new("stale-quorum-deadline".to_string(), clients, 3));
+    let contender = LockRequest::new(resource.clone(), LockType::Exclusive, "new-owner")
+        .with_acquire_timeout(Duration::from_millis(150))
+        .with_ttl(Duration::from_millis(100));
+    let started = tokio::time::Instant::now();
+    let response = lock.acquire_guard(&contender).await.expect("contention should not error");
+    assert!(response.is_none(), "unexpired leases must not be force-reclaimed");
+    assert!(started.elapsed() < Duration::from_millis(350), "acquire must respect its deadline");
+
+    tokio::time::sleep(Duration::from_millis(80)).await;
+    let recovered = lock
+        .acquire_guard(
+            &LockRequest::new(resource, LockType::Exclusive, "new-owner")
+                .with_acquire_timeout(Duration::from_millis(300))
+                .with_ttl(Duration::from_millis(100)),
+        )
+        .await
+        .expect("post-expiry acquire should not error")
+        .expect("quorum should recover after local reapers clear stale leases");
+    drop(recovered);
+}
+
 #[tokio::test]
 async fn test_namespace_lock_distributed_retries_transient_acquire_timeout() {
     let managers = (0..3).map(|_| Arc::new(GlobalLockManager::new())).collect::<Vec<_>>();