test(ci): stabilize lifecycle timeout coverage (#5404)

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

crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs

@@ -11112,6 +11112,7 @@ mod tests {
     #[tokio::test(flavor = "multi_thread")]
     #[serial]
     async fn concurrent_resend_same_part_commits_one_generation() {
+        use crate::set_disk::{MultipartCommitBarrier, MultipartCommitPause};
         use crate::storage_api_contracts::object::ObjectIO as _;
 
         let (_paths, ecstore) = setup_test_env().await;
@@ -11133,51 +11134,36 @@ mod tests {
             })
             .collect();
 
-        // Two independent causes can produce a spurious lock-acquire timeout
-        // here, and both must stay covered:
-        // 1. A lost/stolen fast-lock wakeup could strand a waiter until the
-        //    deadline — fixed for real in fast_lock::shard by bounding each
-        //    notification wait (NOTIFY_WAIT_CAP re-polling).
-        // 2. Under the full nextest suite on loaded CI disks, the
-        //    *legitimately serialized* cross-disk commits can exceed the
-        //    acquire deadline all by themselves — observed on CI at the 5s
-        //    default and the 30s production default with six resends, and
-        //    again at 60s, which is a hard ceiling: fast_lock clamps every
-        //    requested timeout to MAX_ACQUIRE_TIMEOUT (60s), so raising the
-        //    env override higher is a no-op (the Timeout error still reports
-        //    the requested value). Keep the guard about the correctness
-        //    property, not disk latency: request the full 60s ceiling and cap
-        //    the queue depth at three resends, so the last waiter sits behind
-        //    at most two serialized commits (~12s each on the slowest observed
-        //    CI runner, comfortably inside the deadline). Three concurrent
-        //    resends still race the streaming phase and contend on the commit
-        //    lock, which is all the generation-mixing regression needs.
-        //    `#[serial]` keeps the process-wide env override isolated.
-        let results = temp_env::async_with_vars([(rustfs_config::ENV_OBJECT_LOCK_ACQUIRE_TIMEOUT, Some("60"))], async {
-            let mut tasks = tokio::task::JoinSet::new();
-            for payload in candidates.iter().cloned() {
-                let store = ecstore.clone();
-                let bucket = bucket.clone();
-                let upload_id = upload.upload_id.clone();
-                tasks.spawn(async move {
-                    let mut data = PutObjReader::from_vec(payload.clone());
-                    store
-                        .put_object_part(&bucket, object, &upload_id, 1, &mut data, &ObjectOptions::default())
-                        .await
-                        .map(|info| (info, payload))
-                });
-            }
+        let commit_barrier = MultipartCommitBarrier::install(&bucket, object, MultipartCommitPause::PutPartBeforeLockLost);
+        let start = Arc::new(tokio::sync::Barrier::new(candidates.len() + 1));
+        let mut tasks = tokio::task::JoinSet::new();
+        for payload in candidates.iter().cloned() {
+            let store = ecstore.clone();
+            let bucket = bucket.clone();
+            let upload_id = upload.upload_id.clone();
+            let start = Arc::clone(&start);
+            tasks.spawn(async move {
+                start.wait().await;
+                let mut data = PutObjReader::from_vec(payload.clone());
+                store
+                    .put_object_part(&bucket, object, &upload_id, 1, &mut data, &ObjectOptions::default())
+                    .await
+                    .map(|info| (info, payload))
+            });
+        }
+        start.wait().await;
 
-            // Every concurrent resend must succeed; the commit lock must never
-            // starve a waiter into a timeout.
-            let mut results = Vec::new();
-            while let Some(joined) = tasks.join_next().await {
-                let outcome = joined.expect("put_object_part task should not panic");
-                results.push(outcome.expect("every concurrent same-part resend must succeed without lock timeout"));
-            }
-            results
-        })
-        .await;
+        // The first writer holds the uploadId commit lock while the other
+        // resends reach the same critical section. Releasing it proves the
+        // handoff without depending on saturated CI disk latency.
+        commit_barrier.wait_until_paused().await;
+        commit_barrier.release();
+
+        let mut results = Vec::new();
+        while let Some(joined) = tasks.join_next().await {
+            let outcome = joined.expect("put_object_part task should not panic");
+            results.push(outcome.expect("every concurrent same-part resend must succeed without lock timeout"));
+        }
         assert_eq!(results.len(), candidates.len());
 
         // Exactly one generation is visible after the serialized commits, and its

crates/ecstore/src/set_disk/mod.rs

@@ -687,6 +687,8 @@ mod core;
 mod ctx;
 mod metadata;
 mod ops;
+#[cfg(test)]
+pub(crate) use ops::multipart::{MultipartCommitBarrier, MultipartCommitPause};
 #[cfg(feature = "test-util")]
 pub(crate) use ops::object::TransitionCleanupStoreBarrier as SetDiskTransitionCleanupStoreBarrier;
 pub(crate) use ops::object::body_cache_plaintext_len;

crates/ecstore/src/set_disk/ops/multipart.rs

@@ -26,6 +26,8 @@ use crate::crash_inject::{self, CrashPoint};
 use crate::multipart_listing::paginate_multipart_listing;
 use futures::{StreamExt, stream};
 use std::future::Future;
+#[cfg(test)]
+use std::sync::atomic::{AtomicBool, Ordering};
 use std::time::Duration;
 use tokio::task::JoinSet;
 
@@ -33,7 +35,7 @@ const MULTIPART_LIST_IO_CONCURRENCY: usize = 16;
 
 #[cfg(test)]
 #[derive(Clone, Copy, PartialEq, Eq)]
-enum MultipartCommitPause {
+pub(crate) enum MultipartCommitPause {
     PutPartBeforeLockLost,
     PutPartAfterRename,
     BeforeLockLost,
@@ -45,12 +47,13 @@ struct MultipartCommitBarrierState {
     bucket: String,
     object: String,
     pause: MultipartCommitPause,
+    armed: AtomicBool,
     arrived: tokio::sync::Notify,
     release: tokio::sync::Notify,
 }
 
 #[cfg(test)]
-struct MultipartCommitBarrier {
+pub(crate) struct MultipartCommitBarrier {
     state: Arc<MultipartCommitBarrierState>,
 }
 
@@ -60,11 +63,12 @@ static MULTIPART_COMMIT_BARRIER: std::sync::OnceLock<std::sync::Mutex<Option<Arc
 
 #[cfg(test)]
 impl MultipartCommitBarrier {
-    fn install(bucket: &str, object: &str, pause: MultipartCommitPause) -> Self {
+    pub(crate) fn install(bucket: &str, object: &str, pause: MultipartCommitPause) -> Self {
         let state = Arc::new(MultipartCommitBarrierState {
             bucket: bucket.to_string(),
             object: object.to_string(),
             pause,
+            armed: AtomicBool::new(true),
             arrived: tokio::sync::Notify::new(),
             release: tokio::sync::Notify::new(),
         });
@@ -78,13 +82,13 @@ impl MultipartCommitBarrier {
         Self { state }
     }
 
-    async fn wait_until_paused(&self) {
+    pub(crate) async fn wait_until_paused(&self) {
         tokio::time::timeout(Duration::from_secs(30), self.state.arrived.notified())
             .await
             .expect("multipart completion should reach the deterministic commit barrier");
     }
 
-    fn release(&self) {
+    pub(crate) fn release(&self) {
         self.state.release.notify_one();
     }
 }
@@ -112,7 +116,9 @@ async fn pause_multipart_commit(bucket: &str, object: &str, pause: MultipartComm
         .as_ref()
         .filter(|barrier| barrier.bucket == bucket && barrier.object == object && barrier.pause == pause)
         .cloned();
-    if let Some(barrier) = barrier {
+    if let Some(barrier) = barrier
+        && barrier.armed.swap(false, Ordering::AcqRel)
+    {
         barrier.arrived.notify_one();
         barrier.release.notified().await;
     }

crates/ecstore/src/store/object.rs

@@ -2601,10 +2601,10 @@ mod tests {
     // (backlog#1304): restore entry no longer serializes on the object lock.
     // The replacement semantics — non-blocking reads during the copy-back and
     // fast rejection of a concurrent restore — are covered end-to-end by
-    // `restore_object_usecase_reports_ongoing_conflict_and_completion`
-    // (rustfs/src/app/lifecycle_transition_api_test.rs) and at the lock level
-    // by the accept-guard test below; restore-vs-reader data protection lives
-    // in the inner put_object/complete_multipart_upload commit locks.
+    // `restore_object_usecase_reports_ongoing_conflict`
+    // (rustfs/src/app/lifecycle_transition_api_test.rs), while the SetDisks
+    // transition matrix covers the final local commit. Restore-vs-reader data
+    // protection lives in the inner put_object/complete_multipart_upload locks.
     #[tokio::test]
     #[serial_test::serial]
     async fn restore_accept_guard_serializes_concurrent_accepts() {