From 5af56cbb02a09026f4a37b58d3d81dc65fb1c32e Mon Sep 17 00:00:00 2001 From: houseme Date: Wed, 29 Jul 2026 09:37:29 +0800 Subject: [PATCH] test(ci): stabilize lifecycle timeout coverage (#5404) Co-authored-by: heihutu --- .config/nextest.toml | 23 +++--- .../bucket/lifecycle/bucket_lifecycle_ops.rs | 74 +++++++----------- crates/ecstore/src/set_disk/mod.rs | 2 + crates/ecstore/src/set_disk/ops/multipart.rs | 18 +++-- crates/ecstore/src/store/object.rs | 8 +- .../src/app/lifecycle_transition_api_test.rs | 78 +------------------ 6 files changed, 61 insertions(+), 142 deletions(-) diff --git a/.config/nextest.toml b/.config/nextest.toml index 16418d2e6..0f1917bda 100644 --- a/.config/nextest.toml +++ b/.config/nextest.toml @@ -1,17 +1,14 @@ # nextest configuration for RustFS. # -# Serialize two known load-sensitive / global-state-sharing ecstore test groups -# so the full parallel nextest suite stops producing spurious failures -# (backlog #937). These tests pass in isolation but flake under the loaded -# parallel run for two distinct reasons: +# Serialize the ecstore tests that share the process-wide disk registry or +# exercise a multi-disk commit handoff across nextest process boundaries. # # * store::bucket::tests::bucket_delete_* share process/global state (disk # registry, lock client) and race make_bucket into InsufficientWriteQuorum # when run concurrently with other ecstore tests. # * bucket_lifecycle_ops::tests::concurrent_resend_same_part_commits_one_generation -# asserts a lock-acquire correctness property whose serialized cross-disk -# commits exceed the (already max'd, 60s) acquire deadline only when the -# suite saturates disk I/O. +# uses the shared multipart fixture and a deterministic uploadId-lock +# handoff, so it must not overlap another process mutating that fixture. # # serial_test's #[serial] attribute does NOT serialize these across runs: # nextest executes each test in its own process, where the in-process @@ -100,13 +97,6 @@ path = "junit.xml" # profile's own overrides list, not the default profile's). # =========================================================================== -# QUARANTINE: OPEN backlog#937 — concurrent_resend lock-acquire deadline flakes -# under saturated disk I/O in the full parallel suite. -[[profile.ci.overrides]] -filter = 'package(rustfs-ecstore) & test(concurrent_resend_same_part_commits_one_generation)' -test-group = 'ecstore-serial-flaky' -retries = 2 - # QUARANTINE: OPEN backlog#937 — store::bucket::tests::bucket_delete_* race # make_bucket into InsufficientWriteQuorum via shared global state under load. [[profile.ci.overrides]] @@ -114,6 +104,11 @@ filter = 'package(rustfs-ecstore) & test(/^store::bucket::tests::bucket_delete_( test-group = 'ecstore-serial-flaky' retries = 2 +# Keep the deterministic multipart handoff isolated across nextest processes. +[[profile.ci.overrides]] +filter = 'package(rustfs-ecstore) & test(concurrent_resend_same_part_commits_one_generation)' +test-group = 'ecstore-serial-flaky' + # QUARANTINE: OPEN rustfs#4690 — walk_dir stall-budget accounting test depends # on producer/consumer timing windows that stretch past the budget on loaded # CI runners (regression test for rustfs#4644; failed on a zero-Rust-diff PR). diff --git a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs index 901f4ae50..9033fe275 100644 --- a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs +++ b/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 diff --git a/crates/ecstore/src/set_disk/mod.rs b/crates/ecstore/src/set_disk/mod.rs index bdd66c945..692571243 100644 --- a/crates/ecstore/src/set_disk/mod.rs +++ b/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; diff --git a/crates/ecstore/src/set_disk/ops/multipart.rs b/crates/ecstore/src/set_disk/ops/multipart.rs index fd3416fa8..790e3a519 100644 --- a/crates/ecstore/src/set_disk/ops/multipart.rs +++ b/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, } @@ -60,11 +63,12 @@ static MULTIPART_COMMIT_BARRIER: std::sync::OnceLock 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; } diff --git a/crates/ecstore/src/store/object.rs b/crates/ecstore/src/store/object.rs index 2a8d30ee4..159fbafa3 100644 --- a/crates/ecstore/src/store/object.rs +++ b/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() { diff --git a/rustfs/src/app/lifecycle_transition_api_test.rs b/rustfs/src/app/lifecycle_transition_api_test.rs index 77aa6dcfd..2053ebe5c 100644 --- a/rustfs/src/app/lifecycle_transition_api_test.rs +++ b/rustfs/src/app/lifecycle_transition_api_test.rs @@ -60,7 +60,6 @@ use uuid::Uuid; static GLOBAL_ENV: OnceLock<(Vec, Arc)> = OnceLock::new(); static INIT: Once = Once::new(); const TRANSITION_WAIT_TIMEOUT: Duration = Duration::from_secs(15); -const RESTORE_COPY_BACK_WAIT_TIMEOUT: Duration = Duration::from_secs(60); const ENV_GET_CODEC_STREAMING_ENABLE: &str = "RUSTFS_GET_CODEC_STREAMING_ENABLE"; const ENV_GET_CODEC_STREAMING_ROLLOUT: &str = "RUSTFS_GET_CODEC_STREAMING_ROLLOUT"; const ENV_GET_CODEC_STREAMING_BODY_COMPAT_CONFIRMED: &str = "RUSTFS_GET_CODEC_STREAMING_BODY_COMPAT_CONFIRMED"; @@ -339,45 +338,6 @@ async fn wait_for_transition(ecstore: &Arc, bucket: &str, object: &str, } } -async fn wait_for_restore_completion( - ecstore: &Arc, - backend: &MockWarmBackend, - bucket: &str, - object: &str, - timeout: Duration, -) -> Result { - let deadline = tokio::time::Instant::now() + timeout; - let mut last_state = None; - - loop { - if tokio::time::Instant::now() >= deadline { - let tier_gets = backend.get_count().await; - let op_log = backend.op_log().await; - return Err(format!( - "restore copy-back should complete within {timeout:?}; tier_gets={tier_gets}, op_log={op_log:?}; last observed state: {}", - last_state.unwrap_or_else(|| "no object info observed".to_string()) - )); - } - - match (**ecstore).get_object_info(bucket, object, &ObjectOptions::default()).await { - Ok(info) => { - if !info.restore_ongoing && info.restore_expires.is_some() { - return Ok(info); - } - last_state = Some(format!( - "restore_ongoing={}, restore_expires={:?}, transitioned_status={}", - info.restore_ongoing, info.restore_expires, info.transitioned_object.status - )); - } - Err(err) => { - last_state = Some(format!("get_object_info failed: {err}")); - } - } - - tokio::time::sleep(Duration::from_millis(500)).await; - } -} - // SAFETY: this helper is used only by `#[serial]` tests and runs under the single-threaded Tokio // runtime (`worker_threads = 1`), so no concurrent test can mutate process environment during the // `env::set_var` / `env::remove_var` window. @@ -2097,10 +2057,9 @@ async fn put_bucket_lifecycle_configuration_rejects_zero_day_expiration() { /// POST restore(days=1) is accepted and flips the object to /// `x-amz-restore: ongoing-request="true"` while the mock tier GET barrier /// proves the background copy-back has reached the remote read; a second POST -/// during that window is rejected with 409 `RestoreAlreadyInProgress`; once the -/// copy-back completes the object reports `ongoing-request="false"` with a -/// future expiry-date; and a full GET is then served from the local restored -/// copy (the mock tier records no further `get` calls). +/// during that window is rejected with 409 `RestoreAlreadyInProgress`. +/// Synchronous SetDisks transition tests cover copy-back completion, restore +/// metadata, and local byte-identical reads. /// /// Re-enabled in the serial lane by backlog#1304: the accept path now flips /// the ongoing flag under a short compare-and-set guard and the copy-back @@ -2110,7 +2069,7 @@ async fn put_bucket_lifecycle_configuration_rejects_zero_day_expiration() { #[tokio::test(flavor = "multi_thread", worker_threads = 2)] #[serial] #[ignore = "global-state ILM integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial and rustfs/backlog#1148 (ilm-8)"] -async fn restore_object_usecase_reports_ongoing_conflict_and_completion() { +async fn restore_object_usecase_reports_ongoing_conflict() { let (_disk_paths, ecstore) = setup_test_env().await; let usecase = DefaultObjectUsecase::from_global(); @@ -2177,35 +2136,6 @@ async fn restore_object_usecase_reports_ongoing_conflict_and_completion() { ); get_barrier.release(); - - // Completion: ongoing flips to false and a future expiry-date appears. - let completed = - wait_for_restore_completion(&ecstore, &backend, bucket.as_str(), object, RESTORE_COPY_BACK_WAIT_TIMEOUT).await; - let completed = completed.unwrap_or_else(|err| panic!("{err}")); - - let now_secs = std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .expect("clock before unix epoch") - .as_secs() as i64; - let expires = completed.restore_expires.expect("completed restore carries an expiry"); - assert!( - expires.unix_timestamp() > now_secs, - "restore expiry-date must be in the future, got {expires}" - ); - assert_eq!( - completed.transitioned_object.status, "complete", - "restore must not clear the transitioned state" - ); - - // The restored copy serves GET locally: no further tier GETs. - let tier_gets_after_restore = backend.get_count().await; - let data = read_object_bytes(&ecstore, bucket.as_str(), object).await; - assert_eq!(data, payload, "restored GET must return the original bytes"); - assert_eq!( - backend.get_count().await, - tier_gets_after_restore, - "GET of a restored object must be served locally, not from the tier" - ); } /// backlog#1304: the restore-accept compare-and-set itself, under real