fix: retry namespace lock quorum contention (#3098)

* fix: retry namespace lock quorum contention

* fix(lock): treat remote lock rpc failures as hard

* fix(lock): classify lock contention timeout and avoid hard rollback blocks

* fix(lock): handle namespace lock quorum timeouts

* fix(lock): refine quorum contention handling

* fix(lock): refine unrecoverable quorum handling logs

* fix: address namespace lock quorum review feedback

* test: stabilize batch quorum timing tests

* fix(lock): classify remote quorum failures

* fix: drop remote lock timeout grace period

* fix: satisfy clippy on lock quorum error

* fix: classify remote lock rpc timeouts as hard failures

* fix: fast fail impossible lock quorum attempts

* fix: retry on transient timeout with pending cleanup

* fix: evict remote lock timeout connections

* refactor: avoid duplicate timeout eviction warning

---------

Co-authored-by: houseme <housemecn@gmail.com>

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;

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<dyn std::error::Error + Send + Sync>>;
+
+async fn put_object(client: Client, payload: Vec<u8>, 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<aws_sdk_s3::operation::put_object::PutObjectError>, writer_id: usize) -> String {
+    match err {
+        SdkError::ServiceError(service_err) => {
+            let err = service_err.err();
+            let code = err.meta().code().unwrap_or("<unknown>");
+            let message = err.meta().message().unwrap_or("<empty>");
+            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(())
+}

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 }

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)),

crates/ecstore/src/rpc/remote_locker.rs

@@ -91,8 +91,8 @@ impl RemoteClient {
     where
         F: std::future::Future<Output = std::result::Result<T, tonic::Status>>,
     {
-        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<LockResponse> {
-        requests.iter().map(Self::timeout_failure_response).collect()
-    }
-
     fn rpc_failure_batch(requests: &[LockRequest], err: &LockError) -> Vec<LockResponse> {
         requests
             .iter()
@@ -131,6 +122,13 @@ impl RemoteClient {
             .collect()
     }
 
+    fn rpc_timeout_failure_batch(requests: &[LockRequest], err: &LockError) -> Vec<LockResponse> {
+        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));

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<LockResponse>);
 struct LockAcquireQuorumResult {
     response: LockResponse,
     individual_locks: Vec<(LockId, Arc<dyn LockClient>)>,
-    pending_cleanup_spawned: bool,
+    failure_kind: Option<LockAcquireFailureKind>,
 }
 
 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<dyn LockClient>)>, 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<dyn LockClient>)> = 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<dyn LockClient> {
+            Arc::new(self)
+        }
+    }
+
+    #[async_trait::async_trait]
+    impl LockClient for ResponseClient {
+        async fn acquire_lock(&self, _request: &LockRequest) -> crate::Result<LockResponse> {
+            if !self.delay.is_zero() {
+                tokio::time::sleep(self.delay).await;
+            }
+            Ok(self.response.clone())
+        }
+
+        async fn release(&self, _lock_id: &LockId) -> crate::Result<bool> {
+            Ok(false)
+        }
+
+        async fn refresh(&self, _lock_id: &LockId) -> crate::Result<bool> {
+            Ok(false)
+        }
+
+        async fn force_release(&self, _lock_id: &LockId) -> crate::Result<bool> {
+            Ok(false)
+        }
+
+        async fn check_status(&self, _lock_id: &LockId) -> crate::Result<Option<LockInfo>> {
+            Ok(None)
+        }
+
+        async fn get_stats(&self) -> crate::Result<LockStats> {
+            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<Arc<dyn LockClient>> = 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<Arc<dyn LockClient>> = 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<Arc<dyn LockClient>> = 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<Arc<dyn LockClient>> = 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<Arc<dyn LockClient>> = 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:?}");
+    }
+}

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<LockResponse> {
+        Ok(LockResponse::failure(self.error, Duration::ZERO))
+    }
+
+    async fn release(&self, _lock_id: &LockId) -> crate::Result<bool> {
+        Ok(false)
+    }
+
+    async fn refresh(&self, _lock_id: &LockId) -> crate::Result<bool> {
+        Ok(false)
+    }
+
+    async fn force_release(&self, _lock_id: &LockId) -> crate::Result<bool> {
+        Ok(false)
+    }
+
+    async fn check_status(&self, _lock_id: &LockId) -> crate::Result<Option<LockInfo>> {
+        Ok(None)
+    }
+
+    async fn get_stats(&self) -> crate::Result<LockStats> {
+        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<dyn crate::client::LockClient>,
@@ -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<dyn LockClient> = Arc::new(LocalClient::with_manager(manager));
+    let client_rpc_failed: Arc<dyn LockClient> = Arc::new(FailureResponseClient {
+        error: "Remote lock RPC failed: connection refused",
+    });
+    let client_rpc_timed_out: Arc<dyn LockClient> = Arc::new(FailureResponseClient {
+        error: "Remote lock RPC timed out: RPC timed out after 50ms",
+    });
+    let client_rpc_failed_2: Arc<dyn LockClient> = Arc::new(FailureResponseClient {
+        error: "Remote lock RPC failed: transport error",
+    });
+    let clients: Vec<Arc<dyn LockClient>> = 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<dyn LockClient> = Arc::new(FailureResponseClient {
+        error: "Lock acquisition timeout",
+    });
+    let client_timeout_2: Arc<dyn LockClient> = Arc::new(FailureResponseClient {
+        error: "Lock acquisition timeout",
+    });
+    let client_timeout_3: Arc<dyn LockClient> = Arc::new(FailureResponseClient {
+        error: "Lock acquisition timeout",
+    });
+    let clients: Vec<Arc<dyn LockClient>> = 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());