diff --git a/crates/ecstore/src/ecstore_validation_blackbox.rs b/crates/ecstore/src/ecstore_validation_blackbox.rs index 4db7e5104..b89caafe3 100644 --- a/crates/ecstore/src/ecstore_validation_blackbox.rs +++ b/crates/ecstore/src/ecstore_validation_blackbox.rs @@ -16,10 +16,143 @@ use rustfs_filemeta::{MetacacheReader, MetacacheWriter}; use std::io::Cursor; use std::path::PathBuf; use std::sync::Arc; +use std::sync::atomic::{AtomicBool, Ordering}; use tokio::fs; use tokio::io::AsyncReadExt; use tokio::sync::RwLock; +/// Test-only lock client whose refresh path can be rejected independently of +/// every other lock operation. The observed event is awaitable so lock-loss +/// tests do not depend on sleeps or scheduler timing. +#[derive(Debug)] +pub(crate) struct RefreshLossLockClient { + inner: rustfs_lock::LocalClient, + reject_refresh: AtomicBool, + rejected_refresh: AtomicBool, + rejected_refresh_notify: tokio::sync::Notify, +} + +impl RefreshLossLockClient { + pub(crate) fn with_manager(manager: Arc) -> Self { + Self { + inner: rustfs_lock::LocalClient::with_manager(manager), + reject_refresh: AtomicBool::new(false), + rejected_refresh: AtomicBool::new(false), + rejected_refresh_notify: tokio::sync::Notify::new(), + } + } + + pub(crate) fn reject_refreshes(&self) { + self.reject_refresh.store(true, Ordering::Release); + } + + pub(crate) fn refreshes_rejected(&self) -> bool { + self.rejected_refresh.load(Ordering::Acquire) + } + + pub(crate) async fn wait_for_rejected_refresh( + &self, + timeout: std::time::Duration, + ) -> std::result::Result<(), tokio::time::error::Elapsed> { + tokio::time::timeout(timeout, async { + loop { + let notified = self.rejected_refresh_notify.notified(); + if self.refreshes_rejected() { + return; + } + notified.await; + } + }) + .await + } +} + +#[async_trait::async_trait] +impl rustfs_lock::LockClient for RefreshLossLockClient { + async fn acquire_lock(&self, request: &rustfs_lock::LockRequest) -> rustfs_lock::Result { + rustfs_lock::LockClient::acquire_lock(&self.inner, request).await + } + + async fn release(&self, lock_id: &rustfs_lock::LockId) -> rustfs_lock::Result { + rustfs_lock::LockClient::release(&self.inner, lock_id).await + } + + async fn refresh(&self, lock_id: &rustfs_lock::LockId) -> rustfs_lock::Result { + if self.reject_refresh.load(Ordering::Acquire) { + self.rejected_refresh.store(true, Ordering::Release); + self.rejected_refresh_notify.notify_waiters(); + return Ok(false); + } + rustfs_lock::LockClient::refresh(&self.inner, lock_id).await + } + + async fn force_release(&self, lock_id: &rustfs_lock::LockId) -> rustfs_lock::Result { + rustfs_lock::LockClient::force_release(&self.inner, lock_id).await + } + + async fn check_status(&self, lock_id: &rustfs_lock::LockId) -> rustfs_lock::Result> { + rustfs_lock::LockClient::check_status(&self.inner, lock_id).await + } + + async fn list_lock_leases(&self) -> Vec { + rustfs_lock::LockClient::list_lock_leases(&self.inner).await + } + + async fn get_stats(&self) -> rustfs_lock::Result { + rustfs_lock::LockClient::get_stats(&self.inner).await + } + + async fn close(&self) -> rustfs_lock::Result<()> { + rustfs_lock::LockClient::close(&self.inner).await + } + + async fn is_online(&self) -> bool { + rustfs_lock::LockClient::is_online(&self.inner).await + } + + async fn is_local(&self) -> bool { + rustfs_lock::LockClient::is_local(&self.inner).await + } +} + +#[tokio::test] +async fn refresh_loss_lock_client_keeps_rejection_observable_for_late_waiters() { + let manager = Arc::new(rustfs_lock::GlobalLockManager::Enabled(Arc::new( + rustfs_lock::FastObjectLockManager::new(), + ))); + let client = RefreshLossLockClient::with_manager(manager); + let resource = rustfs_lock::ObjectKey::new("bucket", "object"); + let response = rustfs_lock::LockClient::acquire_lock( + &client, + &rustfs_lock::LockRequest::new(resource, rustfs_lock::LockType::Shared, "refresh-loss-harness"), + ) + .await + .expect("acquire should reach the inner local client"); + let lock_id = response.lock_info.expect("the inner local client should acquire the lock").id; + assert_eq!( + rustfs_lock::LockClient::list_lock_leases(&client).await.len(), + 1, + "lease diagnostics must remain transparent through the refresh wrapper" + ); + + client.reject_refreshes(); + assert!( + !rustfs_lock::LockClient::refresh(&client, &lock_id) + .await + .expect("refresh should return a response") + ); + client + .wait_for_rejected_refresh(std::time::Duration::from_millis(50)) + .await + .expect("a waiter registered after rejection must still observe the event"); + assert!(client.refreshes_rejected()); + assert!( + rustfs_lock::LockClient::release(&client, &lock_id) + .await + .expect("release should reach the inner local client") + ); +} + /// Returns the backing [`tempfile::TempDir`]s alongside the set so callers keep /// them alive for the test's duration and the directories are removed on drop. pub(crate) async fn make_local_set_disks(drive_count: usize, parity_count: usize) -> (Vec, Arc) { diff --git a/crates/ecstore/src/store/object.rs b/crates/ecstore/src/store/object.rs index d75e7cb42..68edea841 100644 --- a/crates/ecstore/src/store/object.rs +++ b/crates/ecstore/src/store/object.rs @@ -3934,7 +3934,7 @@ mod tests { }; use crate::core::pools::{PoolDecommissionInfo, PoolStatus}; use crate::core::sets::make_local_two_set_sets_with_ctx; - use crate::ecstore_validation_blackbox::{make_local_set_disks, make_local_set_disks_with_ctx}; + use crate::ecstore_validation_blackbox::{RefreshLossLockClient, make_local_set_disks, make_local_set_disks_with_ctx}; use crate::layout::{ endpoints::{Endpoints, PoolEndpoints, SetupType}, format::FormatV3, @@ -3949,7 +3949,7 @@ mod tests { use bytes::Bytes; use std::io::Cursor; use std::sync::Arc; - use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; + use std::sync::atomic::{AtomicUsize, Ordering}; use tokio::io::AsyncReadExt; struct WaitForLockLossReader { @@ -3991,68 +3991,17 @@ mod tests { calls: AtomicUsize, } - #[derive(Debug)] - struct RefreshFailureLockClient { - inner: LocalClient, - fail_refresh: AtomicBool, - } - - #[async_trait::async_trait] - impl rustfs_lock::LockClient for RefreshFailureLockClient { - async fn acquire_lock(&self, request: &rustfs_lock::LockRequest) -> rustfs_lock::Result { - rustfs_lock::LockClient::acquire_lock(&self.inner, request).await - } - - async fn release(&self, lock_id: &rustfs_lock::LockId) -> rustfs_lock::Result { - rustfs_lock::LockClient::release(&self.inner, lock_id).await - } - - async fn refresh(&self, lock_id: &rustfs_lock::LockId) -> rustfs_lock::Result { - if self.fail_refresh.load(Ordering::Acquire) { - return Ok(false); - } - rustfs_lock::LockClient::refresh(&self.inner, lock_id).await - } - - async fn force_release(&self, lock_id: &rustfs_lock::LockId) -> rustfs_lock::Result { - rustfs_lock::LockClient::force_release(&self.inner, lock_id).await - } - - async fn check_status(&self, lock_id: &rustfs_lock::LockId) -> rustfs_lock::Result> { - rustfs_lock::LockClient::check_status(&self.inner, lock_id).await - } - - async fn get_stats(&self) -> rustfs_lock::Result { - rustfs_lock::LockClient::get_stats(&self.inner).await - } - - async fn close(&self) -> rustfs_lock::Result<()> { - rustfs_lock::LockClient::close(&self.inner).await - } - - async fn is_online(&self) -> bool { - rustfs_lock::LockClient::is_online(&self.inner).await - } - - async fn is_local(&self) -> bool { - rustfs_lock::LockClient::is_local(&self.inner).await - } - } - async fn refresh_failure_test_guard( owner: &'static str, ) -> ( ObjectLockDiagGuard, Arc, - Arc, + Arc, ) { let manager = Arc::new(rustfs_lock::GlobalLockManager::Enabled(Arc::new( rustfs_lock::FastObjectLockManager::new(), ))); - let client = Arc::new(RefreshFailureLockClient { - inner: LocalClient::with_manager(manager), - fail_refresh: AtomicBool::new(false), - }); + let client = Arc::new(RefreshLossLockClient::with_manager(manager)); let namespace_lock = rustfs_lock::NamespaceLock::with_clients_and_quorum( owner.to_string(), vec![Arc::clone(&client) as Arc], @@ -4087,7 +4036,7 @@ mod tests { ) -> ( Arc, Arc, - Arc, + Arc, ) { let (guard, signal, client) = refresh_failure_test_guard(owner).await; (Arc::new(SelectObjectSnapshotLease::new(vec![guard])), signal, client) @@ -4220,7 +4169,11 @@ mod tests { let release_signal = Arc::clone(&signal); let release_task = tokio::spawn(async move { poll_started_rx.await.expect("reader poll should start"); - release_client.fail_refresh.store(true, Ordering::Release); + release_client.reject_refreshes(); + release_client + .wait_for_rejected_refresh(Duration::from_secs(5)) + .await + .expect("refresh rejection should be observed"); tokio::time::timeout(Duration::from_secs(5), release_signal.notified()) .await .expect("heartbeat should observe the rejected refresh"); @@ -4258,7 +4211,11 @@ mod tests { #[tokio::test(flavor = "current_thread")] async fn select_snapshot_reader_checks_guards_at_eof_before_monitor_runs() { let (guard, signal, client) = refresh_failure_test_guard("select-snapshot-eof-fence").await; - client.fail_refresh.store(true, Ordering::Release); + client.reject_refreshes(); + client + .wait_for_rejected_refresh(Duration::from_secs(5)) + .await + .expect("refresh rejection should be observed"); tokio::time::timeout(Duration::from_secs(5), signal.notified()) .await .expect("heartbeat should observe the rejected refresh"); @@ -4323,7 +4280,11 @@ mod tests { second_started_rx .await .expect("second inner reader should reach Poll::Pending"); - second_client.fail_refresh.store(true, Ordering::Release); + second_client.reject_refreshes(); + second_client + .wait_for_rejected_refresh(Duration::from_secs(5)) + .await + .expect("refresh rejection should be observed"); let (first_result, second_result) = tokio::join!( tokio::time::timeout(Duration::from_secs(5), first_read_task), tokio::time::timeout(Duration::from_secs(5), second_read_task), @@ -4336,7 +4297,7 @@ mod tests { assert_eq!(error.kind(), std::io::ErrorKind::Other); } - assert!(!first_client.fail_refresh.load(Ordering::Acquire)); + assert!(!first_client.refreshes_rejected()); assert!(!first_signal.is_lost()); assert!(second_signal.is_lost()); }