diff --git a/crates/ecstore/src/services/tier/tier.rs b/crates/ecstore/src/services/tier/tier.rs index 27e7ed643..aa771c514 100644 --- a/crates/ecstore/src/services/tier/tier.rs +++ b/crates/ecstore/src/services/tier/tier.rs @@ -5611,6 +5611,80 @@ mod tests { assert_eq!(lock_unpoisoned(&calls).as_slice(), &[format!("peer-a:commit:{mutation_id}:etag-new")]); } + #[tokio::test] + async fn committed_replay_mixed_version_peer_matrix_fails_before_local_publish() { + for (case, peer_error, expected_error) in [ + ( + "old_unimplemented", + "peer tier mutation commit RPC failed: status: Unimplemented, message: \"old peer has no tier mutation control service\"", + "old peer", + ), + ( + "deadline_exceeded", + "peer tier mutation commit RPC failed: status: DeadlineExceeded, message: \"peer tier mutation control timed out\"", + "timed out", + ), + ( + "unavailable", + "peer tier mutation commit RPC failed: status: Unavailable, message: \"peer tier mutation control unavailable\"", + "unavailable", + ), + ] { + let store = Arc::new(CasConfigStore::default()); + let mut persisted = empty_mgr(); + persisted.tiers.insert("COLD-A".to_string(), build_rustfs_tier("COLD-A")); + persisted + .save_tiering_config_if_current(store.clone(), None) + .await + .expect("tier config fixture should persist"); + + let mutation_id = uuid::Uuid::from_u128(match case { + "old_unimplemented" => 31, + "deadline_exceeded" => 32, + "unavailable" => 33, + _ => unreachable!("matrix cases are enumerated above"), + }); + let intent = committed_remove_intent("COLD-A", mutation_id, "etag-new"); + crate::services::tier::tier_mutation_intent::save_tier_mutation_intent_record(store.clone(), &intent) + .await + .expect("committed intent fixture should persist"); + let writes_after_fixture = store.next_etag.load(Ordering::SeqCst); + let calls = Arc::new(Mutex::new(Vec::new())); + let handle = TierConfigMgr::new(); + + let err = TIER_MUTATION_TEST_COMMIT_PEERS + .scope(vec![FakeTierMutationCommitPeer::boxed("peer-a", calls.clone(), Err(peer_error))], async { + TierConfigMgr::reload_handle_with(&handle, store.clone()).await + }) + .await + .expect_err("mixed-version peer failure must abort committed replay"); + + assert!(err.to_string().contains(expected_error), "{case} returned {err}"); + assert_eq!(lock_unpoisoned(&calls).as_slice(), &[format!("peer-a:commit:{mutation_id}:etag-new")]); + assert!( + !handle.read().await.tiers.contains_key("COLD-A"), + "{case} must not publish the local tier config after peer replay fails" + ); + assert_eq!( + store.next_etag.load(Ordering::SeqCst), + writes_after_fixture, + "{case} must not perform a second tier-config CAS write after replay fails" + ); + let reloaded = TierConfigMgr::load_tier_mutation_intents(store) + .await + .expect("intent scan should succeed after failed mixed-version replay"); + assert!( + reloaded.iter().any(|intent| intent.mutation_id == mutation_id), + "{case} must keep the committed intent for retry" + ); + let blocked = match TierConfigMgr::acquire_operation_lease(&handle, "COLD-A").await { + Ok(_) => panic!("{case} must keep committed replay blocking old tier operations"), + Err(err) => err, + }; + assert!(blocked.message.contains("being replaced"), "{case} returned {blocked}"); + } + } + #[tokio::test] async fn committed_mutation_recovery_requires_every_peer_to_commit() { let mutation_id = uuid::Uuid::from_u128(17); diff --git a/rustfs/src/storage/rpc/node_service.rs b/rustfs/src/storage/rpc/node_service.rs index 70e6c2f4e..07a9c1473 100644 --- a/rustfs/src/storage/rpc/node_service.rs +++ b/rustfs/src/storage/rpc/node_service.rs @@ -2318,6 +2318,33 @@ mod tests { assert_eq!(error.code(), tonic::Code::InvalidArgument); } + #[tokio::test] + async fn tier_mutation_control_rejects_old_protocol_version_before_store_lookup() { + let service = make_tier_mutation_control_server_for_context(None); + let mutation_id = uuid::Uuid::new_v4(); + let payload = Bytes::from_static(b"intent"); + let mut request = Request::new(TierMutationPrepareRequest { + version: rustfs_protos::TIER_MUTATION_RPC_PROTOCOL_VERSION - 1, + mutation_id: mutation_id.to_string(), + canonical_payload: payload, + }); + let body = rustfs_protos::canonical_tier_mutation_rpc_body( + request.get_ref().version, + rustfs_protos::TierMutationRpcPhase::Prepare, + mutation_id, + &request.get_ref().canonical_payload, + ) + .expect("old-version request should encode for rejection test"); + set_tonic_canonical_body_digest(&mut request, &body).expect("digest metadata should encode"); + mark_v2_authenticated(&mut request); + + let error = service + .prepare_tier_mutation(request) + .await + .expect_err("old tier mutation protocol version must fail closed"); + assert_eq!(error.code(), tonic::Code::FailedPrecondition); + } + #[tokio::test] async fn tier_mutation_control_rejects_oversized_prepare_before_auth_and_store_lookup() { let service = make_tier_mutation_control_server_for_context(None);