From c3242f83bac93e6a4d0419c373bbf0b20505128d Mon Sep 17 00:00:00 2001 From: houseme Date: Sun, 26 Jul 2026 20:37:14 +0800 Subject: [PATCH] test(tier): cover reporter committed prepare replay (#5297) Add a deterministic four-hot AddTier regression for the reporter path where one peer has already durably committed prepare replay and the remaining peers still need commit fanout before local publish. Co-authored-by: heihutu --- crates/ecstore/src/services/tier/tier.rs | 108 +++++++++++++++++++++++ 1 file changed, 108 insertions(+) diff --git a/crates/ecstore/src/services/tier/tier.rs b/crates/ecstore/src/services/tier/tier.rs index 89b4663a5..c501d52ec 100644 --- a/crates/ecstore/src/services/tier/tier.rs +++ b/crates/ecstore/src/services/tier/tier.rs @@ -6748,6 +6748,114 @@ mod tests { ); } + #[tokio::test] + async fn coordinator_fanout_converges_four_hot_reporter_committed_prepare_replay() { + let manager = TierConfigMgr::new(); + let store = Arc::new(CasConfigStore::default()); + empty_mgr() + .save_tiering_config_if_current(store.clone(), None) + .await + .expect("empty tier config fixture should persist"); + let (candidate, version) = load_tier_config_for_update(store.clone()) + .await + .expect("tier config fixture should reload"); + let update = TierConfigMgr::admin_update_lock(&manager).await; + let calls = Arc::new(Mutex::new(Vec::new())); + + TIER_MUTATION_TEST_PEERS + .scope( + vec![ + FakeTierMutationPeer::boxed_with_prepare_commit( + "peer-a", + calls.clone(), + Ok(PeerTierMutationState::Committed), + Ok(PeerTierMutationState::Committed), + ), + FakeTierMutationPeer::boxed_with_prepare_commit( + "peer-b", + calls.clone(), + Ok(PeerTierMutationState::Prepared), + Ok(PeerTierMutationState::Committed), + ), + FakeTierMutationPeer::boxed_with_prepare_commit( + "peer-c", + calls.clone(), + Ok(PeerTierMutationState::Prepared), + Ok(PeerTierMutationState::Committed), + ), + ], + async { + TierConfigMgr::update_candidate_owned( + &manager, + store.clone(), + candidate, + version, + TierCandidateMutation::Add(build_rustfs_tier("COLD-A"), true), + update, + None, + ) + .await + .expect("four-hot AddTier reporter replay should publish after prepared peers commit"); + }, + ) + .await; + + let calls = lock_unpoisoned(&calls).clone(); + let reporter_prepare_call = calls + .iter() + .find(|call| call.starts_with("peer-a:prepare:")) + .expect("reporter peer should receive prepare replay"); + let reporter_mutation_id = reporter_prepare_call + .split(':') + .nth(2) + .expect("reporter prepare call should record the mutation id"); + assert!( + !calls.iter().any(|call| call.starts_with("peer-a:commit:")), + "already-committed reporter must not receive a duplicate commit: {calls:?}" + ); + + let saved_etag = load_tier_config_for_update(store) + .await + .expect("saved tier config should reload") + .1 + .expect("saved tier config should have an ETag"); + for peer in ["peer-b", "peer-c"] { + let prepare_index = calls + .iter() + .position(|call| call.starts_with(&format!("{peer}:prepare:"))) + .expect("prepared peer should receive prepare"); + let commit_index = calls + .iter() + .position(|call| call.starts_with(&format!("{peer}:commit:"))) + .expect("prepared peer should receive commit"); + assert!(prepare_index < commit_index, "{peer} commit must follow prepare: {calls:?}"); + let prepared_mutation_id = calls[prepare_index] + .split(':') + .nth(2) + .expect("prepare call should record the mutation id"); + let committed_mutation_id = calls[commit_index] + .split(':') + .nth(2) + .expect("commit call should record the mutation id"); + assert_eq!( + prepared_mutation_id, reporter_mutation_id, + "{peer} prepare should share the reporter mutation id" + ); + assert_eq!( + committed_mutation_id, reporter_mutation_id, + "{peer} commit should share the reporter mutation id" + ); + assert!( + calls[commit_index].ends_with(&format!(":{saved_etag}")), + "{peer} commit should carry the saved tier config ETag: {calls:?}" + ); + } + assert!( + manager.read().await.tiers.contains_key("COLD-A"), + "local manager should publish the AddTier candidate after peer convergence" + ); + } + #[tokio::test] async fn reload_handle_replays_committed_mutation_and_removes_intent_record() { let store = Arc::new(CasConfigStore::default());