gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-19 19:16:17 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(admin): report refreshed lock lease TTLs

Browse Source
This commit is contained in:
马登山
2026-08-19 07:59:56 +08:00
parent a4ea36b298
commit e6d1c3449f
7 changed files with 491 additions and 69 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/lock/src/client/local.rs
+82 -6
View File
@@ -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<crate::LockLeaseInfo> {
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<LockStats> {
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());
crates/lock/src/client/mod.rs
+8 -1
View File
@@ -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<Option<LockInfo>>;
/// 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<LockLeaseInfo> {
Vec::new()
}
/// Get statistics
async fn get_stats(&self) -> Result<LockStats>;
crates/lock/src/fast_lock/manager.rs
+25 -1
View File
@@ -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<crate::fast_lock::types::ObjectLockInfo> {
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;
}
crates/lock/src/fast_lock/shard.rs
+29 -16
View File
@@ -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<crate::fast_lock::types::ObjectLockInfo> {
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,
));
}
}
}
crates/lock/src/lib.rs
+2 -2
View File
@@ -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,
},
};
crates/lock/src/types.rs
+15
View File
@@ -79,6 +79,21 @@ pub struct LockInfo {
pub wait_start_time: Option<SystemTime>,
}
/// 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 {