From 7f569b67cb997f951219e3c1ef4128cd0288d2bf Mon Sep 17 00:00:00 2001 From: houseme Date: Tue, 21 Jul 2026 16:10:31 +0800 Subject: [PATCH] fix(tier): add mutation intent CAS advance (#5078) Add idempotent terminal transition handling for tier mutation intents and a compare-and-swap record update helper backed by config-object ETags. Cover commit and abort retries, conflicting terminal updates, stale ETag rejection, and exact-limit scan pagination. Co-authored-by: heihutu --- .../src/services/tier/tier_mutation_intent.rs | 156 +++++++++++++- crates/ecstore/src/store/init.rs | 193 +++++++++++++++++- 2 files changed, 344 insertions(+), 5 deletions(-) diff --git a/crates/ecstore/src/services/tier/tier_mutation_intent.rs b/crates/ecstore/src/services/tier/tier_mutation_intent.rs index 061617736..dec3260e0 100644 --- a/crates/ecstore/src/services/tier/tier_mutation_intent.rs +++ b/crates/ecstore/src/services/tier/tier_mutation_intent.rs @@ -21,8 +21,9 @@ use uuid::Uuid; use crate::config::com; use crate::disk::RUSTFS_META_BUCKET; use crate::error::{Error, Result as EcstoreResult}; +use crate::object_api::ObjectOptions; use crate::services::tier::tier::TierDestinationId; -use crate::storage_api_contracts::list::ListOperations as _; +use crate::storage_api_contracts::{list::ListOperations as _, object::HTTPPreconditions}; use crate::store::ECStore; pub(crate) const TIER_MUTATION_INTENT_SCHEMA: &str = "rustfs-tier-mutation-intent-v1"; @@ -215,6 +216,42 @@ impl TierMutationIntent { self.validate() } + pub(crate) fn advance_idempotent( + &mut self, + next: TierMutationIntentState, + committed_config_etag: Option, + ) -> Result { + match (self.state, next) { + (TierMutationIntentState::Prepared, TierMutationIntentState::Committed | TierMutationIntentState::Aborted) => { + self.advance(next, committed_config_etag)?; + Ok(true) + } + (TierMutationIntentState::Committed, TierMutationIntentState::Committed) => { + let committed_config_etag = + committed_config_etag.ok_or(TierMutationIntentError::Corrupt("commit requires config etag"))?; + if committed_config_etag.is_empty() { + return Err(TierMutationIntentError::Corrupt("commit config etag is empty")); + } + if self.committed_config_etag.as_deref() != Some(committed_config_etag.as_str()) { + return Err(TierMutationIntentError::Corrupt("committed config etag does not match")); + } + self.validate()?; + Ok(false) + } + (TierMutationIntentState::Aborted, TierMutationIntentState::Aborted) => { + if committed_config_etag.is_some() { + return Err(TierMutationIntentError::Corrupt("abort must not carry committed config etag")); + } + self.validate()?; + Ok(false) + } + _ => Err(TierMutationIntentError::InvalidStateChange { + from: self.state, + to: next, + }), + } + } + pub(crate) fn encode(&self) -> Result> { self.validate()?; let intent_bytes = serde_json::to_vec(self)?; @@ -310,9 +347,48 @@ pub(crate) async fn save_tier_mutation_intent_record(api: Arc, intent: } pub(crate) async fn load_tier_mutation_intent_record(api: Arc, mutation_id: Uuid) -> EcstoreResult { + let (intent, _) = load_tier_mutation_intent_record_with_etag(api, mutation_id).await?; + Ok(intent) +} + +pub(crate) async fn load_tier_mutation_intent_record_with_etag( + api: Arc, + mutation_id: Uuid, +) -> EcstoreResult<(TierMutationIntent, String)> { let object = tier_mutation_intent_record_object_name(mutation_id).map_err(tier_mutation_intent_store_error)?; - let data = com::read_config(api, &object).await?; - TierMutationIntent::decode(mutation_id, &data).map_err(tier_mutation_intent_store_error) + let (data, object_info) = com::read_config_with_metadata(api, &object, &ObjectOptions::default()).await?; + let etag = object_info + .etag + .filter(|etag| !etag.trim().is_empty()) + .ok_or_else(|| Error::other("tier mutation intent record is missing an ETag"))?; + let intent = TierMutationIntent::decode(mutation_id, &data).map_err(tier_mutation_intent_store_error)?; + Ok((intent, etag)) +} + +pub(crate) async fn save_tier_mutation_intent_record_if_current( + api: Arc, + intent: &TierMutationIntent, + current_etag: &str, +) -> EcstoreResult<()> { + if current_etag.trim().is_empty() { + return Err(Error::other("tier mutation intent current ETag is empty")); + } + let object = tier_mutation_intent_record_object_name(intent.mutation_id).map_err(tier_mutation_intent_store_error)?; + let data = intent.encode().map_err(tier_mutation_intent_store_error)?; + com::save_config_with_opts( + api, + &object, + data, + &ObjectOptions { + max_parity: true, + http_preconditions: Some(HTTPPreconditions { + if_match: Some(current_etag.to_string()), + ..Default::default() + }), + ..Default::default() + }, + ) + .await } pub(crate) async fn delete_tier_mutation_intent_record(api: Arc, mutation_id: Uuid) -> EcstoreResult<()> { @@ -323,6 +399,22 @@ pub(crate) async fn delete_tier_mutation_intent_record(api: Arc, mutati } } +pub(crate) async fn advance_tier_mutation_intent_record_idempotent( + api: Arc, + mutation_id: Uuid, + next: TierMutationIntentState, + committed_config_etag: Option, +) -> EcstoreResult<(TierMutationIntent, bool)> { + let (mut intent, current_etag) = load_tier_mutation_intent_record_with_etag(api.clone(), mutation_id).await?; + let advanced = intent + .advance_idempotent(next, committed_config_etag) + .map_err(tier_mutation_intent_store_error)?; + if advanced { + save_tier_mutation_intent_record_if_current(api, &intent, ¤t_etag).await?; + } + Ok((intent, advanced)) +} + pub(crate) async fn list_tier_mutation_intent_records( api: Arc, limit: usize, @@ -497,6 +589,64 @@ mod tests { )); } + #[test] + fn intent_idempotent_commit_retry_keeps_revision_and_rejects_conflicts() { + let mut intent = prepared_intent(); + + assert!( + intent + .advance_idempotent(TierMutationIntentState::Committed, Some("new-etag".to_string())) + .expect("initial commit should advance") + ); + assert_eq!(intent.revision, 2); + assert!( + !intent + .advance_idempotent(TierMutationIntentState::Committed, Some("new-etag".to_string())) + .expect("same commit retry should be idempotent") + ); + assert_eq!(intent.revision, 2); + assert!(matches!( + intent.advance_idempotent(TierMutationIntentState::Committed, Some("other-etag".to_string())), + Err(TierMutationIntentError::Corrupt("committed config etag does not match")) + )); + assert!(matches!( + intent.advance_idempotent(TierMutationIntentState::Aborted, None), + Err(TierMutationIntentError::InvalidStateChange { + from: TierMutationIntentState::Committed, + to: TierMutationIntentState::Aborted, + }) + )); + } + + #[test] + fn intent_idempotent_abort_retry_keeps_revision_and_rejects_commit() { + let mut intent = prepared_intent(); + + assert!( + intent + .advance_idempotent(TierMutationIntentState::Aborted, None) + .expect("initial abort should advance") + ); + assert_eq!(intent.revision, 2); + assert!( + !intent + .advance_idempotent(TierMutationIntentState::Aborted, None) + .expect("same abort retry should be idempotent") + ); + assert_eq!(intent.revision, 2); + assert!(matches!( + intent.advance_idempotent(TierMutationIntentState::Aborted, Some("unexpected-etag".to_string())), + Err(TierMutationIntentError::Corrupt("abort must not carry committed config etag")) + )); + assert!(matches!( + intent.advance_idempotent(TierMutationIntentState::Committed, Some("new-etag".to_string())), + Err(TierMutationIntentError::InvalidStateChange { + from: TierMutationIntentState::Aborted, + to: TierMutationIntentState::Committed, + }) + )); + } + #[test] fn intent_validation_rejects_placeholder_identity() { let mut intent = prepared_intent(); diff --git a/crates/ecstore/src/store/init.rs b/crates/ecstore/src/store/init.rs index 81af5fc3f..d7da6dad9 100644 --- a/crates/ecstore/src/store/init.rs +++ b/crates/ecstore/src/store/init.rs @@ -549,8 +549,9 @@ mod tests { tier::TierConfigMgr, tier_mutation_intent::{ TIER_MUTATION_INTENT_RECORD_PREFIX, TierMutationIntent, TierMutationIntentKind, TierMutationIntentState, - TierMutationIntentTarget, delete_tier_mutation_intent_record, list_tier_mutation_intent_records, - load_tier_mutation_intent_record, save_tier_mutation_intent_record, + TierMutationIntentTarget, advance_tier_mutation_intent_record_idempotent, delete_tier_mutation_intent_record, + list_tier_mutation_intent_records, load_tier_mutation_intent_record, load_tier_mutation_intent_record_with_etag, + save_tier_mutation_intent_record, save_tier_mutation_intent_record_if_current, }, warm_backend::WarmBackend, }, @@ -1340,6 +1341,194 @@ mod tests { assert_eq!(scan.next_marker, None); } + #[cfg(feature = "test-util")] + #[tokio::test] + #[serial_test::serial(storage_class_env)] + async fn tier_mutation_intent_record_advance_is_idempotent_in_config_store() { + let temp_dir = tempfile::tempdir().expect("create temp store dir"); + let (_ctx, store, _shutdown) = + without_storage_class_env(build_isolated_test_store(temp_dir.path(), "tier-mutation-intent-advance", &[4])).await; + let mutation_id = uuid::Uuid::new_v4(); + let intent = TierMutationIntent { + mutation_id, + revision: 1, + kind: TierMutationIntentKind::Edit, + state: TierMutationIntentState::Prepared, + old_config_etag: Some("old-etag".to_string()), + committed_config_etag: None, + candidate_digest: [3; 32], + affected_targets: vec![TierMutationIntentTarget { + tier_name: "COLD-A".to_string(), + old_backend_identity: Some([1; 32]), + new_backend_identity: Some([2; 32]), + }], + expires_at_unix_nanos: 1_780_000_000_000_000_000, + }; + save_tier_mutation_intent_record(store.clone(), &intent) + .await + .expect("prepared tier mutation intent record should persist"); + let (loaded_before_advance, stale_etag) = load_tier_mutation_intent_record_with_etag(store.clone(), mutation_id) + .await + .expect("prepared tier mutation intent record should load with etag"); + assert_eq!(loaded_before_advance, intent); + assert!(!stale_etag.is_empty()); + + let (committed, first_advanced) = advance_tier_mutation_intent_record_idempotent( + store.clone(), + mutation_id, + TierMutationIntentState::Committed, + Some("new-etag".to_string()), + ) + .await + .expect("first commit should advance the record"); + assert!(first_advanced); + assert_eq!(committed.state, TierMutationIntentState::Committed); + assert_eq!(committed.revision, 2); + assert_eq!(committed.committed_config_etag.as_deref(), Some("new-etag")); + + let (retried, retry_advanced) = advance_tier_mutation_intent_record_idempotent( + store.clone(), + mutation_id, + TierMutationIntentState::Committed, + Some("new-etag".to_string()), + ) + .await + .expect("same commit retry should be idempotent"); + assert!(!retry_advanced); + assert_eq!(retried, committed); + + let conflict = advance_tier_mutation_intent_record_idempotent( + store.clone(), + mutation_id, + TierMutationIntentState::Committed, + Some("other-etag".to_string()), + ) + .await + .expect_err("conflicting commit retry should fail closed"); + assert!(matches!(conflict, Error::Io(_))); + assert!(conflict.to_string().contains("committed config etag does not match")); + + let mut stale_conflict = intent; + stale_conflict + .advance_idempotent(TierMutationIntentState::Committed, Some("other-etag".to_string())) + .expect("stale conflicting intent should advance locally"); + let stale_save = save_tier_mutation_intent_record_if_current(store.clone(), &stale_conflict, &stale_etag) + .await + .expect_err("stale etag must fail closed instead of overwriting the committed record"); + assert!(matches!(stale_save, Error::PreconditionFailed)); + + let loaded = load_tier_mutation_intent_record(store.clone(), mutation_id) + .await + .expect("conflicting retry must not overwrite the durable record"); + assert_eq!(loaded, committed); + + let abort_id = uuid::Uuid::new_v4(); + let abort_intent = TierMutationIntent { + mutation_id: abort_id, + revision: 1, + kind: TierMutationIntentKind::Edit, + state: TierMutationIntentState::Prepared, + old_config_etag: Some("old-etag".to_string()), + committed_config_etag: None, + candidate_digest: [4; 32], + affected_targets: vec![TierMutationIntentTarget { + tier_name: "COLD-B".to_string(), + old_backend_identity: Some([1; 32]), + new_backend_identity: Some([2; 32]), + }], + expires_at_unix_nanos: 1_780_000_000_000_000_000, + }; + save_tier_mutation_intent_record(store.clone(), &abort_intent) + .await + .expect("prepared abort intent record should persist"); + + let (aborted, first_abort_advanced) = + advance_tier_mutation_intent_record_idempotent(store.clone(), abort_id, TierMutationIntentState::Aborted, None) + .await + .expect("first abort should advance the record"); + assert!(first_abort_advanced); + assert_eq!(aborted.state, TierMutationIntentState::Aborted); + assert_eq!(aborted.revision, 2); + assert_eq!(aborted.committed_config_etag, None); + + let (aborted_retry, retry_abort_advanced) = + advance_tier_mutation_intent_record_idempotent(store, abort_id, TierMutationIntentState::Aborted, None) + .await + .expect("same abort retry should be idempotent"); + assert!(!retry_abort_advanced); + assert_eq!(aborted_retry, aborted); + } + + #[cfg(feature = "test-util")] + #[tokio::test] + #[serial_test::serial(storage_class_env)] + async fn tier_mutation_intent_record_scan_paginates_exact_limit() { + let temp_dir = tempfile::tempdir().expect("create temp store dir"); + let (_ctx, store, _shutdown) = + without_storage_class_env(build_isolated_test_store(temp_dir.path(), "tier-mutation-intent-scan-page", &[4])).await; + let build_intent = |mutation_id: uuid::Uuid, tier_name: &str| TierMutationIntent { + mutation_id, + revision: 1, + kind: TierMutationIntentKind::Edit, + state: TierMutationIntentState::Prepared, + old_config_etag: Some("old-etag".to_string()), + committed_config_etag: None, + candidate_digest: [3; 32], + affected_targets: vec![TierMutationIntentTarget { + tier_name: tier_name.to_string(), + old_backend_identity: Some([1; 32]), + new_backend_identity: Some([2; 32]), + }], + expires_at_unix_nanos: 1_780_000_000_000_000_000, + }; + let intent_ids = vec![ + uuid::Uuid::parse_str("11345678-1234-5678-9abc-def012345678").expect("first uuid should parse"), + uuid::Uuid::parse_str("22345678-1234-5678-9abc-def012345678").expect("second uuid should parse"), + uuid::Uuid::parse_str("33345678-1234-5678-9abc-def012345678").expect("third uuid should parse"), + ]; + for (index, mutation_id) in intent_ids.iter().copied().enumerate() { + let intent = build_intent(mutation_id, &format!("COLD-{index}")); + save_tier_mutation_intent_record(store.clone(), &intent) + .await + .expect("tier mutation intent record should persist"); + } + + let exact_page = list_tier_mutation_intent_records(store.clone(), 3, None) + .await + .expect("exact full page should scan"); + assert_eq!(exact_page.scanned, 3); + assert_eq!(exact_page.intents.len(), 3); + assert_eq!(exact_page.failed, 0); + assert!(!exact_page.truncated); + assert_eq!(exact_page.next_marker, None); + + let first_page = list_tier_mutation_intent_records(store.clone(), 2, None) + .await + .expect("first page should scan"); + assert_eq!(first_page.scanned, 2); + assert_eq!(first_page.intents.len(), 2); + assert_eq!(first_page.failed, 0); + assert!(first_page.truncated); + assert!(first_page.next_marker.is_some()); + + let second_page = list_tier_mutation_intent_records(store, 2, first_page.next_marker) + .await + .expect("second page should scan"); + let mut loaded_ids: Vec<_> = first_page + .intents + .into_iter() + .chain(second_page.intents.into_iter()) + .map(|intent| intent.mutation_id) + .collect(); + loaded_ids.sort(); + + assert_eq!(second_page.scanned, 1); + assert_eq!(second_page.failed, 0); + assert!(!second_page.truncated); + assert_eq!(second_page.next_marker, None); + assert_eq!(loaded_ids, intent_ids); + } + #[cfg(feature = "test-util")] #[tokio::test] #[serial_test::serial(storage_class_env)]