From d0a965f2eeacb661c719ea8614f09225226e2376 Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Sun, 5 Jul 2026 03:01:20 +0800 Subject: [PATCH] fix(lock): harden transient quorum and diagnostics (#4268) * fix(lock): retry transient remote quorum gaps * perf(storage): reduce deadlock detector blocking * fix(storage): satisfy deadlock detector clippy --- Cargo.lock | 1 + crates/lock/src/distributed_lock.rs | 174 ++++++++++++++++++++++-- crates/lock/src/namespace/tests.rs | 16 +-- rustfs/Cargo.toml | 1 + rustfs/src/storage/deadlock_detector.rs | 51 +++---- 5 files changed, 197 insertions(+), 46 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 34fef2dca..b7192728a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -8802,6 +8802,7 @@ dependencies = [ "mime_guess", "opentelemetry", "opentelemetry_sdk", + "parking_lot", "percent-encoding", "pin-project-lite", "proptest", diff --git a/crates/lock/src/distributed_lock.rs b/crates/lock/src/distributed_lock.rs index 7b4cb1af9..a056b3b8a 100644 --- a/crates/lock/src/distributed_lock.rs +++ b/crates/lock/src/distributed_lock.rs @@ -36,7 +36,7 @@ const REMOTE_LOCK_RPC_FAILED_PREFIX: &str = "remote lock rpc failed:"; const REMOTE_LOCK_RPC_TIMED_OUT_PREFIX: &str = "remote lock rpc timed out:"; const UNRECOVERABLE_QUORUM_FAILURE_PREFIX: &str = "unrecoverable quorum failure"; -#[derive(Debug)] +#[derive(Clone, Copy, Debug)] enum LockAcquireFailureKind { NonRetryable, RetryableContention, @@ -84,6 +84,14 @@ fn classify_lock_failure(error: &str) -> LockAcquireFailureKind { LockAcquireFailureKind::NonRetryable } +fn classify_lock_error_failure(error: &LockError) -> LockAcquireFailureKind { + if error.is_retryable() { + LockAcquireFailureKind::RetryableContention + } else { + classify_lock_failure(&error.to_string()) + } +} + fn should_warn_lock_failure(error: &str) -> bool { if is_remote_lock_rpc_failure(error) { return !is_remote_lock_rpc_timeout(error); @@ -191,6 +199,7 @@ struct LockAcquireQuorumResult { response: LockResponse, individual_locks: Vec<(LockId, Arc)>, failure_kind: Option, + quorum_impossible: bool, } impl DistributedLock { @@ -528,11 +537,19 @@ impl DistributedLock { attempt += 1; } - Ok(last_result.unwrap_or_else(|| LockAcquireQuorumResult { - response: LockResponse::failure("Lock acquisition timeout", request.acquire_timeout), - individual_locks: Vec::new(), - failure_kind: Some(LockAcquireFailureKind::RetryableContention), - })) + if let Some(mut result) = last_result { + if result.quorum_impossible { + result.failure_kind = Some(LockAcquireFailureKind::UnrecoverableQuorum); + } + Ok(result) + } else { + Ok(LockAcquireQuorumResult { + response: LockResponse::failure("Lock acquisition timeout", request.acquire_timeout), + individual_locks: Vec::new(), + failure_kind: Some(LockAcquireFailureKind::RetryableContention), + quorum_impossible: false, + }) + } } fn lock_acquire_timeout_result(timeout: Duration) -> LockAcquireQuorumResult { @@ -540,6 +557,7 @@ impl DistributedLock { response: LockResponse::failure("Lock acquisition timeout", timeout), individual_locks: Vec::new(), failure_kind: Some(LockAcquireFailureKind::RetryableContention), + quorum_impossible: false, } } @@ -547,8 +565,8 @@ impl DistributedLock { timeout: Duration, individual_locks: Vec<(LockId, Arc)>, last_failure: Option, + last_failure_kind: Option, ) -> LockAcquireQuorumResult { - let last_failure_kind = last_failure.as_deref().map(classify_lock_failure); let failure_kind = if let Some(kind @ (LockAcquireFailureKind::NonRetryable | LockAcquireFailureKind::UnrecoverableQuorum)) = last_failure_kind @@ -568,6 +586,7 @@ impl DistributedLock { response: LockResponse::failure(error, timeout), individual_locks, failure_kind: Some(failure_kind), + quorum_impossible: matches!(failure_kind, LockAcquireFailureKind::UnrecoverableQuorum), } } @@ -580,6 +599,8 @@ impl DistributedLock { let mut individual_locks: Vec<(LockId, Arc)> = Vec::new(); let fallback_lock_id = request.lock_id.clone(); let mut last_failure = None; + let mut last_failure_kind = None; + let mut last_hard_failure_kind = None; let mut hard_failures = 0usize; let start = tokio::time::Instant::now(); @@ -597,6 +618,7 @@ impl DistributedLock { request.acquire_timeout, individual_locks, last_failure, + last_failure_kind, )); } @@ -615,6 +637,7 @@ impl DistributedLock { request.acquire_timeout, individual_locks, last_failure, + last_failure_kind, )); } }; @@ -635,22 +658,30 @@ impl DistributedLock { } } else { let error = resp.error.unwrap_or_else(|| "unknown error".to_string()); + let failure_kind = classify_lock_failure(&error); if is_remote_lock_rpc_failure(&error) && !is_remote_lock_rpc_timeout(&error) { hard_failures += 1; + last_hard_failure_kind = Some(failure_kind); } self.log_failed_lock_response(request, idx, error.clone()); last_failure = Some(error); + last_failure_kind = Some(failure_kind); } } Ok((idx, Err(err))) => { hard_failures += 1; + let failure_kind = classify_lock_error_failure(&err); tracing::warn!("Failed to acquire lock on client {}: {}", idx, err); last_failure = Some(err.to_string()); + last_failure_kind = Some(failure_kind); + last_hard_failure_kind = Some(failure_kind); } Err(err) => { hard_failures += 1; tracing::warn!("Lock acquisition task join failed: {}", err); last_failure = Some(err.to_string()); + last_failure_kind = Some(LockAcquireFailureKind::NonRetryable); + last_hard_failure_kind = Some(LockAcquireFailureKind::NonRetryable); } } @@ -674,10 +705,12 @@ impl DistributedLock { error.push_str(&last_failure); } let resp = LockResponse::failure(error, Duration::ZERO); + let failure_kind = last_hard_failure_kind.unwrap_or(LockAcquireFailureKind::UnrecoverableQuorum); return Ok(LockAcquireQuorumResult { response: resp, individual_locks, - failure_kind: Some(LockAcquireFailureKind::UnrecoverableQuorum), + failure_kind: Some(failure_kind), + quorum_impossible: true, }); } @@ -719,6 +752,7 @@ impl DistributedLock { response: resp, individual_locks, failure_kind: None, + quorum_impossible: false, }); } @@ -735,13 +769,13 @@ impl DistributedLock { } let mut error = format!("Failed to acquire quorum: {rollback_count}/{required_quorum} required"); - let failure_kind = if hard_failures > 0 { + let (failure_kind, quorum_impossible) = if hard_failures > 0 { error = format!( "Unrecoverable quorum failure: {rollback_count}/{required_quorum} required; {hard_failures} clients failed; {error}" ); - LockAcquireFailureKind::UnrecoverableQuorum + (last_hard_failure_kind.unwrap_or(LockAcquireFailureKind::UnrecoverableQuorum), true) } else { - LockAcquireFailureKind::RetryableContention + (LockAcquireFailureKind::RetryableContention, false) }; if let Some(last_failure) = last_failure { error.push_str("; last failure: "); @@ -752,6 +786,7 @@ impl DistributedLock { response: resp, individual_locks, failure_kind: Some(failure_kind), + quorum_impossible, }); } } @@ -768,11 +803,11 @@ impl DistributedLock { response: resp, individual_locks, failure_kind: Some(if hard_failures > 0 { - LockAcquireFailureKind::UnrecoverableQuorum + last_hard_failure_kind.unwrap_or(LockAcquireFailureKind::UnrecoverableQuorum) } else { - let fallback_kind = last_failure.as_deref().map(classify_lock_failure); - fallback_kind.unwrap_or(LockAcquireFailureKind::RetryableContention) + last_failure_kind.unwrap_or(LockAcquireFailureKind::RetryableContention) }), + quorum_impossible: hard_failures > 0, }) } } @@ -932,6 +967,7 @@ mod tests { enum AcquirePlan { Success { delay: Duration }, Failure { error: &'static str, delay: Duration }, + ClientError { message: &'static str, delay: Duration }, } #[derive(Debug)] @@ -993,6 +1029,12 @@ mod tests { } Ok(LockResponse::failure(error, Duration::ZERO)) } + AcquirePlan::ClientError { message, delay } => { + if !delay.is_zero() { + tokio::time::sleep(delay).await; + } + Err(LockError::internal(message)) + } } } @@ -1113,6 +1155,110 @@ mod tests { ); } + #[tokio::test] + async fn acquire_guard_retries_remote_rpc_failures_that_temporarily_preclude_quorum() { + let clients: Vec> = vec![ + Arc::new(SequencedClient::new( + vec![ + AcquirePlan::Failure { + error: "Remote lock RPC failed: node unavailable", + delay: Duration::ZERO, + }, + AcquirePlan::Success { delay: Duration::ZERO }, + ], + Arc::new(Mutex::new(Vec::new())), + )), + Arc::new(SequencedClient::new( + vec![ + AcquirePlan::Failure { + error: "Remote lock RPC failed: connection reset", + delay: Duration::ZERO, + }, + AcquirePlan::Success { delay: Duration::ZERO }, + ], + Arc::new(Mutex::new(Vec::new())), + )), + Arc::new(SequencedClient::new( + vec![ + AcquirePlan::Success { delay: Duration::ZERO }, + AcquirePlan::Success { delay: Duration::ZERO }, + ], + Arc::new(Mutex::new(Vec::new())), + )), + Arc::new(SequencedClient::new( + vec![ + AcquirePlan::Success { delay: Duration::ZERO }, + AcquirePlan::Success { delay: Duration::ZERO }, + ], + Arc::new(Mutex::new(Vec::new())), + )), + ]; + let lock = DistributedLock::new("test".to_string(), clients, 3); + let request = LockRequest::new(ObjectKey::new("bucket", "object"), LockType::Exclusive, "owner") + .with_acquire_timeout(Duration::from_secs(2)); + + let guard = lock + .acquire_guard(&request) + .await + .expect("transient remote RPC quorum gap should retry") + .expect("second attempt should acquire quorum"); + + assert!(guard.entries.len() >= 3, "retry should acquire quorum entries"); + drop(guard); + } + + #[tokio::test] + async fn acquire_guard_retries_client_errors_that_temporarily_preclude_quorum() { + let clients: Vec> = vec![ + Arc::new(SequencedClient::new( + vec![ + AcquirePlan::ClientError { + message: "can not get client, err: temporary transport unavailable", + delay: Duration::ZERO, + }, + AcquirePlan::Success { delay: Duration::ZERO }, + ], + Arc::new(Mutex::new(Vec::new())), + )), + Arc::new(SequencedClient::new( + vec![ + AcquirePlan::ClientError { + message: "can not get client, err: connection pool busy", + delay: Duration::ZERO, + }, + AcquirePlan::Success { delay: Duration::ZERO }, + ], + Arc::new(Mutex::new(Vec::new())), + )), + Arc::new(SequencedClient::new( + vec![ + AcquirePlan::Success { delay: Duration::ZERO }, + AcquirePlan::Success { delay: Duration::ZERO }, + ], + Arc::new(Mutex::new(Vec::new())), + )), + Arc::new(SequencedClient::new( + vec![ + AcquirePlan::Success { delay: Duration::ZERO }, + AcquirePlan::Success { delay: Duration::ZERO }, + ], + Arc::new(Mutex::new(Vec::new())), + )), + ]; + let lock = DistributedLock::new("test".to_string(), clients, 3); + let request = LockRequest::new(ObjectKey::new("bucket", "object"), LockType::Exclusive, "owner") + .with_acquire_timeout(Duration::from_secs(2)); + + let guard = lock + .acquire_guard(&request) + .await + .expect("retryable client error quorum gap should retry") + .expect("second attempt should acquire quorum"); + + assert!(guard.entries.len() >= 3, "retry should acquire quorum entries"); + drop(guard); + } + #[tokio::test] async fn acquire_guard_returns_timeout_when_zero_locks_make_quorum_impossible_for_attempt() { let clients: Vec> = vec![ diff --git a/crates/lock/src/namespace/tests.rs b/crates/lock/src/namespace/tests.rs index a492380df..c99d4fecc 100644 --- a/crates/lock/src/namespace/tests.rs +++ b/crates/lock/src/namespace/tests.rs @@ -967,7 +967,7 @@ async fn test_namespace_lock_distributed_write_lock_fails_with_two_nodes_one_off } #[tokio::test] -async fn test_namespace_lock_distributed_remote_rpc_failures_are_hard_quorum_failures() { +async fn test_namespace_lock_distributed_remote_rpc_failures_retry_before_final_quorum_failure() { let manager = Arc::new(GlobalLockManager::new()); let client_ok: Arc = Arc::new(LocalClient::with_manager(manager)); let client_rpc_failed: Arc = Arc::new(FailureResponseClient { @@ -987,16 +987,16 @@ async fn test_namespace_lock_distributed_remote_rpc_failures_are_hard_quorum_fai let err = lock .get_write_lock(resource, "owner-a", Duration::from_secs(1)) .await - .expect_err("write lock should fail as soon as remote RPC failures make quorum impossible"); + .expect_err("persistent remote RPC failures should end as quorum failure"); assert!( - started.elapsed() < Duration::from_millis(150), - "remote RPC failures should not retry until the full acquire timeout" + started.elapsed() >= Duration::from_millis(250), + "remote RPC failures should be retried before final quorum failure" ); let err_str = err.to_string().to_lowercase(); assert!( err_str.contains("quorum") || err_str.contains("not reached"), - "expected hard quorum failure, got: {err}" + "expected final quorum failure, got: {err}" ); } @@ -1191,7 +1191,7 @@ async fn test_namespace_lock_distributed_read_lock_returns_after_quorum_without_ } #[tokio::test] -async fn test_namespace_lock_distributed_failure_returns_early_and_cleans_up_late_successes() { +async fn test_namespace_lock_distributed_failure_retries_and_cleans_up_late_successes() { let manager_fast = Arc::new(GlobalLockManager::new()); let manager_slow = Arc::new(GlobalLockManager::new()); @@ -1216,8 +1216,8 @@ async fn test_namespace_lock_distributed_failure_returns_early_and_cleans_up_lat .expect_err("write lock should fail when quorum becomes impossible"); assert!( - started.elapsed() < Duration::from_millis(150), - "write lock should fail as soon as quorum becomes impossible" + started.elapsed() >= Duration::from_millis(250), + "write lock should retry before final quorum failure" ); let err_str = err.to_string().to_lowercase(); assert!( diff --git a/rustfs/Cargo.toml b/rustfs/Cargo.toml index 97f9f8587..56b9fc00b 100644 --- a/rustfs/Cargo.toml +++ b/rustfs/Cargo.toml @@ -165,6 +165,7 @@ md5.workspace = true mime_guess = { workspace = true } percent-encoding = { workspace = true } pin-project-lite.workspace = true +parking_lot = { workspace = true } rust-embed = { workspace = true, features = ["interpolate-folder-path"] } s3s.workspace = true shadow-rs = { workspace = true, features = ["build", "metadata"] } diff --git a/rustfs/src/storage/deadlock_detector.rs b/rustfs/src/storage/deadlock_detector.rs index 8783660d1..3955ee1b3 100644 --- a/rustfs/src/storage/deadlock_detector.rs +++ b/rustfs/src/storage/deadlock_detector.rs @@ -58,9 +58,10 @@ // Allow dead_code for public API that may be used by external modules or future features #![allow(dead_code)] +use parking_lot::{Mutex, RwLock}; use std::collections::{HashMap, HashSet}; +use std::sync::Arc; use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; -use std::sync::{Arc, Mutex, RwLock}; use std::time::{Duration, Instant}; use tokio::sync::broadcast; use tracing::{debug, error, warn}; @@ -340,13 +341,13 @@ impl DeadlockDetector { running.store(false, Ordering::Relaxed); }); - *self.detector_task.lock().unwrap() = Some(handle); + *self.detector_task.lock() = Some(handle); } /// Stop the detection task. pub fn stop(&self) { let _ = self.shutdown_tx.send(()); - if let Some(handle) = self.detector_task.lock().unwrap().take() { + if let Some(handle) = self.detector_task.lock().take() { // Don't await the handle as we're in a non-async context handle.abort(); } @@ -362,7 +363,7 @@ impl DeadlockDetector { let request_id = request_id.into(); let tracker = RequestResourceTracker::new(request_id.clone(), description); - self.requests.write().unwrap().insert(request_id.clone(), tracker); + self.requests.write().insert(request_id.clone(), tracker); debug!(request_id = %request_id, "Request registered for deadlock tracking"); } @@ -373,7 +374,7 @@ impl DeadlockDetector { return; } - self.requests.write().unwrap().remove(request_id); + self.requests.write().remove(request_id); debug!(request_id = %request_id, "Request unregistered from deadlock tracking"); } @@ -384,7 +385,7 @@ impl DeadlockDetector { return; } - if let Some(tracker) = self.requests.write().unwrap().get_mut(request_id) { + if let Some(tracker) = self.requests.write().get_mut(request_id) { tracker.held_locks.push(lock); } } @@ -395,7 +396,7 @@ impl DeadlockDetector { return; } - if let Some(tracker) = self.requests.write().unwrap().get_mut(request_id) { + if let Some(tracker) = self.requests.write().get_mut(request_id) { tracker.held_locks.retain(|l| &l.id != lock_id); } } @@ -406,7 +407,7 @@ impl DeadlockDetector { return; } - if let Some(tracker) = self.requests.write().unwrap().get_mut(request_id) { + if let Some(tracker) = self.requests.write().get_mut(request_id) { tracker.waiting_lock = Some(lock); } } @@ -417,7 +418,7 @@ impl DeadlockDetector { return; } - if let Some(tracker) = self.requests.write().unwrap().get_mut(request_id) { + if let Some(tracker) = self.requests.write().get_mut(request_id) { tracker.waiting_lock = None; } } @@ -428,14 +429,14 @@ impl DeadlockDetector { return; } - if let Some(tracker) = self.requests.write().unwrap().get_mut(request_id) { + if let Some(tracker) = self.requests.write().get_mut(request_id) { tracker.resources.insert(resource_type, amount); } } /// Get current number of tracked requests. pub fn tracked_count(&self) -> usize { - self.requests.read().unwrap().len() + self.requests.read().len() } /// Get total deadlocks detected. @@ -449,12 +450,14 @@ impl DeadlockDetector { policy: &DeadlockMonitorPolicy, deadlocks_detected: &Arc, ) { - let requests_guard = requests.read().unwrap(); + let hung_request_snapshot: Vec<_> = requests + .read() + .values() + .filter(|r| r.is_hung(policy.hang_threshold)) + .cloned() + .collect(); - // Find hung requests - let hung_requests: Vec<_> = requests_guard.values().filter(|r| r.is_hung(policy.hang_threshold)).collect(); - - if hung_requests.is_empty() { + if hung_request_snapshot.is_empty() { return; } @@ -462,10 +465,10 @@ impl DeadlockDetector { // Edge: request A -> request B means A is waiting for a lock that B holds let mut wait_graph: Vec = Vec::new(); - for waiting in &hung_requests { + for waiting in &hung_request_snapshot { if let Some(waiting_for) = &waiting.waiting_lock { // Find who holds this lock - for holding in &hung_requests { + for holding in &hung_request_snapshot { if holding.request_id == waiting.request_id { continue; } @@ -490,13 +493,13 @@ impl DeadlockDetector { error!( cycle = ?cycle, wait_graph = ?wait_graph, - hung_requests_count = hung_requests.len(), + hung_requests_count = hung_request_snapshot.len(), "Deadlock detected: circular lock wait chain found" ); // Log each request in the cycle for request_id in &cycle { - if let Some(tracker) = requests_guard.get(request_id) { + if let Some(tracker) = hung_request_snapshot.iter().find(|tracker| &tracker.request_id == request_id) { warn!( request_id = %request_id, description = %tracker.description, @@ -510,7 +513,7 @@ impl DeadlockDetector { } else { // No cycle, but log hung requests for diagnosis debug!( - hung_requests_count = hung_requests.len(), + hung_requests_count = hung_request_snapshot.len(), wait_graph_edges = wait_graph.len(), "Hung requests detected but no deadlock cycle" ); @@ -559,9 +562,9 @@ impl DeadlockDetector { if let Some(neighbors) = graph.get(&node) { for neighbor in neighbors { - if path_set.contains(neighbor) { - // Found cycle - trim path to just the cycle - let cycle_start = path.iter().position(|n| *n == *neighbor).unwrap(); + if path_set.contains(neighbor) + && let Some(cycle_start) = path.iter().position(|n| *n == *neighbor) + { path.drain(0..cycle_start); return true; }