fix(lock): split distributed read and write quorum (#2355)

crates/lock/src/distributed_lock.rs

@@ -174,7 +174,7 @@ pub struct DistributedLock {
     clients: Vec<Arc<dyn LockClient>>,
     /// Namespace identifier
     namespace: String,
-    /// Quorum size for operations (majority for distributed)
+    /// Quorum size for exclusive/write operations
     quorum: usize,
 }
 
@@ -199,6 +199,22 @@ impl DistributedLock {
         &self.namespace
     }
 
+    fn read_quorum(&self) -> usize {
+        let client_count = self.clients.len();
+        if client_count <= 1 {
+            1
+        } else {
+            client_count - (client_count / 2)
+        }
+    }
+
+    fn required_quorum(&self, lock_type: LockType) -> usize {
+        match lock_type {
+            LockType::Shared => self.read_quorum(),
+            LockType::Exclusive => self.quorum,
+        }
+    }
+
     /// Get resource key for this namespace
     pub fn get_resource_key(&self, resource: &ObjectKey) -> String {
         format!("{}:{}", self.namespace, resource)
@@ -215,6 +231,7 @@ impl DistributedLock {
             return Err(LockError::internal("No lock clients available"));
         }
 
+        let required_quorum = self.required_quorum(request.lock_type);
         let (resp, individual_locks) = self.acquire_lock_quorum(request).await?;
         if resp.success {
             // Use aggregate lock_id from LockResponse's LockInfo
@@ -247,10 +264,9 @@ impl DistributedLock {
                 }
                 if error_msg.contains("quorum") {
                     // This is a quorum failure - return appropriate error
-                    // Extract achieved count from error message or use individual_locks.len()
                     let achieved = individual_locks.len();
                     Err(LockError::QuorumNotReached {
-                        required: self.quorum,
+                        required: required_quorum,
                         achieved,
                     })
                 } else if error_msg.contains("timeout") || resp.wait_time >= request.acquire_timeout {
@@ -309,10 +325,11 @@ impl DistributedLock {
         self.acquire_guard(&req).await
     }
 
-    /// Quorum-based lock acquisition: success if at least `self.quorum` clients succeed.
+    /// Quorum-based lock acquisition: success if at least the required quorum succeeds.
     /// Collects all individual lock_ids from successful clients and creates an aggregate lock_id.
     /// Returns the LockResponse with aggregate lock_id and individual lock mappings.
     async fn acquire_lock_quorum(&self, request: &LockRequest) -> Result<(LockResponse, Vec<(LockId, Arc<dyn LockClient>)>)> {
+        let required_quorum = self.required_quorum(request.lock_type);
         let futs: Vec<_> = self
             .clients
             .iter()
@@ -321,6 +338,7 @@ impl DistributedLock {
             .collect();
 
         let results = futures::future::join_all(futs).await;
+
         // Store all individual lock_ids and their corresponding clients
         let mut individual_locks: Vec<(LockId, Arc<dyn LockClient>)> = Vec::new();
 
@@ -362,7 +380,7 @@ impl DistributedLock {
             }
         }
 
-        if individual_locks.len() >= self.quorum {
+        if individual_locks.len() >= required_quorum {
             // Generate a new aggregate lock_id for multiple client locks
             let aggregate_lock_id = generate_aggregate_lock_id(&request.resource);
 
@@ -393,17 +411,17 @@ impl DistributedLock {
         } else {
             // Rollback: release all locks that were successfully acquired
             let rollback_count = individual_locks.len();
-            for (individual_lock_id, client) in individual_locks {
-                if let Err(e) = client.release(&individual_lock_id).await {
+            for (individual_lock_id, client) in &individual_locks {
+                if let Err(e) = client.release(individual_lock_id).await {
                     tracing::warn!("Failed to rollback lock {} on client: {}", individual_lock_id, e);
                 }
             }
 
             let resp = LockResponse::failure(
-                format!("Failed to acquire quorum: {}/{} required", rollback_count, self.quorum),
+                format!("Failed to acquire quorum: {rollback_count}/{required_quorum} required"),
                 Duration::ZERO,
             );
-            Ok((resp, Vec::new()))
+            Ok((resp, individual_locks))
         }
     }
 }

crates/lock/src/namespace/mod.rs

@@ -151,8 +151,9 @@ impl NamespaceLock {
         Self::Distributed(DistributedLock::new(namespace, clients, quorum))
     }
 
-    /// Create namespace lock with clients and an explicit quorum size.
-    /// Quorum will be clamped into [1, clients.len()].
+    /// Create namespace lock with clients and an explicit write quorum size.
+    /// Shared/read locks still use the distributed read quorum derived from client count.
+    /// The write quorum will be clamped into [1, clients.len()].
     pub fn with_clients_and_quorum(namespace: String, clients: Vec<Arc<dyn LockClient>>, quorum: usize) -> Self {
         Self::Distributed(DistributedLock::new(namespace, clients, quorum))
     }

crates/lock/src/namespace/tests.rs

@@ -13,12 +13,54 @@
 // limitations under the License.
 
 use super::*;
-use crate::GlobalLockManager;
 use crate::client::{ClientFactory, local::LocalClient};
 use crate::types::LockType;
+use crate::{GlobalLockManager, LockError, LockInfo, LockResponse, LockStats};
 use std::sync::Arc;
 use std::time::Duration;
 
+#[derive(Debug, Default)]
+struct FailingClient;
+
+#[async_trait::async_trait]
+impl crate::client::LockClient for FailingClient {
+    async fn acquire_lock(&self, _request: &LockRequest) -> crate::Result<LockResponse> {
+        Err(LockError::internal("simulated offline client"))
+    }
+
+    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 {
+        false
+    }
+
+    async fn is_local(&self) -> bool {
+        false
+    }
+}
+
 fn create_test_object_key(bucket: &str, object: &str) -> ObjectKey {
     ObjectKey {
         bucket: Arc::from(bucket),
@@ -368,3 +410,80 @@ async fn test_namespace_lock_distributed_with_clients_and_quorum() {
 
     drop(guard_b);
 }
+
+#[tokio::test]
+async fn test_namespace_lock_distributed_read_lock_succeeds_with_two_nodes_one_offline() {
+    let manager = Arc::new(GlobalLockManager::new());
+    let client_ok: Arc<dyn LockClient> = Arc::new(LocalClient::with_manager(manager));
+    let client_offline: Arc<dyn LockClient> = Arc::new(FailingClient);
+
+    let lock = NamespaceLock::with_clients_and_quorum("two-node".to_string(), vec![client_ok, client_offline], 2);
+    let resource = create_test_object_key("bucket", "object");
+
+    let guard = lock
+        .get_read_lock(resource, "owner-a", Duration::from_millis(100))
+        .await
+        .expect("read lock should succeed with one healthy node in a two-node cluster");
+
+    match guard {
+        NamespaceLockGuard::Standard(_) => {}
+        NamespaceLockGuard::Fast(_) => panic!("Expected Standard guard for distributed lock"),
+    }
+}
+
+#[tokio::test]
+async fn test_namespace_lock_distributed_write_lock_fails_with_two_nodes_one_offline() {
+    let manager = Arc::new(GlobalLockManager::new());
+    let client_ok: Arc<dyn LockClient> = Arc::new(LocalClient::with_manager(manager));
+    let client_offline: Arc<dyn LockClient> = Arc::new(FailingClient);
+
+    let lock = NamespaceLock::with_clients_and_quorum("two-node".to_string(), vec![client_ok, client_offline], 2);
+    let resource = create_test_object_key("bucket", "object");
+
+    let err = lock
+        .get_write_lock(resource, "owner-a", Duration::from_millis(100))
+        .await
+        .expect_err("write lock should fail with one healthy node in a two-node cluster");
+
+    let err_str = err.to_string().to_lowercase();
+    assert!(
+        err_str.contains("quorum") || err_str.contains("not reached"),
+        "expected quorum error, got: {err}"
+    );
+}
+
+#[tokio::test]
+async fn test_namespace_lock_distributed_even_node_read_write_quorum_split() {
+    let manager1 = Arc::new(GlobalLockManager::new());
+    let manager2 = Arc::new(GlobalLockManager::new());
+
+    let client1: Arc<dyn LockClient> = Arc::new(LocalClient::with_manager(manager1));
+    let client2: Arc<dyn LockClient> = Arc::new(LocalClient::with_manager(manager2));
+    let client3: Arc<dyn LockClient> = Arc::new(FailingClient);
+    let client4: Arc<dyn LockClient> = Arc::new(FailingClient);
+
+    let lock = NamespaceLock::with_clients("four-node".to_string(), vec![client1, client2, client3, client4]);
+    let resource = create_test_object_key("bucket", "object");
+
+    let mut read_guard = lock
+        .get_read_lock(resource.clone(), "owner-a", Duration::from_millis(100))
+        .await
+        .expect("read lock should succeed with two healthy nodes in a four-node cluster");
+
+    match &read_guard {
+        NamespaceLockGuard::Standard(_) => {}
+        NamespaceLockGuard::Fast(_) => panic!("Expected Standard guard for distributed lock"),
+    }
+    assert!(read_guard.release(), "read guard should release cleanly");
+
+    let err = lock
+        .get_write_lock(resource, "owner-a", Duration::from_millis(100))
+        .await
+        .expect_err("write lock should fail because four-node cluster requires quorum of 3");
+
+    let err_str = err.to_string().to_lowercase();
+    assert!(
+        err_str.contains("quorum") || err_str.contains("not reached"),
+        "expected quorum error, got: {err}"
+    );
+}