From e86c6b726f7fe6901234298d90e56b8c7a0c7bf9 Mon Sep 17 00:00:00 2001 From: weisd Date: Tue, 31 Mar 2026 16:16:13 +0800 Subject: [PATCH] fix(lock): split distributed read and write quorum (#2355) --- crates/e2e_test/src/reliant/lock.rs | 152 +++++++- crates/ecstore/src/config/com.rs | 549 +++++++++++++++++++++++++++- crates/ecstore/src/set_disk.rs | 153 ++++++++ crates/lock/src/distributed_lock.rs | 36 +- crates/lock/src/namespace/mod.rs | 5 +- crates/lock/src/namespace/tests.rs | 121 +++++- 6 files changed, 997 insertions(+), 19 deletions(-) diff --git a/crates/e2e_test/src/reliant/lock.rs b/crates/e2e_test/src/reliant/lock.rs index 9d4b2c972..7cae2c390 100644 --- a/crates/e2e_test/src/reliant/lock.rs +++ b/crates/e2e_test/src/reliant/lock.rs @@ -14,10 +14,69 @@ // limitations under the License. use super::{grpc_lock_client::GrpcLockClient, grpc_lock_server::spawn_lock_server}; -use rustfs_lock::{GlobalLockManager, NamespaceLock, ObjectKey, client::local::LocalClient}; +use rustfs_lock::client::local::LocalClient; +use rustfs_lock::{GlobalLockManager, LockError, LockInfo, LockResponse, LockStats, NamespaceLock, ObjectKey}; use std::sync::Arc; use std::time::Duration; +fn test_resource() -> ObjectKey { + ObjectKey { + bucket: Arc::from("test-bucket"), + object: Arc::from("test-object"), + version: None, + } +} + +#[derive(Debug, Default)] +struct FailingClient; + +#[async_trait::async_trait] +impl rustfs_lock::LockClient for FailingClient { + async fn acquire_lock(&self, _request: &rustfs_lock::LockRequest) -> rustfs_lock::Result { + Err(LockError::internal("simulated gRPC node failure")) + } + + async fn release(&self, _lock_id: &rustfs_lock::LockId) -> rustfs_lock::Result { + Ok(false) + } + + async fn refresh(&self, _lock_id: &rustfs_lock::LockId) -> rustfs_lock::Result { + Ok(false) + } + + async fn force_release(&self, _lock_id: &rustfs_lock::LockId) -> rustfs_lock::Result { + Ok(false) + } + + async fn check_status(&self, _lock_id: &rustfs_lock::LockId) -> rustfs_lock::Result> { + Ok(None) + } + + async fn get_stats(&self) -> rustfs_lock::Result { + Ok(LockStats::default()) + } + + async fn close(&self) -> rustfs_lock::Result<()> { + Ok(()) + } + + async fn is_online(&self) -> bool { + false + } + + async fn is_local(&self) -> bool { + false + } +} + +async fn failing_grpc_client() -> (Arc, tokio::task::JoinHandle<()>) { + let failing_client: Arc = Arc::new(FailingClient); + let (addr, handle) = spawn_lock_server(failing_client) + .await + .expect("Failed to spawn failing gRPC lock server"); + (Arc::new(GrpcLockClient::new(addr)), handle) +} + #[tokio::test] async fn test_distributed_lock_4_nodes_grpc() { // Spawn 4 gRPC lock servers, each with its own GlobalLockManager @@ -52,11 +111,7 @@ async fn test_distributed_lock_4_nodes_grpc() { let lock = NamespaceLock::with_clients_and_quorum("grpc-4-node".to_string(), clients, 3); assert_eq!(lock.namespace(), "grpc-4-node"); - let resource = ObjectKey { - bucket: Arc::from("test-bucket"), - object: Arc::from("test-object"), - version: None, - }; + let resource = test_resource(); // Test 1: Owner A acquires write lock successfully let mut guard_a = lock @@ -128,3 +183,88 @@ async fn test_distributed_lock_4_nodes_grpc() { handle3.abort(); handle4.abort(); } + +#[tokio::test] +async fn test_distributed_lock_2_nodes_grpc_read_survives_failed_node() { + let manager = Arc::new(GlobalLockManager::new()); + let local_client: Arc = Arc::new(LocalClient::with_manager(manager)); + + let (addr, handle) = spawn_lock_server(local_client).await.expect("Failed to spawn server"); + tokio::time::sleep(Duration::from_millis(100)).await; + + let grpc_client_ok: Arc = Arc::new(GrpcLockClient::new(addr)); + let (grpc_client_bad, failing_handle) = failing_grpc_client().await; + let lock = NamespaceLock::with_clients_and_quorum("grpc-2-node".to_string(), vec![grpc_client_ok, grpc_client_bad], 2); + let resource = test_resource(); + + let guard = lock + .get_read_lock(resource.clone(), "owner-a", Duration::from_secs(2)) + .await + .expect("Read lock should succeed with one healthy node in a two-node gRPC cluster"); + + match guard { + rustfs_lock::NamespaceLockGuard::Standard(_) => {} + rustfs_lock::NamespaceLockGuard::Fast(_) => panic!("Expected Standard guard for distributed lock"), + } + + let err = lock + .get_write_lock(resource, "owner-a", Duration::from_secs(2)) + .await + .expect_err("Write lock should fail with one healthy node in a two-node gRPC cluster"); + + let err_str = err.to_string().to_lowercase(); + assert!( + err_str.contains("quorum") || err_str.contains("not reached"), + "Error should be quorum related, got: {}", + err + ); + + handle.abort(); + failing_handle.abort(); +} + +#[tokio::test] +async fn test_distributed_lock_4_nodes_grpc_read_write_quorum_split_with_two_failed_nodes() { + let manager1 = Arc::new(GlobalLockManager::new()); + let manager2 = Arc::new(GlobalLockManager::new()); + let client1: Arc = Arc::new(LocalClient::with_manager(manager1)); + let client2: Arc = Arc::new(LocalClient::with_manager(manager2)); + + let (addr1, handle1) = spawn_lock_server(client1).await.expect("Failed to spawn server 1"); + let (addr2, handle2) = spawn_lock_server(client2).await.expect("Failed to spawn server 2"); + tokio::time::sleep(Duration::from_millis(100)).await; + + let grpc_client1: Arc = Arc::new(GrpcLockClient::new(addr1)); + let grpc_client2: Arc = Arc::new(GrpcLockClient::new(addr2)); + let (grpc_client3, handle3) = failing_grpc_client().await; + let (grpc_client4, handle4) = failing_grpc_client().await; + + let lock = NamespaceLock::with_clients( + "grpc-4-node-partial".to_string(), + vec![grpc_client1, grpc_client2, grpc_client3, grpc_client4], + ); + let resource = test_resource(); + + let mut read_guard = lock + .get_read_lock(resource.clone(), "owner-a", Duration::from_secs(2)) + .await + .expect("Read lock should succeed with two healthy nodes in a four-node gRPC cluster"); + assert!(read_guard.release(), "Read guard should release cleanly"); + + let err = lock + .get_write_lock(resource, "owner-b", Duration::from_secs(2)) + .await + .expect_err("Write lock should fail when only two of four gRPC nodes are healthy"); + + let err_str = err.to_string().to_lowercase(); + assert!( + err_str.contains("quorum") || err_str.contains("not reached"), + "Error should be quorum related, got: {}", + err + ); + + handle1.abort(); + handle2.abort(); + handle3.abort(); + handle4.abort(); +} diff --git a/crates/ecstore/src/config/com.rs b/crates/ecstore/src/config/com.rs index 488b3e741..5885969b0 100644 --- a/crates/ecstore/src/config/com.rs +++ b/crates/ecstore/src/config/com.rs @@ -710,12 +710,540 @@ async fn apply_dynamic_config_for_sub_sys(cfg: &mut Config, api: mod tests { use super::{ configs_semantically_equal, decode_server_config_blob, encode_server_config_blob, is_standard_object_server_config, - storage_class_kvs_mut, + read_config_with_metadata, storage_class_kvs_mut, }; use crate::config::{Config, oidc}; + use crate::disk::endpoint::Endpoint; + use crate::endpoints::SetupType; + use crate::error::{Error, Result}; + use crate::global::{is_dist_erasure, is_erasure, is_erasure_sd, update_erasure_type}; + use crate::set_disk::SetDisks; + use crate::store_api::{ + BucketInfo, BucketOperations, BucketOptions, CompletePart, DeleteBucketOptions, DeletedObject, GetObjectReader, + HTTPRangeSpec, HealOperations, ListMultipartsInfo, ListObjectVersionsInfo, ListObjectsV2Info, ListOperations, + MakeBucketOptions, MultipartInfo, MultipartOperations, MultipartUploadResult, ObjectIO, ObjectInfo, ObjectOperations, + ObjectOptions, ObjectToDelete, PartInfo, PutObjReader, StorageAPI, WalkOptions, + }; + use http::HeaderMap; use rustfs_config::oidc::IDENTITY_OPENID_SUB_SYS; use rustfs_config::{DEFAULT_DELIMITER, ENABLE_KEY, EnableState}; + use rustfs_filemeta::FileInfo; + use rustfs_lock::client::LockClient; + use rustfs_lock::client::local::LocalClient; + use rustfs_lock::{LockError, LockInfo, LockResponse, LockStats}; use serde_json::Value; + use serial_test::serial; + use std::collections::HashMap; + use std::fmt::{Debug, Formatter}; + use std::io::Cursor; + use std::pin::Pin; + use std::sync::Arc; + use std::task::{Context, Poll}; + use time::OffsetDateTime; + use tokio::io::{AsyncRead, ReadBuf}; + use tokio::sync::RwLock; + use tokio_util::sync::CancellationToken; + + #[derive(Debug, Default)] + struct FailingClient; + + #[async_trait::async_trait] + impl LockClient for FailingClient { + async fn acquire_lock(&self, _request: &rustfs_lock::LockRequest) -> rustfs_lock::Result { + Err(LockError::internal("simulated offline client")) + } + + async fn release(&self, _lock_id: &rustfs_lock::LockId) -> rustfs_lock::Result { + Ok(false) + } + + async fn refresh(&self, _lock_id: &rustfs_lock::LockId) -> rustfs_lock::Result { + Ok(false) + } + + async fn force_release(&self, _lock_id: &rustfs_lock::LockId) -> rustfs_lock::Result { + Ok(false) + } + + async fn check_status(&self, _lock_id: &rustfs_lock::LockId) -> rustfs_lock::Result> { + Ok(None) + } + + async fn get_stats(&self) -> rustfs_lock::Result { + Ok(LockStats::default()) + } + + async fn close(&self) -> rustfs_lock::Result<()> { + Ok(()) + } + + async fn is_online(&self) -> bool { + false + } + + async fn is_local(&self) -> bool { + false + } + } + + struct GuardedCursor { + inner: Cursor>, + _guard: Option, + } + + impl AsyncRead for GuardedCursor { + fn poll_read(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll> { + Pin::new(&mut self.inner).poll_read(cx, buf) + } + } + + struct LockingConfigStorage { + set_disks: Arc, + data: Vec, + } + + impl Debug for LockingConfigStorage { + fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { + f.debug_struct("LockingConfigStorage").finish() + } + } + + struct SetupTypeGuard { + previous: SetupType, + } + + impl SetupTypeGuard { + async fn switch_to(next: SetupType) -> Self { + let previous = current_setup_type().await; + update_erasure_type(next).await; + Self { previous } + } + } + + impl Drop for SetupTypeGuard { + fn drop(&mut self) { + let previous = self.previous.clone(); + let handle = tokio::runtime::Handle::current(); + tokio::task::block_in_place(|| { + handle.block_on(async move { + update_erasure_type(previous).await; + }); + }); + } + } + + async fn current_setup_type() -> SetupType { + if is_dist_erasure().await { + SetupType::DistErasure + } else if is_erasure_sd().await { + SetupType::ErasureSD + } else if is_erasure().await { + SetupType::Erasure + } else { + SetupType::Unknown + } + } + + impl LockingConfigStorage { + async fn new(lockers: Vec>, data: Vec) -> Self { + let endpoints = vec![ + Endpoint::try_from("http://127.0.0.1:9000/data").expect("first endpoint should parse"), + Endpoint::try_from("http://127.0.0.1:9001/data").expect("second endpoint should parse"), + ]; + + let set_disks = SetDisks::new( + "config-test-owner".to_string(), + Arc::new(RwLock::new(vec![None, None])), + 2, + 1, + 0, + 0, + endpoints, + crate::disk::format::FormatV3::new(1, 2), + lockers, + ) + .await; + + Self { set_disks, data } + } + + fn object_info(&self, bucket: &str, object: &str) -> ObjectInfo { + ObjectInfo { + bucket: bucket.to_string(), + name: object.to_string(), + storage_class: None, + mod_time: Some(OffsetDateTime::now_utc()), + size: self.data.len() as i64, + actual_size: self.data.len() as i64, + is_dir: false, + user_defined: HashMap::new(), + parity_blocks: 0, + data_blocks: 0, + version_id: None, + delete_marker: false, + transitioned_object: Default::default(), + restore_ongoing: false, + restore_expires: None, + user_tags: String::new(), + parts: Vec::new(), + is_latest: true, + content_type: Some("application/json".to_string()), + content_encoding: None, + expires: None, + num_versions: 1, + successor_mod_time: None, + put_object_reader: None, + etag: None, + inlined: false, + metadata_only: false, + version_only: false, + replication_status_internal: None, + replication_status: Default::default(), + version_purge_status_internal: None, + version_purge_status: Default::default(), + replication_decision: String::new(), + checksum: None, + } + } + } + + #[async_trait::async_trait] + impl ObjectIO for LockingConfigStorage { + async fn get_object_reader( + &self, + bucket: &str, + object: &str, + _range: Option, + _h: HeaderMap, + opts: &ObjectOptions, + ) -> Result { + let guard = if opts.no_lock { + None + } else { + Some( + self.set_disks + .new_ns_lock(bucket, object) + .await? + .get_read_lock(std::time::Duration::from_millis(100)) + .await + .map_err(|err| Error::other(format!("lock failed: {err}")))?, + ) + }; + + Ok(GetObjectReader { + stream: Box::new(GuardedCursor { + inner: Cursor::new(self.data.clone()), + _guard: guard, + }), + object_info: self.object_info(bucket, object), + }) + } + + async fn put_object( + &self, + _bucket: &str, + _object: &str, + _data: &mut PutObjReader, + _opts: &ObjectOptions, + ) -> Result { + panic!("unused in test") + } + } + + #[async_trait::async_trait] + impl BucketOperations for LockingConfigStorage { + async fn make_bucket(&self, _bucket: &str, _opts: &MakeBucketOptions) -> Result<()> { + panic!("unused in test") + } + + async fn get_bucket_info(&self, _bucket: &str, _opts: &BucketOptions) -> Result { + panic!("unused in test") + } + + async fn list_bucket(&self, _opts: &BucketOptions) -> Result> { + panic!("unused in test") + } + + async fn delete_bucket(&self, _bucket: &str, _opts: &DeleteBucketOptions) -> Result<()> { + panic!("unused in test") + } + } + + #[async_trait::async_trait] + impl ObjectOperations for LockingConfigStorage { + async fn get_object_info(&self, _bucket: &str, _object: &str, _opts: &ObjectOptions) -> Result { + panic!("unused in test") + } + + async fn verify_object_integrity(&self, _bucket: &str, _object: &str, _opts: &ObjectOptions) -> Result<()> { + panic!("unused in test") + } + + async fn copy_object( + &self, + _src_bucket: &str, + _src_object: &str, + _dst_bucket: &str, + _dst_object: &str, + _src_info: &mut ObjectInfo, + _src_opts: &ObjectOptions, + _dst_opts: &ObjectOptions, + ) -> Result { + panic!("unused in test") + } + + async fn delete_object_version( + &self, + _bucket: &str, + _object: &str, + _fi: &FileInfo, + _force_del_marker: bool, + ) -> Result<()> { + panic!("unused in test") + } + + async fn delete_object(&self, _bucket: &str, _object: &str, _opts: ObjectOptions) -> Result { + panic!("unused in test") + } + + async fn delete_objects( + &self, + _bucket: &str, + _objects: Vec, + _opts: ObjectOptions, + ) -> (Vec, Vec>) { + panic!("unused in test") + } + + async fn put_object_metadata(&self, _bucket: &str, _object: &str, _opts: &ObjectOptions) -> Result { + panic!("unused in test") + } + + async fn get_object_tags(&self, _bucket: &str, _object: &str, _opts: &ObjectOptions) -> Result { + panic!("unused in test") + } + + async fn put_object_tags(&self, _bucket: &str, _object: &str, _tags: &str, _opts: &ObjectOptions) -> Result { + panic!("unused in test") + } + + async fn delete_object_tags(&self, _bucket: &str, _object: &str, _opts: &ObjectOptions) -> Result { + panic!("unused in test") + } + + async fn add_partial(&self, _bucket: &str, _object: &str, _version_id: &str) -> Result<()> { + panic!("unused in test") + } + + async fn transition_object(&self, _bucket: &str, _object: &str, _opts: &ObjectOptions) -> Result<()> { + panic!("unused in test") + } + + async fn restore_transitioned_object(self: Arc, _bucket: &str, _object: &str, _opts: &ObjectOptions) -> Result<()> { + panic!("unused in test") + } + } + + #[async_trait::async_trait] + impl ListOperations for LockingConfigStorage { + async fn list_objects_v2( + self: Arc, + _bucket: &str, + _prefix: &str, + _continuation_token: Option, + _delimiter: Option, + _max_keys: i32, + _fetch_owner: bool, + _start_after: Option, + _incl_deleted: bool, + ) -> Result { + panic!("unused in test") + } + + async fn list_object_versions( + self: Arc, + _bucket: &str, + _prefix: &str, + _marker: Option, + _version_marker: Option, + _delimiter: Option, + _max_keys: i32, + ) -> Result { + panic!("unused in test") + } + + async fn walk( + self: Arc, + _rx: CancellationToken, + _bucket: &str, + _prefix: &str, + _result: tokio::sync::mpsc::Sender, + _opts: WalkOptions, + ) -> Result<()> { + panic!("unused in test") + } + } + + #[async_trait::async_trait] + impl MultipartOperations for LockingConfigStorage { + async fn list_multipart_uploads( + &self, + _bucket: &str, + _prefix: &str, + _key_marker: Option, + _upload_id_marker: Option, + _delimiter: Option, + _max_uploads: usize, + ) -> Result { + panic!("unused in test") + } + + async fn new_multipart_upload( + &self, + _bucket: &str, + _object: &str, + _opts: &ObjectOptions, + ) -> Result { + panic!("unused in test") + } + + async fn copy_object_part( + &self, + _src_bucket: &str, + _src_object: &str, + _dst_bucket: &str, + _dst_object: &str, + _upload_id: &str, + _part_id: usize, + _start_offset: i64, + _length: i64, + _src_info: &ObjectInfo, + _src_opts: &ObjectOptions, + _dst_opts: &ObjectOptions, + ) -> Result<()> { + panic!("unused in test") + } + + async fn put_object_part( + &self, + _bucket: &str, + _object: &str, + _upload_id: &str, + _part_id: usize, + _data: &mut PutObjReader, + _opts: &ObjectOptions, + ) -> Result { + panic!("unused in test") + } + + async fn get_multipart_info( + &self, + _bucket: &str, + _object: &str, + _upload_id: &str, + _opts: &ObjectOptions, + ) -> Result { + panic!("unused in test") + } + + async fn list_object_parts( + &self, + _bucket: &str, + _object: &str, + _upload_id: &str, + _part_number_marker: Option, + _max_parts: usize, + _opts: &ObjectOptions, + ) -> Result { + panic!("unused in test") + } + + async fn abort_multipart_upload( + &self, + _bucket: &str, + _object: &str, + _upload_id: &str, + _opts: &ObjectOptions, + ) -> Result<()> { + panic!("unused in test") + } + + async fn complete_multipart_upload( + self: Arc, + _bucket: &str, + _object: &str, + _upload_id: &str, + _uploaded_parts: Vec, + _opts: &ObjectOptions, + ) -> Result { + panic!("unused in test") + } + } + + #[async_trait::async_trait] + impl HealOperations for LockingConfigStorage { + async fn heal_format(&self, _dry_run: bool) -> Result<(rustfs_madmin::heal_commands::HealResultItem, Option)> { + panic!("unused in test") + } + + async fn heal_bucket( + &self, + _bucket: &str, + _opts: &rustfs_common::heal_channel::HealOpts, + ) -> Result { + panic!("unused in test") + } + + async fn heal_object( + &self, + _bucket: &str, + _object: &str, + _version_id: &str, + _opts: &rustfs_common::heal_channel::HealOpts, + ) -> Result<(rustfs_madmin::heal_commands::HealResultItem, Option)> { + panic!("unused in test") + } + + async fn get_pool_and_set(&self, _id: &str) -> Result<(Option, Option, Option)> { + panic!("unused in test") + } + + async fn check_abandoned_parts( + &self, + _bucket: &str, + _object: &str, + _opts: &rustfs_common::heal_channel::HealOpts, + ) -> Result<()> { + panic!("unused in test") + } + } + + #[async_trait::async_trait] + impl StorageAPI for LockingConfigStorage { + async fn new_ns_lock(&self, bucket: &str, object: &str) -> Result { + self.set_disks.new_ns_lock(bucket, object).await + } + + async fn backend_info(&self) -> rustfs_madmin::BackendInfo { + panic!("unused in test") + } + + async fn storage_info(&self) -> rustfs_madmin::StorageInfo { + panic!("unused in test") + } + + async fn local_storage_info(&self) -> rustfs_madmin::StorageInfo { + panic!("unused in test") + } + + async fn get_disks(&self, _pool_idx: usize, _set_idx: usize) -> Result>> { + panic!("unused in test") + } + + fn set_drive_counts(&self) -> Vec { + panic!("unused in test") + } + } #[test] fn test_decode_server_config_accepts_legacy_hidden_if_empty_alias() { @@ -912,4 +1440,23 @@ mod tests { let rhs = decode_server_config_blob(legacy).expect("decode legacy"); assert!(configs_semantically_equal(&lhs, &rhs)); } + + #[tokio::test(flavor = "multi_thread")] + #[serial] + async fn test_read_config_with_metadata_succeeds_with_one_healthy_locker_in_two_node_dist_setup() { + let _setup_type_guard = SetupTypeGuard::switch_to(SetupType::DistErasure).await; + + let manager = Arc::new(rustfs_lock::GlobalLockManager::new()); + let healthy_client: Arc = Arc::new(LocalClient::with_manager(manager)); + let failing_client: Arc = Arc::new(FailingClient); + let storage = Arc::new(LockingConfigStorage::new(vec![healthy_client, failing_client], br#"{"ok":true}"#.to_vec()).await); + + let (data, object_info) = read_config_with_metadata(storage, "config/test.json", &ObjectOptions::default()) + .await + .expect("config read should succeed with one healthy locker"); + + assert_eq!(data, br#"{"ok":true}"#.to_vec()); + assert_eq!(object_info.bucket, crate::disk::RUSTFS_META_BUCKET); + assert_eq!(object_info.name, "config/test.json"); + } } diff --git a/crates/ecstore/src/set_disk.rs b/crates/ecstore/src/set_disk.rs index 5dda520dc..5789d090d 100644 --- a/crates/ecstore/src/set_disk.rs +++ b/crates/ecstore/src/set_disk.rs @@ -4000,12 +4000,116 @@ mod tests { use super::*; use crate::disk::CHECK_PART_UNKNOWN; use crate::disk::CHECK_PART_VOLUME_NOT_FOUND; + use crate::disk::endpoint::Endpoint; use crate::disk::error::DiskError; + use crate::endpoints::SetupType; + use crate::global::{is_dist_erasure, is_erasure, is_erasure_sd, update_erasure_type}; use crate::store_api::{CompletePart, ObjectInfo}; use rustfs_filemeta::ErasureInfo; + use rustfs_lock::client::local::LocalClient; + use rustfs_lock::{LockError, LockInfo, LockResponse, LockStats}; + use serial_test::serial; use std::collections::HashMap; use time::OffsetDateTime; + #[derive(Debug, Default)] + struct FailingClient; + + #[async_trait::async_trait] + impl LockClient for FailingClient { + async fn acquire_lock(&self, _request: &rustfs_lock::LockRequest) -> rustfs_lock::Result { + Err(LockError::internal("simulated offline client")) + } + + async fn release(&self, _lock_id: &rustfs_lock::LockId) -> rustfs_lock::Result { + Ok(false) + } + + async fn refresh(&self, _lock_id: &rustfs_lock::LockId) -> rustfs_lock::Result { + Ok(false) + } + + async fn force_release(&self, _lock_id: &rustfs_lock::LockId) -> rustfs_lock::Result { + Ok(false) + } + + async fn check_status(&self, _lock_id: &rustfs_lock::LockId) -> rustfs_lock::Result> { + Ok(None) + } + + async fn get_stats(&self) -> rustfs_lock::Result { + Ok(LockStats::default()) + } + + async fn close(&self) -> rustfs_lock::Result<()> { + Ok(()) + } + + async fn is_online(&self) -> bool { + false + } + + async fn is_local(&self) -> bool { + false + } + } + + async fn make_test_set_disks(lockers: Vec>) -> Arc { + let endpoints = vec![ + Endpoint::try_from("http://127.0.0.1:9000/data").expect("first endpoint should parse"), + Endpoint::try_from("http://127.0.0.1:9001/data").expect("second endpoint should parse"), + ]; + + SetDisks::new( + "test-owner".to_string(), + Arc::new(RwLock::new(vec![None, None])), + 2, + 1, + 0, + 0, + endpoints, + FormatV3::new(1, 2), + lockers, + ) + .await + } + + struct SetupTypeGuard { + previous: SetupType, + } + + impl SetupTypeGuard { + async fn switch_to(next: SetupType) -> Self { + let previous = current_setup_type().await; + update_erasure_type(next).await; + Self { previous } + } + } + + impl Drop for SetupTypeGuard { + fn drop(&mut self) { + let previous = self.previous.clone(); + let handle = tokio::runtime::Handle::current(); + tokio::task::block_in_place(|| { + handle.block_on(async move { + update_erasure_type(previous).await; + }); + }); + } + } + + async fn current_setup_type() -> SetupType { + if is_dist_erasure().await { + SetupType::DistErasure + } else if is_erasure_sd().await { + SetupType::ErasureSD + } else if is_erasure().await { + SetupType::Erasure + } else { + SetupType::Unknown + } + } + #[test] fn disk_health_entry_returns_cached_value_within_ttl() { let entry = DiskHealthEntry { @@ -4128,6 +4232,55 @@ mod tests { assert_ne!(result3, result4); } + #[tokio::test(flavor = "multi_thread")] + #[serial] + async fn test_new_ns_lock_distributed_read_succeeds_with_two_lockers_one_offline() { + let _setup_type_guard = SetupTypeGuard::switch_to(SetupType::DistErasure).await; + + let manager = Arc::new(rustfs_lock::GlobalLockManager::new()); + let healthy_client: Arc = Arc::new(LocalClient::with_manager(manager)); + let failing_client: Arc = Arc::new(FailingClient); + let set_disks = make_test_set_disks(vec![healthy_client, failing_client]).await; + + let guard = set_disks + .new_ns_lock("bucket", "object") + .await + .expect("namespace lock should be created") + .get_read_lock(Duration::from_millis(100)) + .await + .expect("read lock should succeed with one healthy locker"); + + match guard { + NamespaceLockGuard::Standard(_) => {} + NamespaceLockGuard::Fast(_) => panic!("Expected distributed guard for dist-erasure"), + } + } + + #[tokio::test(flavor = "multi_thread")] + #[serial] + async fn test_new_ns_lock_distributed_write_fails_with_two_lockers_one_offline() { + let _setup_type_guard = SetupTypeGuard::switch_to(SetupType::DistErasure).await; + + let manager = Arc::new(rustfs_lock::GlobalLockManager::new()); + let healthy_client: Arc = Arc::new(LocalClient::with_manager(manager)); + let failing_client: Arc = Arc::new(FailingClient); + let set_disks = make_test_set_disks(vec![healthy_client, failing_client]).await; + + let err = set_disks + .new_ns_lock("bucket", "object") + .await + .expect("namespace lock should be created") + .get_write_lock(Duration::from_millis(100)) + .await + .expect_err("write lock should fail with one healthy locker"); + + let err_str = err.to_string().to_lowercase(); + assert!( + err_str.contains("quorum") || err_str.contains("not reached"), + "expected quorum error, got: {err}" + ); + } + #[test] fn test_common_parity() { // Test common parity calculation diff --git a/crates/lock/src/distributed_lock.rs b/crates/lock/src/distributed_lock.rs index 3f32218f6..f461fffff 100644 --- a/crates/lock/src/distributed_lock.rs +++ b/crates/lock/src/distributed_lock.rs @@ -174,7 +174,7 @@ pub struct DistributedLock { clients: Vec>, /// 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)>)> { + 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)> = 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)) } } } diff --git a/crates/lock/src/namespace/mod.rs b/crates/lock/src/namespace/mod.rs index bc1dfc3b1..690993aeb 100644 --- a/crates/lock/src/namespace/mod.rs +++ b/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>, quorum: usize) -> Self { Self::Distributed(DistributedLock::new(namespace, clients, quorum)) } diff --git a/crates/lock/src/namespace/tests.rs b/crates/lock/src/namespace/tests.rs index bad1b0b0a..f8b48c882 100644 --- a/crates/lock/src/namespace/tests.rs +++ b/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 { + Err(LockError::internal("simulated offline client")) + } + + 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 { + 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 = Arc::new(LocalClient::with_manager(manager)); + let client_offline: Arc = 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 = Arc::new(LocalClient::with_manager(manager)); + let client_offline: Arc = 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 = Arc::new(LocalClient::with_manager(manager1)); + let client2: Arc = Arc::new(LocalClient::with_manager(manager2)); + let client3: Arc = Arc::new(FailingClient); + let client4: Arc = 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}" + ); +}