diff --git a/crates/lock/src/client/local.rs b/crates/lock/src/client/local.rs index 2c65f2040..134db98db 100644 --- a/crates/lock/src/client/local.rs +++ b/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>>; +type GuardStorage = Arc>; + /// 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>>>, + 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>, + 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>>> = - (0..shard_count).map(|_| Arc::new(RwLock::new(HashMap::new()))).collect(); + let guard_storage: Vec = (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>, 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) -> 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, 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 { + 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, 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 { + 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"); + } +} diff --git a/crates/lock/src/namespace/tests.rs b/crates/lock/src/namespace/tests.rs index 0ee5a81bb..3316828f7 100644 --- a/crates/lock/src/namespace/tests.rs +++ b/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::>(); + 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::>(); + let clients = flaky_clients + .iter() + .map(|client| client.clone() as Arc) + .collect::>(); + 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::>(); + let node_clients = managers + .iter() + .map(|manager| Arc::new(LocalClient::with_manager_and_reaper_interval(manager.clone(), Duration::from_millis(5)))) + .collect::>(); + 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) + .collect::>(); + 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::>();