diff --git a/crates/e2e_test/src/lib.rs b/crates/e2e_test/src/lib.rs index 4eb0acff1..96ce1bfa0 100644 --- a/crates/e2e_test/src/lib.rs +++ b/crates/e2e_test/src/lib.rs @@ -142,4 +142,7 @@ mod replication_extension_test; #[cfg(test)] mod snowball_auto_extract_test; +#[cfg(test)] +mod namespace_lock_quorum_test; + pub mod tls_gen; diff --git a/crates/e2e_test/src/namespace_lock_quorum_test.rs b/crates/e2e_test/src/namespace_lock_quorum_test.rs new file mode 100644 index 000000000..fc3c6b0fb --- /dev/null +++ b/crates/e2e_test/src/namespace_lock_quorum_test.rs @@ -0,0 +1,122 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use crate::common::RustFSTestClusterEnvironment; +use aws_sdk_s3::Client; +use aws_sdk_s3::error::SdkError; +use bytes::Bytes; +use serial_test::serial; +use std::sync::Arc; +use tokio::sync::Barrier; +use tracing::{info, warn}; + +const BUCKET: &str = "namespace-lock-quorum-bucket"; +const KEY: &str = "thumb/79/concurrent-overwrite.jpg"; + +type TestResult = Result<(), Box>; + +async fn put_object(client: Client, payload: Vec, writer_id: usize) -> Result<(), String> { + client + .put_object() + .bucket(BUCKET) + .key(KEY) + .body(Bytes::from(payload).into()) + .send() + .await + .map(|_| ()) + .map_err(|err| format_s3_error(err, writer_id)) +} + +fn format_s3_error(err: SdkError, writer_id: usize) -> String { + match err { + SdkError::ServiceError(service_err) => { + let err = service_err.err(); + let code = err.meta().code().unwrap_or(""); + let message = err.meta().message().unwrap_or(""); + format!("writer {writer_id} returned service error {code}: {message}") + } + other => format!("writer {writer_id} returned SDK error: {other:?}"), + } +} + +#[tokio::test] +#[serial] +async fn test_concurrent_cluster_overwrites_do_not_fail_namespace_lock_quorum() -> TestResult { + crate::common::init_logging(); + info!("Starting namespace lock quorum regression test with auto cluster"); + + let mut cluster = RustFSTestClusterEnvironment::new(4).await?; + // Keep the regression focused on false quorum-loss errors, not ordinary lock + // wait exhaustion under a heavily contended same-key overwrite workload. + cluster.set_env("RUSTFS_OBJECT_LOCK_ACQUIRE_TIMEOUT", "20"); + cluster.start().await?; + cluster.create_test_bucket(BUCKET).await?; + + let clients = cluster.create_all_clients()?; + let first_payload = b"initial object contents".to_vec(); + put_object(clients[0].clone(), first_payload, 0).await?; + + let writer_count = clients.len() * 2; + let barrier = Arc::new(Barrier::new(writer_count)); + let mut handles = Vec::with_capacity(writer_count); + + for writer_id in 0..writer_count { + let client = clients[writer_id % clients.len()].clone(); + let barrier = barrier.clone(); + let payload = format!("replacement payload from writer {writer_id:02}").into_bytes(); + handles.push(tokio::spawn(async move { + barrier.wait().await; + put_object(client, payload, writer_id).await + })); + } + + let mut failures = Vec::new(); + for handle in handles { + match handle.await { + Ok(Ok(())) => {} + Ok(Err(err)) => failures.push(err), + Err(err) => failures.push(format!("writer task join failed: {err}")), + } + } + + if !failures.is_empty() { + for failure in &failures { + warn!("concurrent overwrite failure: {}", failure); + } + } + + assert!( + failures.is_empty(), + "concurrent overwrites must not surface namespace lock quorum failures: {failures:#?}" + ); + + let body = clients[0] + .get_object() + .bucket(BUCKET) + .key(KEY) + .send() + .await? + .body + .collect() + .await? + .into_bytes(); + let body = std::str::from_utf8(&body)?; + assert!( + body.starts_with("replacement payload from writer "), + "final object body should be one of the successful overwrite payloads, got {body:?}" + ); + + clients[0].delete_object().bucket(BUCKET).key(KEY).send().await?; + Ok(()) +} diff --git a/crates/ecstore/Cargo.toml b/crates/ecstore/Cargo.toml index bf298ab3c..bb40c626c 100644 --- a/crates/ecstore/Cargo.toml +++ b/crates/ecstore/Cargo.toml @@ -131,7 +131,7 @@ aws-smithy-http-client.workspace = true metrics = { workspace = true } [dev-dependencies] -tokio = { workspace = true, features = ["rt-multi-thread", "macros"] } +tokio = { workspace = true, features = ["rt-multi-thread", "macros", "test-util"] } criterion = { workspace = true, features = ["html_reports"] } temp-env = { workspace = true, features = ["async_closure"] } tracing-subscriber = { workspace = true } diff --git a/crates/ecstore/src/batch_processor.rs b/crates/ecstore/src/batch_processor.rs index 7d2cc11d0..1c678dea8 100644 --- a/crates/ecstore/src/batch_processor.rs +++ b/crates/ecstore/src/batch_processor.rs @@ -274,10 +274,10 @@ mod tests { assert!(successes.len() >= 2); } - #[tokio::test] + #[tokio::test(start_paused = true)] async fn test_batch_processor_quorum_returns_before_slow_tail() { let processor = AsyncBatchProcessor::new(4); - let started = std::time::Instant::now(); + let started = tokio::time::Instant::now(); let tasks: Vec<_> = [(10_u64, Ok(1_i32)), (15, Ok(2)), (250, Ok(3))] .into_iter() @@ -295,10 +295,10 @@ mod tests { assert!(started.elapsed() < Duration::from_millis(100)); } - #[tokio::test] + #[tokio::test(start_paused = true)] async fn test_batch_processor_quorum_fails_once_quorum_becomes_impossible() { let processor = AsyncBatchProcessor::new(4); - let started = std::time::Instant::now(); + let started = tokio::time::Instant::now(); let tasks: Vec<_> = vec![ (10_u64, Ok(1_i32)), diff --git a/crates/ecstore/src/rpc/remote_locker.rs b/crates/ecstore/src/rpc/remote_locker.rs index 9cea3154a..0926e8966 100644 --- a/crates/ecstore/src/rpc/remote_locker.rs +++ b/crates/ecstore/src/rpc/remote_locker.rs @@ -91,8 +91,8 @@ impl RemoteClient { where F: std::future::Future>, { - let timeout_duration = Self::rpc_timeout(timeout_duration); - match timeout(timeout_duration, future).await { + let lock_timeout = Self::rpc_timeout(timeout_duration); + match timeout(lock_timeout, future).await { Ok(Ok(response)) => Ok(response), Ok(Err(err)) => { let reason = err.to_string(); @@ -100,30 +100,21 @@ impl RemoteClient { Err(LockError::internal(format!("{op} RPC failed: {reason}"))) } Err(_) => { - let reason = format!("RPC timed out after {:?}", timeout_duration); - warn!( - addr = %self.addr, - op, - reason, - "Remote lock RPC timed out without evicting cached connection" - ); - Err(LockError::timeout(format!("remote lock RPC {op} on {}", self.addr), timeout_duration)) + let reason = format!("RPC timed out after {:?}", lock_timeout); + self.evict_connection(op, &reason).await; + Err(LockError::timeout(format!("remote lock RPC {op} on {}", self.addr), lock_timeout)) } } } - fn timeout_failure_response(request: &LockRequest) -> LockResponse { - LockResponse::failure("Lock acquisition timeout", request.acquire_timeout) + fn rpc_timeout_failure_response(request: &LockRequest, err: &LockError) -> LockResponse { + LockResponse::failure(format!("Remote lock RPC timed out: {err}"), request.acquire_timeout) } fn rpc_failure_response(_request: &LockRequest, err: &LockError) -> LockResponse { LockResponse::failure(format!("Remote lock RPC failed: {err}"), Duration::ZERO) } - fn timeout_failure_batch(requests: &[LockRequest]) -> Vec { - requests.iter().map(Self::timeout_failure_response).collect() - } - fn rpc_failure_batch(requests: &[LockRequest], err: &LockError) -> Vec { requests .iter() @@ -131,6 +122,13 @@ impl RemoteClient { .collect() } + fn rpc_timeout_failure_batch(requests: &[LockRequest], err: &LockError) -> Vec { + requests + .iter() + .map(|request| Self::rpc_timeout_failure_response(request, err)) + .collect() + } + fn batch_rpc_timeout(requests: &[LockRequest]) -> Duration { requests .iter() @@ -191,7 +189,7 @@ impl LockClient for RemoteClient { let resp = match self.execute_rpc("lock", request.acquire_timeout, client.lock(req)).await { Ok(resp) => resp.into_inner(), - Err(LockError::Timeout { .. }) => return Ok(Self::timeout_failure_response(request)), + Err(err @ LockError::Timeout { .. }) => return Ok(Self::rpc_timeout_failure_response(request, &err)), Err(err) => return Ok(Self::rpc_failure_response(request, &err)), }; @@ -231,7 +229,7 @@ impl LockClient for RemoteClient { .await { Ok(resp) => resp.into_inner(), - Err(LockError::Timeout { .. }) => return Ok(Self::timeout_failure_batch(requests)), + Err(err @ LockError::Timeout { .. }) => return Ok(Self::rpc_timeout_failure_batch(requests, &err)), Err(err) => return Ok(Self::rpc_failure_batch(requests, &err)), }; @@ -499,7 +497,7 @@ mod tests { } #[tokio::test] - async fn test_remote_client_acquire_lock_respects_request_timeout_without_evicting_connection() { + async fn test_remote_client_acquire_lock_respects_request_timeout_and_evicts_connection() { ensure_test_rpc_secret(); let (addr, accept_task) = spawn_hanging_listener().await; cache_lazy_channel(&addr).await; @@ -516,17 +514,24 @@ mod tests { "remote lock RPC should honor request timeout" ); assert!(!response.success, "timed out lock acquisition should fail"); - assert_eq!(response.error.as_deref(), Some("Lock acquisition timeout")); assert!( - GLOBAL_CONN_MAP.read().await.contains_key(&addr), - "local timeout should keep cached connection" + response + .error + .as_deref() + .is_some_and(|error| error.contains("Remote lock RPC timed out")), + "expected remote RPC timeout marker, got {:?}", + response.error + ); + assert!( + !GLOBAL_CONN_MAP.read().await.contains_key(&addr), + "timeout should evict cached connection" ); accept_task.abort(); } #[tokio::test] - async fn test_remote_client_acquire_locks_batch_respects_request_timeout_without_evicting_connection() { + async fn test_remote_client_acquire_locks_batch_respects_request_timeout_and_evicts_connection() { ensure_test_rpc_secret(); let (addr, accept_task) = spawn_hanging_listener().await; cache_lazy_channel(&addr).await; @@ -544,35 +549,22 @@ mod tests { ); assert_eq!(responses.len(), 1); assert!(!responses[0].success, "timed out batch lock acquisition should fail"); - assert_eq!(responses[0].error.as_deref(), Some("Lock acquisition timeout")); assert!( - GLOBAL_CONN_MAP.read().await.contains_key(&addr), - "local batch timeout should keep cached connection" + responses[0] + .error + .as_deref() + .is_some_and(|error| error.contains("Remote lock RPC timed out")), + "expected remote RPC timeout marker, got {:?}", + responses[0].error + ); + assert!( + !GLOBAL_CONN_MAP.read().await.contains_key(&addr), + "batch timeout should evict cached connection" ); accept_task.abort(); } - #[tokio::test] - async fn test_remote_client_rpc_status_error_evicts_connection() { - let addr = "http://127.0.0.1:1".to_string(); - cache_lazy_channel(&addr).await; - assert!(GLOBAL_CONN_MAP.read().await.contains_key(&addr)); - - let client = RemoteClient::new(addr.clone()); - let result = client - .execute_rpc("lock", Duration::from_millis(50), async { - Err::<(), _>(tonic::Status::unavailable("connection unavailable")) - }) - .await; - - assert!(result.is_err(), "RPC status errors should be returned"); - assert!( - !GLOBAL_CONN_MAP.read().await.contains_key(&addr), - "RPC status errors should evict cached connection" - ); - } - #[test] fn test_remote_client_zero_timeout_is_clamped() { assert_eq!(RemoteClient::rpc_timeout(Duration::ZERO), Duration::from_millis(1)); diff --git a/crates/lock/src/distributed_lock.rs b/crates/lock/src/distributed_lock.rs index 30a930cc3..d475b0710 100644 --- a/crates/lock/src/distributed_lock.rs +++ b/crates/lock/src/distributed_lock.rs @@ -32,6 +32,16 @@ const UNLOCK_RETRY_ATTEMPTS: usize = 3; const UNLOCK_RETRY_BACKOFF: Duration = Duration::from_millis(100); const LOCK_ACQUIRE_RETRY_ATTEMPTS: usize = 3; const LOCK_ACQUIRE_RETRY_INITIAL_BACKOFF: Duration = Duration::from_millis(250); +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)] +enum LockAcquireFailureKind { + NonRetryable, + RetryableContention, + UnrecoverableQuorum, +} /// Generate a new aggregate lock ID for multiple client locks fn generate_aggregate_lock_id(resource: &ObjectKey) -> LockId { @@ -41,6 +51,35 @@ fn generate_aggregate_lock_id(resource: &ObjectKey) -> LockId { } } +fn is_remote_lock_rpc_failure(error: &str) -> bool { + has_case_insensitive_prefix(error, REMOTE_LOCK_RPC_FAILED_PREFIX) + || has_case_insensitive_prefix(error, REMOTE_LOCK_RPC_TIMED_OUT_PREFIX) +} + +fn has_case_insensitive_prefix(error: &str, expected_prefix: &str) -> bool { + error + .get(0..expected_prefix.len()) + .is_some_and(|prefix| prefix.eq_ignore_ascii_case(expected_prefix)) +} + +fn is_unrecoverable_quorum_error(error: &str) -> bool { + error + .get(0..UNRECOVERABLE_QUORUM_FAILURE_PREFIX.len()) + .is_some_and(|prefix| prefix.eq_ignore_ascii_case(UNRECOVERABLE_QUORUM_FAILURE_PREFIX)) +} + +fn classify_lock_failure(error: &str) -> LockAcquireFailureKind { + if is_unrecoverable_quorum_error(error) { + return LockAcquireFailureKind::UnrecoverableQuorum; + } + + if error.to_ascii_lowercase().contains("timeout") || is_remote_lock_rpc_failure(error) { + return LockAcquireFailureKind::RetryableContention; + } + + LockAcquireFailureKind::NonRetryable +} + /// A RAII guard for distributed locks that releases the lock asynchronously when dropped. #[derive(Debug)] pub struct DistributedLockGuard { @@ -136,7 +175,7 @@ type LockAcquireTaskResult = (usize, Result); struct LockAcquireQuorumResult { response: LockResponse, individual_locks: Vec<(LockId, Arc)>, - pending_cleanup_spawned: bool, + failure_kind: Option, } impl DistributedLock { @@ -196,6 +235,7 @@ impl DistributedLock { let LockAcquireQuorumResult { response: resp, individual_locks, + failure_kind, .. } = self.acquire_lock_quorum_with_retry(request).await?; if resp.success { @@ -219,26 +259,36 @@ impl DistributedLock { "acquire_lock_quorum contention: {}", error_msg ); - } else { + } else if matches!(failure_kind, Some(LockAcquireFailureKind::NonRetryable)) { warn!( resource = %request.resource, owner = %request.owner, "acquire_lock_quorum error: {}", error_msg ); + } else if matches!(failure_kind, Some(LockAcquireFailureKind::RetryableContention)) { + debug!( + resource = %request.resource, + owner = %request.owner, + "acquire_lock_quorum contention: {}", + error_msg + ); + } else { + debug!( + resource = %request.resource, + owner = %request.owner, + "acquire_lock_quorum final failure: {}", + error_msg + ); } - if error_msg.contains("quorum") { - // This is a quorum failure - return appropriate error + + if matches!(failure_kind, Some(LockAcquireFailureKind::UnrecoverableQuorum)) { let achieved = individual_locks.len(); Err(LockError::QuorumNotReached { required: required_quorum, achieved, }) - } else if error_msg.contains("timeout") || resp.wait_time >= request.acquire_timeout { - // This is a timeout - return None so caller can convert to timeout error - Ok(None) } else { - // Other failure - return None for backward compatibility Ok(None) } } else { @@ -306,11 +356,7 @@ impl DistributedLock { fn is_retryable_lock_failure(resp: &LockResponse) -> bool { resp.error .as_deref() - .map(|error| { - let error = error.to_ascii_lowercase(); - error.contains("timeout") || error.contains("rpc failed") - }) - .unwrap_or(false) + .is_some_and(|error| matches!(classify_lock_failure(error), LockAcquireFailureKind::RetryableContention)) } async fn release_entries(entries: Vec<(LockId, Arc)>, context: &'static str) { @@ -452,7 +498,6 @@ impl DistributedLock { let result = self.acquire_lock_quorum_once(&attempt_request).await?; if result.response.success || !result.individual_locks.is_empty() - || result.pending_cleanup_spawned || !Self::is_retryable_lock_failure(&result.response) || attempt >= LOCK_ACQUIRE_RETRY_ATTEMPTS { @@ -472,7 +517,7 @@ impl DistributedLock { Ok(last_result.unwrap_or_else(|| LockAcquireQuorumResult { response: LockResponse::failure("Lock acquisition timeout", request.acquire_timeout), individual_locks: Vec::new(), - pending_cleanup_spawned: false, + failure_kind: Some(LockAcquireFailureKind::RetryableContention), })) } @@ -485,6 +530,7 @@ 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 hard_failures = 0usize; while let Some(join_result) = pending.join_next().await { match join_result { @@ -503,22 +549,53 @@ impl DistributedLock { } } else { let error = resp.error.unwrap_or_else(|| "unknown error".to_string()); + if is_remote_lock_rpc_failure(&error) { + hard_failures += 1; + } self.log_failed_lock_response(request, idx, error.clone()); last_failure = Some(error); } } Ok((idx, Err(err))) => { + hard_failures += 1; tracing::warn!("Failed to acquire lock on client {}: {}", idx, err); last_failure = Some(err.to_string()); } Err(err) => { + hard_failures += 1; tracing::warn!("Lock acquisition task join failed: {}", err); last_failure = Some(err.to_string()); } } + if self.clients.len().saturating_sub(hard_failures) < required_quorum { + let rollback_count = individual_locks.len(); + Self::spawn_release_cleanup(individual_locks.clone(), "distributed_lock_quorum_rollback"); + if !pending.is_empty() { + Self::spawn_pending_cleanup( + pending, + self.clients.clone(), + fallback_lock_id.clone(), + "distributed_lock_failure_cleanup", + ); + } + + let mut error = format!( + "Unrecoverable quorum failure: {rollback_count}/{required_quorum} required; {hard_failures} clients failed" + ); + if let Some(last_failure) = last_failure { + error.push_str("; last failure: "); + error.push_str(&last_failure); + } + let resp = LockResponse::failure(error, Duration::ZERO); + return Ok(LockAcquireQuorumResult { + response: resp, + individual_locks, + failure_kind: Some(LockAcquireFailureKind::UnrecoverableQuorum), + }); + } + if individual_locks.len() >= required_quorum { - let pending_cleanup_spawned = !pending.is_empty(); if !pending.is_empty() { Self::spawn_pending_cleanup( pending, @@ -555,14 +632,13 @@ impl DistributedLock { return Ok(LockAcquireQuorumResult { response: resp, individual_locks, - pending_cleanup_spawned, + failure_kind: None, }); } - if !individual_locks.is_empty() && individual_locks.len() + pending.len() < required_quorum { + if individual_locks.len() + pending.len() < required_quorum { let rollback_count = individual_locks.len(); Self::spawn_release_cleanup(individual_locks.clone(), "distributed_lock_quorum_rollback"); - let pending_cleanup_spawned = !pending.is_empty(); if !pending.is_empty() { Self::spawn_pending_cleanup( pending, @@ -573,6 +649,14 @@ impl DistributedLock { } let mut error = format!("Failed to acquire quorum: {rollback_count}/{required_quorum} required"); + let failure_kind = if hard_failures > 0 { + error = format!( + "Unrecoverable quorum failure: {rollback_count}/{required_quorum} required; {hard_failures} clients failed; {error}" + ); + LockAcquireFailureKind::UnrecoverableQuorum + } else { + LockAcquireFailureKind::RetryableContention + }; if let Some(last_failure) = last_failure { error.push_str("; last failure: "); error.push_str(&last_failure); @@ -581,7 +665,7 @@ impl DistributedLock { return Ok(LockAcquireQuorumResult { response: resp, individual_locks, - pending_cleanup_spawned, + failure_kind: Some(failure_kind), }); } } @@ -589,15 +673,20 @@ impl DistributedLock { let rollback_count = individual_locks.len(); Self::spawn_release_cleanup(individual_locks.clone(), "distributed_lock_quorum_rollback"); let mut error = format!("Failed to acquire quorum: {rollback_count}/{required_quorum} required"); - if let Some(last_failure) = last_failure { + if let Some(last_failure) = &last_failure { error.push_str("; last failure: "); - error.push_str(&last_failure); + error.push_str(last_failure); } let resp = LockResponse::failure(error, Duration::ZERO); Ok(LockAcquireQuorumResult { response: resp, individual_locks, - pending_cleanup_spawned: false, + failure_kind: Some(if hard_failures > 0 { + LockAcquireFailureKind::UnrecoverableQuorum + } else { + let fallback_kind = last_failure.as_deref().map(classify_lock_failure); + fallback_kind.unwrap_or(LockAcquireFailureKind::RetryableContention) + }), }) } } @@ -617,3 +706,236 @@ fn record_lock_held_release(lock_type: LockType) { LockType::Exclusive => record_write_lock_held_release(), } } + +#[cfg(test)] +mod tests { + use super::{DistributedLock, is_remote_lock_rpc_failure}; + use crate::{LockError, LockId, LockInfo, LockRequest, LockResponse, LockStats, LockType, ObjectKey, client::LockClient}; + use std::{sync::Arc, time::Duration}; + + #[derive(Debug)] + struct ResponseClient { + response: LockResponse, + delay: Duration, + } + + impl ResponseClient { + fn new(response: LockResponse) -> Self { + Self { + response, + delay: Duration::ZERO, + } + } + + fn with_delay(mut self, delay: Duration) -> Self { + self.delay = delay; + self + } + + fn into_client(self) -> Arc { + Arc::new(self) + } + } + + #[async_trait::async_trait] + impl LockClient for ResponseClient { + async fn acquire_lock(&self, _request: &LockRequest) -> crate::Result { + if !self.delay.is_zero() { + tokio::time::sleep(self.delay).await; + } + Ok(self.response.clone()) + } + + async fn release(&self, _lock_id: &LockId) -> crate::Result { + Ok(false) + } + + async fn refresh(&self, _lock_id: &LockId) -> crate::Result { + Ok(false) + } + + async fn force_release(&self, _lock_id: &LockId) -> crate::Result { + Ok(false) + } + + async fn check_status(&self, _lock_id: &LockId) -> crate::Result> { + Ok(None) + } + + async fn get_stats(&self) -> crate::Result { + Ok(LockStats::default()) + } + + async fn close(&self) -> crate::Result<()> { + Ok(()) + } + + async fn is_online(&self) -> bool { + true + } + + async fn is_local(&self) -> bool { + false + } + } + + #[test] + fn test_is_remote_lock_rpc_failure() { + assert!(is_remote_lock_rpc_failure("Remote lock RPC failed: backend unreachable")); + assert!(is_remote_lock_rpc_failure("remote lock rpc failed: temporary network issue")); + assert!(is_remote_lock_rpc_failure("Remote lock RPC timed out: RPC timed out after 50ms")); + assert!(!is_remote_lock_rpc_failure("Lock is already held")); + assert!(!is_remote_lock_rpc_failure("acquired lock failed for other reason")); + } + + #[tokio::test] + async fn acquire_guard_returns_quorum_error_when_rpc_failures_make_quorum_impossible() { + let clients: Vec> = vec![ + ResponseClient::new(LockResponse::failure("Remote lock RPC failed: node unavailable", Duration::ZERO)).into_client(), + ResponseClient::new(LockResponse::failure("lock already held", Duration::ZERO)) + .with_delay(Duration::from_millis(50)) + .into_client(), + ResponseClient::new(LockResponse::failure("Remote lock RPC failed: connection refused", Duration::ZERO)) + .into_client(), + ResponseClient::new(LockResponse::failure("lock already held", Duration::ZERO)) + .with_delay(Duration::from_millis(50)) + .into_client(), + ]; + 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(1)); + + let result = lock.acquire_guard(&request).await; + + assert!( + matches!( + result, + Err(LockError::QuorumNotReached { + required: 3, + achieved: 0 + }) + ), + "unexpected result: {result:?}" + ); + } + + #[tokio::test] + async fn acquire_guard_returns_quorum_error_when_rpc_timeouts_make_quorum_impossible() { + let clients: Vec> = vec![ + ResponseClient::new(LockResponse::failure( + "Remote lock RPC timed out: RPC timed out after 50ms", + Duration::ZERO, + )) + .into_client(), + ResponseClient::new(LockResponse::failure("lock already held", Duration::ZERO)) + .with_delay(Duration::from_millis(50)) + .into_client(), + ResponseClient::new(LockResponse::failure( + "Remote lock RPC timed out: RPC timed out after 50ms", + Duration::ZERO, + )) + .into_client(), + ResponseClient::new(LockResponse::failure("lock already held", Duration::ZERO)) + .with_delay(Duration::from_millis(50)) + .into_client(), + ]; + 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(1)); + + let result = lock.acquire_guard(&request).await; + + assert!( + matches!( + result, + Err(LockError::QuorumNotReached { + required: 3, + achieved: 0 + }) + ), + "unexpected result: {result:?}" + ); + } + + #[tokio::test] + async fn acquire_guard_returns_timeout_when_zero_locks_make_quorum_impossible_for_attempt() { + let clients: Vec> = vec![ + ResponseClient::new(LockResponse::failure("lock already held", Duration::ZERO)).into_client(), + ResponseClient::new(LockResponse::failure("lock already held", Duration::ZERO)).into_client(), + ResponseClient::new(LockResponse::failure("lock already held", Duration::ZERO)) + .with_delay(Duration::from_secs(1)) + .into_client(), + ResponseClient::new(LockResponse::failure("lock already held", Duration::ZERO)) + .with_delay(Duration::from_secs(1)) + .into_client(), + ]; + 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 started = tokio::time::Instant::now(); + let result = lock.acquire_guard(&request).await; + + assert!(matches!(result, Ok(None)), "unexpected result: {result:?}"); + assert!( + started.elapsed() < Duration::from_secs(1), + "acquire should fail this attempt before waiting for delayed impossible-quorum tasks" + ); + } + + #[tokio::test] + async fn acquire_guard_retries_transient_timeout_before_quorum() { + let clients: Vec> = vec![ + ResponseClient::new(LockResponse::failure( + "Timeout { resource: \"bucket/object-flaky-acquire@latest\", timeout: 1s }", + Duration::ZERO, + )) + .into_client(), + ResponseClient::new(LockResponse::failure( + "Timeout { resource: \"bucket/object-flaky-acquire@latest\", timeout: 1s }", + Duration::ZERO, + )) + .into_client(), + ResponseClient::new(LockResponse::failure( + "Timeout { resource: \"bucket/object-flaky-acquire@latest\", timeout: 1s }", + Duration::ZERO, + )) + .into_client(), + ResponseClient::new(LockResponse::failure( + "Timeout { resource: \"bucket/object-flaky-acquire@latest\", timeout: 1s }", + Duration::ZERO, + )) + .into_client(), + ]; + 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_millis(900)); + + let started = tokio::time::Instant::now(); + let result = lock.acquire_guard(&request).await; + let elapsed = started.elapsed(); + + assert!(matches!(result, Ok(None)), "unexpected result: {result:?}"); + assert!( + elapsed >= Duration::from_millis(250), + "expected at least one retry attempt for transient timeout, got {elapsed:?}" + ); + } + + #[tokio::test] + async fn acquire_guard_returns_timeout_when_quorum_remains_contended() { + let clients: Vec> = vec![ + ResponseClient::new(LockResponse::failure("lock already held", Duration::ZERO)).into_client(), + ResponseClient::new(LockResponse::failure("lock already held", Duration::ZERO)).into_client(), + ResponseClient::new(LockResponse::failure("lock already held", Duration::ZERO)).into_client(), + ResponseClient::new(LockResponse::failure("lock already held", Duration::ZERO)).into_client(), + ]; + 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_millis(120)); + + let result = lock.acquire_guard(&request).await; + + assert!(matches!(result, Ok(None)), "unexpected result: {result:?}"); + } +} diff --git a/crates/lock/src/namespace/tests.rs b/crates/lock/src/namespace/tests.rs index 06d981e4f..a87fb863b 100644 --- a/crates/lock/src/namespace/tests.rs +++ b/crates/lock/src/namespace/tests.rs @@ -64,6 +64,50 @@ impl crate::client::LockClient for FailingClient { } } +#[derive(Debug)] +struct FailureResponseClient { + error: &'static str, +} + +#[async_trait::async_trait] +impl crate::client::LockClient for FailureResponseClient { + async fn acquire_lock(&self, _request: &LockRequest) -> crate::Result { + Ok(LockResponse::failure(self.error, Duration::ZERO)) + } + + async fn release(&self, _lock_id: &LockId) -> crate::Result { + Ok(false) + } + + async fn refresh(&self, _lock_id: &LockId) -> crate::Result { + Ok(false) + } + + async fn force_release(&self, _lock_id: &LockId) -> crate::Result { + Ok(false) + } + + async fn check_status(&self, _lock_id: &LockId) -> crate::Result> { + Ok(None) + } + + async fn get_stats(&self) -> crate::Result { + Ok(LockStats::default()) + } + + async fn close(&self) -> crate::Result<()> { + Ok(()) + } + + async fn is_online(&self) -> bool { + true + } + + async fn is_local(&self) -> bool { + false + } +} + #[derive(Debug)] struct DelayedClient { inner: Arc, @@ -702,10 +746,10 @@ async fn test_namespace_lock_distributed_eight_node_write_releases_all_nodes() { .get_write_lock(resource.clone(), "owner-b", Duration::from_millis(100)) .await .expect_err("owner-b should not acquire while owner-a holds all node locks"); - let err_str = err.to_string(); + let err_str = err.to_string().to_lowercase(); assert!( - err_str.contains("required 5") && err_str.contains("achieved"), - "expected 8-node quorum failure below required write quorum, got: {err}" + err_str.contains("timeout"), + "expected owner-b contention to exhaust acquire timeout, got: {err}" ); assert!(guard.release(), "distributed guard should enqueue release"); @@ -877,6 +921,68 @@ 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() { + let manager = Arc::new(GlobalLockManager::new()); + let client_ok: Arc = Arc::new(LocalClient::with_manager(manager)); + let client_rpc_failed: Arc = Arc::new(FailureResponseClient { + error: "Remote lock RPC failed: connection refused", + }); + let client_rpc_timed_out: Arc = Arc::new(FailureResponseClient { + error: "Remote lock RPC timed out: RPC timed out after 50ms", + }); + let client_rpc_failed_2: Arc = Arc::new(FailureResponseClient { + error: "Remote lock RPC failed: transport error", + }); + let clients: Vec> = vec![client_ok, client_rpc_failed, client_rpc_timed_out, client_rpc_failed_2]; + let lock = NamespaceLock::with_clients("remote-rpc-hard-failure".to_string(), clients); + let resource = create_test_object_key("bucket", "object-rpc-hard-failure"); + + let started = tokio::time::Instant::now(); + 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"); + + assert!( + started.elapsed() < Duration::from_millis(150), + "remote RPC failures should not retry until the full acquire timeout" + ); + let err_str = err.to_string().to_lowercase(); + assert!( + err_str.contains("quorum") || err_str.contains("not reached"), + "expected hard quorum failure, got: {err}" + ); +} + +#[tokio::test] +async fn test_namespace_lock_distributed_contention_quorum_miss_times_out() { + let client_timeout_1: Arc = Arc::new(FailureResponseClient { + error: "Lock acquisition timeout", + }); + let client_timeout_2: Arc = Arc::new(FailureResponseClient { + error: "Lock acquisition timeout", + }); + let client_timeout_3: Arc = Arc::new(FailureResponseClient { + error: "Lock acquisition timeout", + }); + let clients: Vec> = vec![client_timeout_1, client_timeout_2, client_timeout_3]; + let lock = NamespaceLock::with_clients("contention-timeout".to_string(), clients); + let resource = create_test_object_key("bucket", "object-contention-timeout"); + + let err = lock + .get_write_lock(resource, "owner-a", Duration::from_millis(40)) + .await + .expect_err("ordinary contention should exhaust acquire timeout"); + + let err_str = err.to_string().to_lowercase(); + assert!(err_str.contains("timeout"), "expected timeout after contention retries, got: {err}"); + assert!( + !err_str.contains("quorum not reached"), + "ordinary contention should not surface as quorum loss: {err}" + ); +} + #[tokio::test] async fn test_namespace_lock_distributed_quorum_failure_rolls_back_successful_nodes() { let manager1 = Arc::new(GlobalLockManager::new());