test(ecstore): add targeted refresh-loss harness (#6924)

Co-authored-by: heihutu <heihutu@gmail.com>

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<rustfs_lock::GlobalLockManager>) -> 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::LockResponse> {
+        rustfs_lock::LockClient::acquire_lock(&self.inner, request).await
+    }
+
+    async fn release(&self, lock_id: &rustfs_lock::LockId) -> rustfs_lock::Result<bool> {
+        rustfs_lock::LockClient::release(&self.inner, lock_id).await
+    }
+
+    async fn refresh(&self, lock_id: &rustfs_lock::LockId) -> rustfs_lock::Result<bool> {
+        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<bool> {
+        rustfs_lock::LockClient::force_release(&self.inner, lock_id).await
+    }
+
+    async fn check_status(&self, lock_id: &rustfs_lock::LockId) -> rustfs_lock::Result<Option<rustfs_lock::LockInfo>> {
+        rustfs_lock::LockClient::check_status(&self.inner, lock_id).await
+    }
+
+    async fn list_lock_leases(&self) -> Vec<rustfs_lock::LockLeaseInfo> {
+        rustfs_lock::LockClient::list_lock_leases(&self.inner).await
+    }
+
+    async fn get_stats(&self) -> rustfs_lock::Result<rustfs_lock::LockStats> {
+        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<tempfile::TempDir>, Arc<SetDisks>) {

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::LockResponse> {
-            rustfs_lock::LockClient::acquire_lock(&self.inner, request).await
-        }
-
-        async fn release(&self, lock_id: &rustfs_lock::LockId) -> rustfs_lock::Result<bool> {
-            rustfs_lock::LockClient::release(&self.inner, lock_id).await
-        }
-
-        async fn refresh(&self, lock_id: &rustfs_lock::LockId) -> rustfs_lock::Result<bool> {
-            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<bool> {
-            rustfs_lock::LockClient::force_release(&self.inner, lock_id).await
-        }
-
-        async fn check_status(&self, lock_id: &rustfs_lock::LockId) -> rustfs_lock::Result<Option<rustfs_lock::LockInfo>> {
-            rustfs_lock::LockClient::check_status(&self.inner, lock_id).await
-        }
-
-        async fn get_stats(&self) -> rustfs_lock::Result<rustfs_lock::LockStats> {
-            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<rustfs_lock::distributed_lock::LockLostSignal>,
-        Arc<RefreshFailureLockClient>,
+        Arc<RefreshLossLockClient>,
     ) {
         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<dyn rustfs_lock::LockClient>],
@@ -4087,7 +4036,7 @@ mod tests {
     ) -> (
         Arc<SelectObjectSnapshotLease>,
         Arc<rustfs_lock::distributed_lock::LockLostSignal>,
-        Arc<RefreshFailureLockClient>,
+        Arc<RefreshLossLockClient>,
     ) {
         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());
     }