diff --git a/crates/lock/src/client/local.rs b/crates/lock/src/client/local.rs index 95adaffc3..84ef32a88 100644 --- a/crates/lock/src/client/local.rs +++ b/crates/lock/src/client/local.rs @@ -48,6 +48,7 @@ pub struct LocalClient { struct LocalGuardEntry { guard: FastLockGuard, acquired_at: SystemTime, + last_refreshed: SystemTime, expires_at: SystemTime, deadline: Instant, ttl: Duration, @@ -60,6 +61,7 @@ impl LocalGuardEntry { Self { guard, acquired_at, + last_refreshed: acquired_at, expires_at: acquired_at.checked_add(ttl).unwrap_or(acquired_at), deadline: monotonic_now.checked_add(ttl).unwrap_or(monotonic_now), ttl, @@ -74,6 +76,7 @@ impl LocalGuardEntry { let now = SystemTime::now(); let monotonic_now = Instant::now(); self.expires_at = now.checked_add(self.ttl).unwrap_or(now); + self.last_refreshed = now; self.deadline = monotonic_now.checked_add(self.ttl).unwrap_or(monotonic_now); } } @@ -347,7 +350,7 @@ impl LockClient for LocalClient { owner: entry.guard.owner().to_string(), acquired_at: entry.acquired_at, expires_at: entry.expires_at, - last_refreshed: SystemTime::now(), + last_refreshed: entry.last_refreshed, metadata: LockMetadata::default(), priority: LockPriority::Normal, wait_start_time: None, @@ -371,6 +374,7 @@ impl LockClient for LocalClient { owner: entry.guard.owner().to_string(), acquired_at: entry.acquired_at, remaining_ttl: entry.deadline.saturating_duration_since(Instant::now()), + guard_id: (!entry.guard.is_disabled()).then(|| entry.guard.guard_id()), })); } leases @@ -484,6 +488,12 @@ mod tests { .success ); let initial = client.list_lock_leases().await.pop().expect("acquired lock should be listed"); + let initial_status = client + .check_status(&lock_id) + .await + .expect("initial lock status should be readable") + .expect("newly acquired lock should remain held"); + assert_eq!(initial_status.last_refreshed, initial_status.acquired_at); tokio::time::advance(Duration::from_secs(20)).await; let aging = client @@ -492,6 +502,13 @@ mod tests { .pop() .expect("held lock should remain listed before refresh"); assert_eq!(aging.remaining_ttl, Duration::from_secs(10)); + let aging_status = client + .check_status(&lock_id) + .await + .expect("aging lock status should be readable") + .expect("aging lock should remain held"); + assert_eq!(aging_status.last_refreshed, initial_status.last_refreshed); + assert!(client.refresh(&lock_id).await.expect("refresh should return a result")); let refreshed = client @@ -506,7 +523,9 @@ mod tests { .expect("refreshed lock should remain held"); assert_eq!(refreshed.acquired_at, initial.acquired_at); + assert_eq!(refreshed.guard_id, initial.guard_id); assert_eq!(status.acquired_at, initial.acquired_at); + assert!(status.last_refreshed > initial_status.last_refreshed); assert_eq!(refreshed.remaining_ttl, Duration::from_secs(30)); tokio::time::advance(Duration::from_secs(30)).await; diff --git a/crates/lock/src/fast_lock/manager.rs b/crates/lock/src/fast_lock/manager.rs index 30a1c4d18..fecb2d92f 100644 --- a/crates/lock/src/fast_lock/manager.rs +++ b/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>)> { + 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; } diff --git a/crates/lock/src/fast_lock/shard.rs b/crates/lock/src/fast_lock/shard.rs index 8b55b968b..5da4d355b 100644 --- a/crates/lock/src/fast_lock/shard.rs +++ b/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, +} + +#[derive(Debug, PartialEq, Eq, Hash)] +struct GuardHolderKey { + key: ObjectKey, + mode: LockMode, + owner: Arc, +} /// 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>, + active_guards: parking_lot::Mutex>>, } /// 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) { 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>)> { + // 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> = 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 diff --git a/crates/lock/src/fast_lock/types.rs b/crates/lock/src/fast_lock/types.rs index 8337971fe..fbf51f9ae 100644 --- a/crates/lock/src/fast_lock/types.rs +++ b/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, diff --git a/crates/lock/src/types.rs b/crates/lock/src/types.rs index 5336fb93e..77a344dea 100644 --- a/crates/lock/src/types.rs +++ b/crates/lock/src/types.rs @@ -90,6 +90,8 @@ pub struct LockLeaseInfo { pub owner: String, /// Original acquisition time. Refreshes do not change this value. pub acquired_at: SystemTime, + /// Opaque guard identity used to reject stale diagnostic snapshots. + pub guard_id: Option, /// Remaining lease duration derived from the monotonic lease deadline. pub remaining_ttl: Duration, } diff --git a/rustfs/src/admin/handlers/diagnostics.rs b/rustfs/src/admin/handlers/diagnostics.rs index 35410c159..83c3625fe 100644 --- a/rustfs/src/admin/handlers/diagnostics.rs +++ b/rustfs/src/admin/handlers/diagnostics.rs @@ -42,6 +42,7 @@ use s3s::{Body, S3Error, S3ErrorCode, S3Request, S3Response, S3Result, StdError, use serde::{Deserialize, Serialize}; use std::collections::HashMap; use std::pin::Pin; +use std::sync::Arc; use std::task::{Context, Poll}; use std::time::{Duration, SystemTime}; use tokio::sync::{Semaphore, SemaphorePermit, mpsc}; @@ -248,13 +249,14 @@ struct LeaseHolderState { acquired_at: SystemTime, ttl_secs: u64, holder_count: u32, + guard_ids: Option>, } fn build_top_locks_response( limit: usize, now: SystemTime, lease_infos: Vec, - fast_infos: Vec<(rustfs_lock::ObjectLockInfo, u32)>, + fast_infos: Vec<(rustfs_lock::ObjectLockInfo, u32, Option>)>, ) -> TopLocksResponse { let mut lease_holders = HashMap::with_capacity(lease_infos.len()); @@ -277,17 +279,27 @@ fn build_top_locks_response( } state.ttl_secs = state.ttl_secs.max(ttl_secs); state.holder_count = state.holder_count.saturating_add(1); + match (state.guard_ids.as_mut(), info.guard_id) { + (Some(guard_ids), Some(guard_id)) => guard_ids.push(guard_id), + _ => state.guard_ids = None, + } }) .or_insert(LeaseHolderState { acquired_at: info.acquired_at, ttl_secs, holder_count: 1, + guard_ids: info.guard_id.map(|guard_id| vec![guard_id]), }); } + for state in lease_holders.values_mut() { + if let Some(guard_ids) = &mut state.guard_ids { + guard_ids.sort_unstable(); + } + } let mut infos: Vec<_> = fast_infos .into_iter() - .map(|(info, holder_count)| { + .map(|(info, holder_count, guard_ids)| { let mode = match info.mode { LockMode::Shared => TopLockMode::Read, LockMode::Exclusive => TopLockMode::Write, @@ -298,11 +310,10 @@ fn build_top_locks_response( owner: info.owner.to_string(), }; let priority = lock_priority_label(info.priority); - // Shared-owner timestamps do not roll back when a newer sibling releases, so only their count is stable. + // Match the complete holder cohort so replacements cannot reuse stale lease data. let state = match lease_holders.remove(&key) { Some(lease) - if lease.holder_count == holder_count - && (mode == TopLockMode::Read || info.acquired_at <= lease.acquired_at) => + if lease.holder_count == holder_count && lease.guard_ids.is_some() && lease.guard_ids == guard_ids => { TopLockState { acquired_at: lease.acquired_at, @@ -349,8 +360,11 @@ fn build_top_locks_response( } } -async fn collect_top_locks(limit: usize) -> TopLocksResponse { - let manager = get_global_lock_manager(); +async fn collect_top_locks_with_clients( + limit: usize, + manager: Arc, + clients: Vec>, +) -> TopLocksResponse { let Some(fast) = manager.as_fast_lock_manager() else { return TopLocksResponse { total: 0, @@ -362,21 +376,29 @@ async fn collect_top_locks(limit: usize) -> TopLocksResponse { }; }; - let lease_infos = if let Some(clients) = get_global_lock_clients() { - join_all(clients.values().map(|client| client.list_lock_leases())) + let lease_infos = if clients.is_empty() { + Vec::new() + } else { + join_all(clients.iter().map(|client| client.list_lock_leases())) .await .into_iter() .flatten() .collect() - } else { - Vec::new() }; // Capture holders last so released or replaced lease guards fail the merge checks. - let fast_infos = fast.list_locks_with_holder_counts(); + let fast_infos = fast.list_locks_with_holder_generations(); build_top_locks_response(limit, SystemTime::now(), lease_infos, fast_infos) } +async fn collect_top_locks(limit: usize) -> TopLocksResponse { + let manager = get_global_lock_manager(); + let clients = get_global_lock_clients() + .map(|clients| clients.values().cloned().collect()) + .unwrap_or_default(); + collect_top_locks_with_clients(limit, manager, clients).await +} + fn parse_top_locks_limit(uri: &Uri) -> usize { query_value(uri, "count") .and_then(|v| v.parse::().ok()) @@ -1229,6 +1251,7 @@ mod tests { let mixed_resource = ObjectKey::new("bucket", "mixed-object"); let replaced_resource = ObjectKey::new("bucket", "replaced-object"); let remaining_shared_resource = ObjectKey::new("bucket", "remaining-shared-object"); + let opaque_resource = ObjectKey::new("bucket", "opaque-object"); let response = build_top_locks_response( TOP_LOCKS_DEFAULT_LIMIT, @@ -1239,6 +1262,7 @@ mod tests { lock_type: LockType::Shared, owner: "owner-a".to_string(), acquired_at: now - Duration::from_secs(50), + guard_id: Some(11), remaining_ttl: Duration::from_secs(5), }, LockLeaseInfo { @@ -1246,6 +1270,7 @@ mod tests { lock_type: LockType::Shared, owner: "owner-a".to_string(), acquired_at: now - Duration::from_secs(40), + guard_id: Some(18), remaining_ttl: Duration::from_secs(20), }, LockLeaseInfo { @@ -1253,6 +1278,7 @@ mod tests { lock_type: LockType::Shared, owner: "owner-c".to_string(), acquired_at: now - Duration::from_secs(30), + guard_id: Some(13), remaining_ttl: Duration::from_secs(25), }, LockLeaseInfo { @@ -1260,6 +1286,7 @@ mod tests { lock_type: LockType::Exclusive, owner: "owner-d".to_string(), acquired_at: now - Duration::from_secs(15), + guard_id: Some(12), remaining_ttl: Duration::from_secs(18), }, LockLeaseInfo { @@ -1267,6 +1294,7 @@ mod tests { lock_type: LockType::Exclusive, owner: "owner-e".to_string(), acquired_at: now - Duration::from_secs(30), + guard_id: Some(14), remaining_ttl: Duration::from_secs(25), }, LockLeaseInfo { @@ -1274,8 +1302,17 @@ mod tests { lock_type: LockType::Shared, owner: "owner-f".to_string(), acquired_at: now - Duration::from_secs(30), + guard_id: Some(16), remaining_ttl: Duration::from_secs(22), }, + LockLeaseInfo { + resource: opaque_resource.clone(), + lock_type: LockType::Exclusive, + owner: "owner-g".to_string(), + acquired_at: now - Duration::from_secs(30), + guard_id: None, + remaining_ttl: Duration::from_secs(30), + }, ], vec![ ( @@ -1288,6 +1325,7 @@ mod tests { priority: rustfs_lock::fast_lock::LockPriority::Normal, }, 1, + Some(vec![15]), ), ( rustfs_lock::ObjectLockInfo { @@ -1299,6 +1337,7 @@ mod tests { priority: rustfs_lock::fast_lock::LockPriority::Normal, }, 1, + Some(vec![16]), ), ( rustfs_lock::ObjectLockInfo { @@ -1310,6 +1349,7 @@ mod tests { priority: rustfs_lock::fast_lock::LockPriority::Normal, }, 1, + Some(vec![12]), ), ( rustfs_lock::ObjectLockInfo { @@ -1321,6 +1361,7 @@ mod tests { priority: rustfs_lock::fast_lock::LockPriority::Normal, }, 2, + Some(vec![11, 18]), ), ( rustfs_lock::ObjectLockInfo { @@ -1332,6 +1373,7 @@ mod tests { priority: rustfs_lock::fast_lock::LockPriority::Normal, }, 1, + Some(vec![17]), ), ( rustfs_lock::ObjectLockInfo { @@ -1343,11 +1385,24 @@ mod tests { priority: rustfs_lock::fast_lock::LockPriority::Normal, }, 2, + None, + ), + ( + rustfs_lock::ObjectLockInfo { + key: opaque_resource, + mode: LockMode::Exclusive, + owner: "owner-g".into(), + acquired_at: now - Duration::from_secs(4), + expires_at: now + Duration::from_secs(6), + priority: rustfs_lock::fast_lock::LockPriority::Normal, + }, + 1, + None, ), ], ); - assert_eq!(response.total, 6); + assert_eq!(response.total, 7); let leased = response .locks .iter() @@ -1392,6 +1447,47 @@ mod tests { .find(|entry| entry.object == "remaining-shared-object") .expect("an older surviving shared lease should remain lease-backed"); assert_eq!(remaining_shared.ttl_secs, 22); + + let opaque = response + .locks + .iter() + .find(|entry| entry.object == "opaque-object") + .expect("generation-less holder should remain visible"); + assert_eq!(opaque.ttl_secs, 6); + } + + #[test] + fn top_locks_rejects_replaced_shared_generation() { + let now = SystemTime::UNIX_EPOCH + Duration::from_secs(100); + let resource = ObjectKey::new("bucket", "replaced-shared-object"); + let response = build_top_locks_response( + TOP_LOCKS_DEFAULT_LIMIT, + now, + vec![LockLeaseInfo { + resource: resource.clone(), + lock_type: LockType::Shared, + owner: "owner-a".to_string(), + acquired_at: now - Duration::from_secs(30), + guard_id: Some(1), + remaining_ttl: Duration::from_secs(20), + }], + vec![( + rustfs_lock::ObjectLockInfo { + key: resource, + mode: LockMode::Shared, + owner: "owner-a".into(), + acquired_at: now - Duration::from_secs(2), + expires_at: now + Duration::from_secs(4), + priority: rustfs_lock::fast_lock::LockPriority::Normal, + }, + 1, + Some(vec![2]), + )], + ); + + let entry = response.locks.first().expect("replacement remains visible"); + assert_eq!(entry.ttl_secs, 4); + assert_eq!(entry.elapsed_secs, 2); } #[tokio::test] @@ -1431,4 +1527,36 @@ mod tests { drop(guard); } + + #[tokio::test(start_paused = true)] + async fn collect_top_locks_uses_refreshed_local_lease() { + use rustfs_lock::{FastObjectLockManager, GlobalLockManager, LocalClient, LockClient, LockRequest}; + + let manager = Arc::new(GlobalLockManager::Enabled(Arc::new(FastObjectLockManager::new()))); + let client = Arc::new(LocalClient::with_manager(manager.clone())); + let request = LockRequest::new(ObjectKey::new("diag-bucket", "renewed-object"), LockType::Exclusive, "diag-owner") + .with_ttl(Duration::from_secs(30)); + let lock_id = request.lock_id.clone(); + assert!( + client + .acquire_lock(&request) + .await + .expect("local lock acquisition should succeed") + .success + ); + + tokio::time::advance(Duration::from_secs(20)).await; + assert!(client.refresh(&lock_id).await.expect("local lease refresh should succeed")); + + let clients: Vec> = vec![client.clone()]; + let response = collect_top_locks_with_clients(TOP_LOCKS_DEFAULT_LIMIT, manager, clients).await; + let entry = response + .locks + .iter() + .find(|entry| entry.bucket == "diag-bucket" && entry.object == "renewed-object") + .expect("refreshed local lock should be listed"); + assert!(entry.ttl_secs >= 29, "collector must use the refreshed lease deadline"); + + assert!(client.release(&lock_id).await.expect("local lock release should succeed")); + } }