mirror of
https://github.com/rustfs/rustfs.git
synced 2026-08-19 02:56:18 +00:00
Compare commits
1 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| e6d1c3449f |
@@ -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());
|
||||
|
||||
@@ -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>;
|
||||
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
|
||||
|
||||
@@ -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,
|
||||
));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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,
|
||||
},
|
||||
};
|
||||
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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<String> {
|
||||
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<LockLeaseInfo>,
|
||||
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<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
|
||||
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()
|
||||
|
||||
Reference in New Issue
Block a user