From e6d1c3449f8ea9b6b93606082521460d49148492 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=A9=AC=E7=99=BB=E5=B1=B1?= Date: Wed, 19 Aug 2026 07:59:56 +0800 Subject: [PATCH] fix(admin): report refreshed lock lease TTLs --- crates/lock/src/client/local.rs | 88 +++++- crates/lock/src/client/mod.rs | 9 +- crates/lock/src/fast_lock/manager.rs | 26 +- crates/lock/src/fast_lock/shard.rs | 45 ++- crates/lock/src/lib.rs | 4 +- crates/lock/src/types.rs | 15 + rustfs/src/admin/handlers/diagnostics.rs | 373 ++++++++++++++++++++--- 7 files changed, 491 insertions(+), 69 deletions(-) diff --git a/crates/lock/src/client/local.rs b/crates/lock/src/client/local.rs index 3b2a185aa..95adaffc3 100644 --- a/crates/lock/src/client/local.rs +++ b/crates/lock/src/client/local.rs @@ -47,6 +47,7 @@ pub struct LocalClient { #[derive(Debug)] struct LocalGuardEntry { guard: FastLockGuard, + acquired_at: SystemTime, expires_at: SystemTime, deadline: Instant, ttl: Duration, @@ -54,11 +55,12 @@ struct LocalGuardEntry { impl LocalGuardEntry { fn new(guard: FastLockGuard, ttl: Duration) -> Self { - let now = SystemTime::now(); + let acquired_at = SystemTime::now(); let monotonic_now = Instant::now(); Self { guard, - expires_at: now.checked_add(ttl).unwrap_or(now), + acquired_at, + expires_at: acquired_at.checked_add(ttl).unwrap_or(acquired_at), deadline: monotonic_now.checked_add(ttl).unwrap_or(monotonic_now), ttl, } @@ -231,13 +233,14 @@ impl LockClient for LocalClient { 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.checked_add(request.ttl).unwrap_or(acquired_at); + let entry = LocalGuardEntry::new(guard, request.ttl); + let acquired_at = entry.acquired_at; + let expires_at = entry.expires_at; { let shard = self.get_shard(&lock_id); let mut guards = shard.write().await; - guards.insert(lock_id.clone(), LocalGuardEntry::new(guard, request.ttl)); + guards.insert(lock_id.clone(), entry); } let lock_info = LockInfo { @@ -342,7 +345,7 @@ impl LockClient for LocalClient { lock_type, status, owner: entry.guard.owner().to_string(), - acquired_at: SystemTime::now(), + acquired_at: entry.acquired_at, expires_at: entry.expires_at, last_refreshed: SystemTime::now(), metadata: LockMetadata::default(), @@ -354,6 +357,25 @@ impl LockClient for LocalClient { } } + async fn list_lock_leases(&self) -> Vec { + 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 { + resource: lock_id.resource.clone(), + lock_type: match entry.guard.mode() { + crate::LockMode::Shared => LockType::Shared, + crate::LockMode::Exclusive => LockType::Exclusive, + }, + owner: entry.guard.owner().to_string(), + acquired_at: entry.acquired_at, + remaining_ttl: entry.deadline.saturating_duration_since(Instant::now()), + })); + } + leases + } + async fn get_stats(&self) -> Result { Ok(LockStats::default()) } @@ -403,6 +425,10 @@ mod tests { 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; + assert!( + client.list_lock_leases().await.is_empty(), + "reaped guards must disappear from lease diagnostics" + ); let direct = manager .acquire_lock(crate::ObjectLockRequest::new_write(request.resource.clone(), "owner-b")) @@ -442,6 +468,56 @@ mod tests { wait_until_reaped(&client, &lock_id).await; } + #[tokio::test(start_paused = true)] + async fn lease_snapshot_tracks_refresh_without_resetting_acquisition_time() { + let manager = Arc::new(GlobalLockManager::new()); + let client = LocalClient::with_manager_and_reaper_interval(manager, Duration::from_secs(60)); + client.reaper_started.store(true, Ordering::Release); + let lock_request = request(crate::ObjectKey::new("bucket", "lease-snapshot"), "owner-a", Duration::from_secs(30)); + let lock_id = lock_request.lock_id.clone(); + + assert!( + client + .acquire_lock(&lock_request) + .await + .expect("lease-backed lock should acquire") + .success + ); + let initial = client.list_lock_leases().await.pop().expect("acquired lock should be listed"); + + tokio::time::advance(Duration::from_secs(20)).await; + let aging = client + .list_lock_leases() + .await + .pop() + .expect("held lock should remain listed before refresh"); + assert_eq!(aging.remaining_ttl, Duration::from_secs(10)); + assert!(client.refresh(&lock_id).await.expect("refresh should return a result")); + + let refreshed = client + .list_lock_leases() + .await + .pop() + .expect("refreshed lock should be listed"); + let status = client + .check_status(&lock_id) + .await + .expect("lock status should be readable") + .expect("refreshed lock should remain held"); + + assert_eq!(refreshed.acquired_at, initial.acquired_at); + assert_eq!(status.acquired_at, initial.acquired_at); + assert_eq!(refreshed.remaining_ttl, Duration::from_secs(30)); + + tokio::time::advance(Duration::from_secs(30)).await; + let expired = client + .list_lock_leases() + .await + .pop() + .expect("unreaped lease should remain listed"); + assert_eq!(expired.remaining_ttl, Duration::ZERO); + } + #[tokio::test(start_paused = true)] async fn refresh_after_expiry_releases_guard_without_reviving_it() { let manager = Arc::new(GlobalLockManager::new()); diff --git a/crates/lock/src/client/mod.rs b/crates/lock/src/client/mod.rs index d16db4bb6..bf378c3bf 100644 --- a/crates/lock/src/client/mod.rs +++ b/crates/lock/src/client/mod.rs @@ -15,7 +15,7 @@ pub mod local; // pub mod remote; -use crate::{LockId, LockInfo, LockRequest, LockResponse, LockStats, Result}; +use crate::{LockId, LockInfo, LockLeaseInfo, LockRequest, LockResponse, LockStats, Result}; use async_trait::async_trait; use futures::future::join_all; use std::sync::Arc; @@ -54,6 +54,13 @@ pub trait LockClient: Send + Sync + std::fmt::Debug { /// Check lock status async fn check_status(&self, lock_id: &LockId) -> Result>; + /// Return authoritative lease information when this client owns lease state. + /// + /// Clients that do not manage renewable leases return an empty snapshot. + async fn list_lock_leases(&self) -> Vec { + Vec::new() + } + /// Get statistics async fn get_stats(&self) -> Result; diff --git a/crates/lock/src/fast_lock/manager.rs b/crates/lock/src/fast_lock/manager.rs index b24f96d6f..30a1c4d18 100644 --- a/crates/lock/src/fast_lock/manager.rs +++ b/crates/lock/src/fast_lock/manager.rs @@ -295,9 +295,17 @@ impl FastObjectLockManager { /// Powers the admin "top locks" view. Order is shard-then-insertion and is /// not otherwise stable across calls. pub fn list_locks(&self) -> Vec { + self.list_locks_with_holder_counts() + .into_iter() + .map(|(info, _)| info) + .collect() + } + + /// Enumerate held locks with the number of guards represented by each owner. + pub fn list_locks_with_holder_counts(&self) -> Vec<(crate::fast_lock::types::ObjectLockInfo, u32)> { let mut infos = Vec::new(); for shard in &self.shards { - infos.extend(shard.list_locks()); + infos.extend(shard.list_locks_with_holder_counts()); } infos } @@ -556,6 +564,10 @@ mod tests { .acquire_read_lock(read_key.clone(), "reader") .await .expect("read lock should acquire"); + let _second_read_guard = manager + .acquire_read_lock(read_key.clone(), "reader") + .await + .expect("second read lock should acquire"); let mut locks = manager.list_locks(); locks.sort_by(|a, b| a.key.object.cmp(&b.key.object)); @@ -569,6 +581,18 @@ mod tests { assert_eq!(write.mode, LockMode::Exclusive); assert_eq!(write.owner.as_ref(), "writer"); + let counts = manager.list_locks_with_holder_counts(); + let (_, read_holder_count) = counts + .iter() + .find(|(info, _)| info.key == read_key) + .expect("read holder count listed"); + assert_eq!(*read_holder_count, 2); + let (_, write_holder_count) = counts + .iter() + .find(|(info, _)| info.key == write_key) + .expect("write holder count listed"); + assert_eq!(*write_holder_count, 1); + manager.shutdown().await; } diff --git a/crates/lock/src/fast_lock/shard.rs b/crates/lock/src/fast_lock/shard.rs index a0ca7eaa7..8b55b968b 100644 --- a/crates/lock/src/fast_lock/shard.rs +++ b/crates/lock/src/fast_lock/shard.rs @@ -544,6 +544,13 @@ impl LockShard { /// holder. Entries for objects that are tracked but not currently locked /// (e.g. pooled-but-idle state) are skipped. pub fn list_locks(&self) -> Vec { + self.list_locks_with_holder_counts() + .into_iter() + .map(|(info, _)| info) + .collect() + } + + pub(crate) fn list_locks_with_holder_counts(&self) -> Vec<(crate::fast_lock::types::ObjectLockInfo, u32)> { let objects = self.objects.read(); let mut infos = Vec::new(); for (key, state) in objects.iter() { @@ -558,14 +565,17 @@ impl LockShard { .acquired_at .checked_add(info.lock_timeout) .unwrap_or_else(|| info.acquired_at + crate::fast_lock::DEFAULT_LOCK_TIMEOUT); - infos.push(crate::fast_lock::types::ObjectLockInfo { - key: key.clone(), - mode, - owner: info.owner, - acquired_at: info.acquired_at, - expires_at, - priority, - }); + infos.push(( + crate::fast_lock::types::ObjectLockInfo { + key: key.clone(), + mode, + owner: info.owner, + acquired_at: info.acquired_at, + expires_at, + priority, + }, + 1, + )); } } LockMode::Shared => { @@ -574,14 +584,17 @@ impl LockShard { .acquired_at .checked_add(entry.lock_timeout) .unwrap_or_else(|| entry.acquired_at + crate::fast_lock::DEFAULT_LOCK_TIMEOUT); - infos.push(crate::fast_lock::types::ObjectLockInfo { - key: key.clone(), - mode, - owner: entry.owner.clone(), - acquired_at: entry.acquired_at, - expires_at, - priority, - }); + infos.push(( + crate::fast_lock::types::ObjectLockInfo { + key: key.clone(), + mode, + owner: entry.owner.clone(), + acquired_at: entry.acquired_at, + expires_at, + priority, + }, + entry.count, + )); } } } diff --git a/crates/lock/src/lib.rs b/crates/lock/src/lib.rs index 6562f8675..be1f1c68d 100644 --- a/crates/lock/src/lib.rs +++ b/crates/lock/src/lib.rs @@ -51,8 +51,8 @@ pub use crate::{ namespace::{NamespaceLock, NamespaceLockGuard, NamespaceLockWrapper}, // Core types types::{ - HealthInfo, HealthStatus, LockId, LockInfo, LockMetadata, LockPriority, LockRequest, LockResponse, LockStats, LockStatus, - LockType, + HealthInfo, HealthStatus, LockId, LockInfo, LockLeaseInfo, LockMetadata, LockPriority, LockRequest, LockResponse, + LockStats, LockStatus, LockType, }, }; diff --git a/crates/lock/src/types.rs b/crates/lock/src/types.rs index eebe158ff..5336fb93e 100644 --- a/crates/lock/src/types.rs +++ b/crates/lock/src/types.rs @@ -79,6 +79,21 @@ pub struct LockInfo { pub wait_start_time: Option, } +/// Point-in-time lease information exposed by lock clients for diagnostics. +#[derive(Debug, Clone)] +pub struct LockLeaseInfo { + /// Resource protected by the lock. + pub resource: ObjectKey, + /// Shared or exclusive lock mode. + pub lock_type: LockType, + /// Lock owner recorded by the local lock backend. + pub owner: String, + /// Original acquisition time. Refreshes do not change this value. + pub acquired_at: SystemTime, + /// Remaining lease duration derived from the monotonic lease deadline. + pub remaining_ttl: Duration, +} + impl LockInfo { /// Check if the lock has expired pub fn has_expired(&self) -> bool { diff --git a/rustfs/src/admin/handlers/diagnostics.rs b/rustfs/src/admin/handlers/diagnostics.rs index 523df19db..35410c159 100644 --- a/rustfs/src/admin/handlers/diagnostics.rs +++ b/rustfs/src/admin/handlers/diagnostics.rs @@ -28,17 +28,19 @@ use crate::admin::router::{AdminOperation, Operation, S3Router}; use crate::admin::storage_api::access::spawn_traced; use crate::auth::{check_key_valid, get_session_token}; use crate::server::{ADMIN_PREFIX, RemoteAddr}; +use crate::storage::storage_api::get_global_lock_clients; use bytes::Bytes; -use futures::{Stream, StreamExt}; +use futures::{Stream, StreamExt, future::join_all}; use http::{HeaderMap, HeaderValue, Uri, header::CONTENT_LENGTH}; use hyper::{Method, StatusCode}; use matchit::Params; -use rustfs_lock::{LockMode, ObjectKey, get_global_lock_manager}; +use rustfs_lock::{LockLeaseInfo, LockMode, LockType, ObjectKey, get_global_lock_manager}; use rustfs_policy::policy::action::{Action, AdminAction}; use s3s::header::CONTENT_TYPE; use s3s::stream::{ByteStream, DynByteStream}; use s3s::{Body, S3Error, S3ErrorCode, S3Request, S3Response, S3Result, StdError, s3_error}; use serde::{Deserialize, Serialize}; +use std::collections::HashMap; use std::pin::Pin; use std::task::{Context, Poll}; use std::time::{Duration, SystemTime}; @@ -212,7 +214,142 @@ fn system_time_to_rfc3339(t: SystemTime) -> Option { dt.format(&time::format_description::well_known::Rfc3339).ok() } -fn collect_top_locks(limit: usize) -> TopLocksResponse { +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +enum TopLockMode { + Read, + Write, +} + +impl TopLockMode { + fn label(self) -> &'static str { + match self { + Self::Read => "READ", + Self::Write => "WRITE", + } + } +} + +#[derive(Debug, PartialEq, Eq, Hash)] +struct LockHolderKey { + resource: ObjectKey, + mode: TopLockMode, + owner: String, +} + +#[derive(Debug)] +struct TopLockState { + acquired_at: SystemTime, + ttl_secs: u64, + priority: &'static str, +} + +#[derive(Debug)] +struct LeaseHolderState { + acquired_at: SystemTime, + ttl_secs: u64, + holder_count: u32, +} + +fn build_top_locks_response( + limit: usize, + now: SystemTime, + lease_infos: Vec, + fast_infos: Vec<(rustfs_lock::ObjectLockInfo, u32)>, +) -> TopLocksResponse { + let mut lease_holders = HashMap::with_capacity(lease_infos.len()); + + for info in lease_infos { + let mode = match info.lock_type { + LockType::Shared => TopLockMode::Read, + LockType::Exclusive => TopLockMode::Write, + }; + let key = LockHolderKey { + resource: info.resource, + mode, + owner: info.owner, + }; + let ttl_secs = info.remaining_ttl.as_secs(); + lease_holders + .entry(key) + .and_modify(|state: &mut LeaseHolderState| { + if info.acquired_at < state.acquired_at { + state.acquired_at = info.acquired_at; + } + state.ttl_secs = state.ttl_secs.max(ttl_secs); + state.holder_count = state.holder_count.saturating_add(1); + }) + .or_insert(LeaseHolderState { + acquired_at: info.acquired_at, + ttl_secs, + holder_count: 1, + }); + } + + let mut infos: Vec<_> = fast_infos + .into_iter() + .map(|(info, holder_count)| { + let mode = match info.mode { + LockMode::Shared => TopLockMode::Read, + LockMode::Exclusive => TopLockMode::Write, + }; + let key = LockHolderKey { + resource: info.key, + mode, + 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. + let state = match lease_holders.remove(&key) { + Some(lease) + if lease.holder_count == holder_count + && (mode == TopLockMode::Read || info.acquired_at <= lease.acquired_at) => + { + TopLockState { + acquired_at: lease.acquired_at, + ttl_secs: lease.ttl_secs, + priority, + } + } + _ => TopLockState { + acquired_at: info.acquired_at, + ttl_secs: info.expires_at.duration_since(now).unwrap_or(Duration::ZERO).as_secs(), + priority, + }, + }; + (key, state) + }) + .collect(); + // Longest-held first, matching MinIO's `top locks` ordering intent. + infos.sort_by_key(|(_, state)| state.acquired_at); + let total = infos.len(); + let truncated = total > limit; + + let locks = infos + .into_iter() + .take(limit) + .map(|(holder, state)| LockEntry { + resource: format!("{}/{}", holder.resource.bucket, holder.resource.object), + bucket: holder.resource.bucket.to_string(), + object: holder.resource.object.to_string(), + version: holder.resource.version.as_ref().map(|version| version.to_string()), + lock_type: holder.mode.label(), + owner: holder.owner, + priority: state.priority, + since: system_time_to_rfc3339(state.acquired_at), + elapsed_secs: now.duration_since(state.acquired_at).unwrap_or(Duration::ZERO).as_secs(), + ttl_secs: state.ttl_secs, + }) + .collect(); + + TopLocksResponse { + total, + truncated, + locks, + capability_note: None, + } +} + +async fn collect_top_locks(limit: usize) -> TopLocksResponse { let manager = get_global_lock_manager(); let Some(fast) = manager.as_fast_lock_manager() else { return TopLocksResponse { @@ -225,43 +362,19 @@ fn collect_top_locks(limit: usize) -> TopLocksResponse { }; }; - let now = SystemTime::now(); - let mut infos = fast.list_locks(); - // Longest-held first, matching MinIO's `top locks` ordering intent. - infos.sort_by_key(|i| i.acquired_at); - let total = infos.len(); - let truncated = total > limit; + let lease_infos = if let Some(clients) = get_global_lock_clients() { + join_all(clients.values().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 locks = infos - .into_iter() - .take(limit) - .map(|info| { - let elapsed_secs = now.duration_since(info.acquired_at).unwrap_or(Duration::ZERO).as_secs(); - let ttl_secs = info.expires_at.duration_since(now).unwrap_or(Duration::ZERO).as_secs(); - LockEntry { - resource: format!("{}/{}", info.key.bucket, info.key.object), - bucket: info.key.bucket.to_string(), - object: info.key.object.to_string(), - version: info.key.version.as_ref().map(|v| v.to_string()), - lock_type: match info.mode { - LockMode::Exclusive => "WRITE", - LockMode::Shared => "READ", - }, - owner: info.owner.to_string(), - priority: lock_priority_label(info.priority), - since: system_time_to_rfc3339(info.acquired_at), - elapsed_secs, - ttl_secs, - } - }) - .collect(); - - TopLocksResponse { - total, - truncated, - locks, - capability_note: None, - } + build_top_locks_response(limit, SystemTime::now(), lease_infos, fast_infos) } fn parse_top_locks_limit(uri: &Uri) -> usize { @@ -279,7 +392,7 @@ impl Operation for TopLocksHandler { async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { authorize(&req, AdminAction::TopLocksAdminAction).await?; let limit = parse_top_locks_limit(&req.uri); - let response = collect_top_locks(limit); + let response = collect_top_locks(limit).await; json_response(StatusCode::OK, &response) } } @@ -1107,6 +1220,180 @@ mod tests { assert_eq!(parse_top_locks_limit(&Uri::from_static("/x?count=999999")), TOP_LOCKS_MAX_LIMIT); } + #[test] + fn top_locks_prefers_renewable_lease_deadlines() { + let now = SystemTime::UNIX_EPOCH + Duration::from_secs(100); + let leased_resource = ObjectKey::new("bucket", "shared-object"); + let exclusive_resource = ObjectKey::new("bucket", "write-object"); + let direct_resource = ObjectKey::new("bucket", "direct-object"); + 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 response = build_top_locks_response( + TOP_LOCKS_DEFAULT_LIMIT, + now, + vec![ + LockLeaseInfo { + resource: leased_resource.clone(), + lock_type: LockType::Shared, + owner: "owner-a".to_string(), + acquired_at: now - Duration::from_secs(50), + remaining_ttl: Duration::from_secs(5), + }, + LockLeaseInfo { + resource: leased_resource.clone(), + lock_type: LockType::Shared, + owner: "owner-a".to_string(), + acquired_at: now - Duration::from_secs(40), + remaining_ttl: Duration::from_secs(20), + }, + LockLeaseInfo { + resource: mixed_resource.clone(), + lock_type: LockType::Shared, + owner: "owner-c".to_string(), + acquired_at: now - Duration::from_secs(30), + remaining_ttl: Duration::from_secs(25), + }, + LockLeaseInfo { + resource: exclusive_resource.clone(), + lock_type: LockType::Exclusive, + owner: "owner-d".to_string(), + acquired_at: now - Duration::from_secs(15), + remaining_ttl: Duration::from_secs(18), + }, + LockLeaseInfo { + resource: replaced_resource.clone(), + lock_type: LockType::Exclusive, + owner: "owner-e".to_string(), + acquired_at: now - Duration::from_secs(30), + remaining_ttl: Duration::from_secs(25), + }, + LockLeaseInfo { + resource: remaining_shared_resource.clone(), + lock_type: LockType::Shared, + owner: "owner-f".to_string(), + acquired_at: now - Duration::from_secs(30), + remaining_ttl: Duration::from_secs(22), + }, + ], + vec![ + ( + rustfs_lock::ObjectLockInfo { + key: replaced_resource, + mode: LockMode::Exclusive, + owner: "owner-e".into(), + acquired_at: now - Duration::from_secs(5), + expires_at: now + Duration::from_secs(4), + priority: rustfs_lock::fast_lock::LockPriority::Normal, + }, + 1, + ), + ( + rustfs_lock::ObjectLockInfo { + key: remaining_shared_resource, + mode: LockMode::Shared, + owner: "owner-f".into(), + acquired_at: now - Duration::from_secs(5), + expires_at: now + Duration::from_secs(3), + priority: rustfs_lock::fast_lock::LockPriority::Normal, + }, + 1, + ), + ( + rustfs_lock::ObjectLockInfo { + key: exclusive_resource, + mode: LockMode::Exclusive, + owner: "owner-d".into(), + acquired_at: now - Duration::from_secs(15), + expires_at: now + Duration::from_secs(2), + priority: rustfs_lock::fast_lock::LockPriority::Normal, + }, + 1, + ), + ( + rustfs_lock::ObjectLockInfo { + key: leased_resource, + mode: LockMode::Shared, + owner: "owner-a".into(), + acquired_at: now - Duration::from_secs(50), + expires_at: now + Duration::from_secs(1), + priority: rustfs_lock::fast_lock::LockPriority::Normal, + }, + 2, + ), + ( + rustfs_lock::ObjectLockInfo { + key: direct_resource, + mode: LockMode::Exclusive, + owner: "owner-b".into(), + acquired_at: now - Duration::from_secs(10), + expires_at: now + Duration::from_secs(7), + priority: rustfs_lock::fast_lock::LockPriority::Normal, + }, + 1, + ), + ( + rustfs_lock::ObjectLockInfo { + key: mixed_resource, + mode: LockMode::Shared, + owner: "owner-c".into(), + acquired_at: now - Duration::from_secs(30), + expires_at: now + Duration::from_secs(9), + priority: rustfs_lock::fast_lock::LockPriority::Normal, + }, + 2, + ), + ], + ); + + assert_eq!(response.total, 6); + let leased = response + .locks + .iter() + .find(|entry| entry.object == "shared-object") + .expect("lease-backed shared owner should be listed once"); + assert_eq!(leased.lock_type, "READ"); + assert_eq!(leased.elapsed_secs, 50); + assert_eq!(leased.ttl_secs, 20); + + let exclusive = response + .locks + .iter() + .find(|entry| entry.object == "write-object") + .expect("lease-backed exclusive holder should be listed"); + assert_eq!(exclusive.lock_type, "WRITE"); + assert_eq!(exclusive.ttl_secs, 18); + + let direct = response + .locks + .iter() + .find(|entry| entry.object == "direct-object") + .expect("direct fast lock should remain visible"); + assert_eq!(direct.ttl_secs, 7); + + let mixed = response + .locks + .iter() + .find(|entry| entry.object == "mixed-object") + .expect("mixed direct and leased shared holders should remain visible"); + assert_eq!(mixed.ttl_secs, 9); + + let replaced = response + .locks + .iter() + .find(|entry| entry.object == "replaced-object") + .expect("a replaced lease holder should remain visible"); + assert_eq!(replaced.ttl_secs, 4); + + let remaining_shared = response + .locks + .iter() + .find(|entry| entry.object == "remaining-shared-object") + .expect("an older surviving shared lease should remain lease-backed"); + assert_eq!(remaining_shared.ttl_secs, 22); + } + #[tokio::test] async fn collect_top_locks_reports_live_lock() { // Acquire a real lock through the global manager and confirm it surfaces. @@ -1115,20 +1402,20 @@ mod tests { // The fast-lock manager exposes the acquire API; if the lock subsystem is // disabled in this environment, the response must carry a capability note. let Some(fast) = manager.as_fast_lock_manager() else { - let response = collect_top_locks(TOP_LOCKS_DEFAULT_LIMIT); + let response = collect_top_locks(TOP_LOCKS_DEFAULT_LIMIT).await; assert!(response.capability_note.is_some() || response.locks.is_empty()); return; }; let guard = match fast.acquire_write_lock(key.clone(), "diag-owner").await { Ok(g) => g, Err(_) => { - let response = collect_top_locks(TOP_LOCKS_DEFAULT_LIMIT); + let response = collect_top_locks(TOP_LOCKS_DEFAULT_LIMIT).await; assert!(response.capability_note.is_some() || response.locks.is_empty()); return; } }; - let response = collect_top_locks(TOP_LOCKS_DEFAULT_LIMIT); + let response = collect_top_locks(TOP_LOCKS_DEFAULT_LIMIT).await; let found = response .locks .iter()