diff --git a/crates/ecstore/src/bucket/lifecycle/tier_delete_journal.rs b/crates/ecstore/src/bucket/lifecycle/tier_delete_journal.rs index 474036ef8..39d07ed05 100644 --- a/crates/ecstore/src/bucket/lifecycle/tier_delete_journal.rs +++ b/crates/ecstore/src/bucket/lifecycle/tier_delete_journal.rs @@ -152,13 +152,13 @@ pub(crate) fn tier_delete_journal_object_name(je: &Jentry) -> String { ) } -fn decode_tier_delete_journal_entry(data: &[u8]) -> Result { +pub(crate) fn decode_tier_delete_journal_entry(data: &[u8]) -> Result { let persisted: PersistedTierDeleteJournalEntry = serde_json::from_slice(data).map_err(|err| Error::other(format!("decode tier delete journal failed: {err}")))?; persisted.into_jentry() } -fn encode_tier_delete_journal_entry(je: &Jentry) -> Result> { +pub(crate) fn encode_tier_delete_journal_entry(je: &Jentry) -> Result> { serde_json::to_vec(&PersistedTierDeleteJournalEntry::from_jentry(je)) .map_err(|err| Error::other(format!("encode tier delete journal failed: {err}"))) } diff --git a/crates/ecstore/src/bucket/lifecycle/transition_transaction.rs b/crates/ecstore/src/bucket/lifecycle/transition_transaction.rs index 282a090c2..823ae1d2a 100644 --- a/crates/ecstore/src/bucket/lifecycle/transition_transaction.rs +++ b/crates/ecstore/src/bucket/lifecycle/transition_transaction.rs @@ -613,6 +613,11 @@ pub enum TransitionTransactionRecoveryOutcome { Retained, } +pub(crate) fn decode_transition_transaction_record(object: &str, data: &[u8]) -> Result { + let transaction_id = transition_transaction_id_from_record_object_name(object)?; + TransitionTransaction::decode(transaction_id, data) +} + fn transition_transaction_id_from_record_object_name(object: &str) -> Result { let prefix = format!("{TRANSITION_TRANSACTION_RECORD_PREFIX}/"); let suffix = object diff --git a/crates/ecstore/src/services/tier/tier.rs b/crates/ecstore/src/services/tier/tier.rs index e63cf5d72..e3e81b40d 100644 --- a/crates/ecstore/src/services/tier/tier.rs +++ b/crates/ecstore/src/services/tier/tier.rs @@ -65,6 +65,10 @@ use crate::storage_api_contracts::{ range::HTTPRangeSpec, }; use crate::{ + bucket::lifecycle::{ + tier_delete_journal::{TIER_DELETE_JOURNAL_PREFIX, decode_tier_delete_journal_entry}, + transition_transaction::{TRANSITION_TRANSACTION_RECORD_PREFIX, decode_transition_transaction_record}, + }, cluster::rpc::peer_rest_client::{PeerRestClient, PeerTierMutationState}, config::com::{CONFIG_PREFIX, read_config, read_config_with_metadata}, disk::{MIGRATING_META_BUCKET, RUSTFS_META_BUCKET}, @@ -639,17 +643,20 @@ async fn ensure_no_authoritative_tier_object_references( where S: TierReferenceProofStore, { - for target in affected_targets { - if target.old_backend_identity.is_some() { - ensure_no_authoritative_target_references(api.clone(), target).await?; - } + let targets = affected_targets + .iter() + .filter(|target| target.old_backend_identity.is_some()) + .cloned() + .collect::>(); + if targets.is_empty() { + return Ok(()); } - Ok(()) + ensure_no_authoritative_target_references(api, &targets).await } async fn ensure_no_authoritative_target_references( api: Arc, - target: &TierMutationIntentTarget, + targets: &[TierMutationIntentTarget], ) -> std::result::Result<(), AdminError> where S: TierReferenceProofStore, @@ -675,8 +682,8 @@ where .await .map_err(tier_reference_proof_admin_error)?; for object in &page.objects { - if tier_object_blocks_target_rebind(object, target).map_err(tier_reference_proof_admin_error)? { - return Err(tier_reference_proof_in_use_error(&target.tier_name, object)); + if tier_object_blocks_any_target_rebind(object, targets).map_err(tier_reference_proof_admin_error)? { + return Err(tier_reference_proof_in_use_error(&object.transitioned_object.tier, object)); } } if !page.is_truncated { @@ -691,21 +698,122 @@ where } } } - Ok(()) + ensure_no_authoritative_persisted_references(api, targets).await } -fn tier_object_blocks_target_rebind(object: &ObjectInfo, target: &TierMutationIntentTarget) -> io::Result { - if object.transitioned_object.status != rustfs_filemeta::TRANSITION_COMPLETE - || object.transitioned_object.tier != target.tier_name +async fn ensure_no_authoritative_persisted_references( + api: Arc, + targets: &[TierMutationIntentTarget], +) -> std::result::Result<(), AdminError> +where + S: TierReferenceProofStore, +{ + ensure_no_authoritative_persisted_references_with(api.clone(), TIER_DELETE_JOURNAL_PREFIX, |_object, data| { + let journal = decode_tier_delete_journal_entry(data).map_err(io::Error::other)?; + Ok(( + journal.tier_name.clone(), + tier_persisted_reference_blocks_any_target(&journal.tier_name, journal.backend_identity, targets), + )) + }) + .await?; + ensure_no_authoritative_persisted_references_with(api, TRANSITION_TRANSACTION_RECORD_PREFIX, |object, data| { + let transaction = decode_transition_transaction_record(object, data).map_err(io::Error::other)?; + Ok(( + transaction.tier_name.clone(), + tier_persisted_reference_blocks_any_target(&transaction.tier_name, Some(transaction.backend_fingerprint), targets), + )) + }) + .await +} + +async fn ensure_no_authoritative_persisted_references_with( + api: Arc, + prefix: &str, + blocks_target: F, +) -> std::result::Result<(), AdminError> +where + S: TierReferenceProofStore, + F: Fn(&str, &[u8]) -> io::Result<(String, bool)>, +{ + let mut marker = None; + loop { + let page = api + .clone() + .list_objects_v2( + RUSTFS_META_BUCKET, + prefix, + marker.take(), + None, + TIER_REFERENCE_PROOF_LIST_LIMIT, + false, + None, + false, + ) + .await + .map_err(tier_reference_proof_admin_error)?; + for object in &page.objects { + let data = read_config(api.clone(), &object.name) + .await + .map_err(tier_reference_proof_admin_error)?; + let (tier_name, blocks) = blocks_target(&object.name, &data).map_err(tier_reference_proof_admin_error)?; + if blocks { + return Err(tier_reference_proof_persisted_in_use_error(&tier_name, &object.name)); + } + } + if !page.is_truncated { + return Ok(()); + } + marker = page.next_continuation_token; + if marker.is_none() { + return Err(tier_reference_proof_admin_error(io::Error::other( + "tier persisted reference proof listing is truncated without a next marker", + ))); + } + } +} + +fn tier_object_blocks_any_target_rebind(object: &ObjectInfo, targets: &[TierMutationIntentTarget]) -> io::Result { + if object.transitioned_object.status != rustfs_filemeta::TRANSITION_COMPLETE && !object.transitioned_object.free_version { + return Ok(false); + } + if !targets + .iter() + .any(|target| target.tier_name == object.transitioned_object.tier) { return Ok(false); } let current_identity = tier_destination_id_from_metadata(&object.user_defined)?; - Ok(match (¤t_identity, &target.new_backend_identity) { - (_, None) => true, - (Some(current), Some(new)) => current != new, - (None, Some(_)) => true, - }) + Ok(tier_persisted_reference_blocks_any_target( + &object.transitioned_object.tier, + current_identity, + targets, + )) +} + +fn tier_persisted_reference_blocks_any_target( + tier_name: &str, + backend_identity: Option, + targets: &[TierMutationIntentTarget], +) -> bool { + targets + .iter() + .any(|target| tier_persisted_reference_blocks_target(tier_name, backend_identity, target)) +} + +fn tier_object_blocks_target_rebind(object: &ObjectInfo, target: &TierMutationIntentTarget) -> io::Result { + tier_object_blocks_any_target_rebind(object, std::slice::from_ref(target)) +} + +fn tier_persisted_reference_blocks_target( + tier_name: &str, + backend_identity: Option, + target: &TierMutationIntentTarget, +) -> bool { + tier_name == target.tier_name + && match target.new_backend_identity { + None => true, + Some(new_backend_identity) => backend_identity != Some(new_backend_identity), + } } fn tier_reference_proof_in_use_error(tier_name: &str, object: &ObjectInfo) -> AdminError { @@ -717,6 +825,12 @@ fn tier_reference_proof_in_use_error(tier_name: &str, object: &ObjectInfo) -> Ad err } +fn tier_reference_proof_persisted_in_use_error(tier_name: &str, object: &str) -> AdminError { + let mut err = ERR_TIER_BACKEND_IN_USE.clone(); + err.message = format!("Remote tier {tier_name} still has a persisted reference, for example {object}"); + err +} + fn tier_reference_proof_admin_error(err: impl std::fmt::Display) -> AdminError { let mut admin_err = ERR_TIER_INVALID_CONFIG.clone(); admin_err.message = format!("Remote tier reference proof failed: {err}"); @@ -8029,6 +8143,13 @@ mod tests { .push(object); } + async fn insert_config_object(&self, object: String, data: Vec) { + self.objects + .lock() + .await + .insert(object, (data, "reference-proof-etag".to_string())); + } + fn omit_truncated_reference_marker(&self) { self.truncate_reference_page_without_marker.store(true, Ordering::SeqCst); } @@ -8641,6 +8762,92 @@ mod tests { assert!(err.message.contains("photos/2026/old-destination.jpg"), "{}", err.message); } + #[tokio::test] + async fn zero_reference_proof_blocks_persisted_journal_transaction_and_free_version_references() { + let current = build_rustfs_tier("COLD-A"); + let current_identity = tier_backend_identity(¤t).expect("current identity should encode"); + let replacement = build_azure_tier("account-a"); + let replacement_identity = tier_backend_identity(&replacement).expect("replacement identity should encode"); + let target = TierMutationIntentTarget { + tier_name: "COLD-A".to_string(), + old_backend_identity: Some(current_identity), + new_backend_identity: Some(replacement_identity), + }; + + let journal_store = Arc::new(CasConfigStore::default()); + let journal = crate::bucket::lifecycle::tier_sweeper::Jentry { + obj_name: "remote/journal-object".to_string(), + version_id: "v1".to_string(), + tier_name: "COLD-A".to_string(), + backend_identity: Some(current_identity), + version_id_exact: false, + }; + journal_store + .insert_config_object( + crate::bucket::lifecycle::tier_delete_journal::tier_delete_journal_object_name(&journal), + crate::bucket::lifecycle::tier_delete_journal::encode_tier_delete_journal_entry(&journal) + .expect("journal record should encode"), + ) + .await; + let err = ensure_no_authoritative_tier_object_references(journal_store, std::slice::from_ref(&target)) + .await + .expect_err("unfinished delete journal for the old backend must block rebind"); + assert_eq!(err.code, ERR_TIER_BACKEND_IN_USE.code); + assert!(err.message.contains(TIER_DELETE_JOURNAL_PREFIX), "{}", err.message); + + let transaction_store = Arc::new(CasConfigStore::default()); + let transaction = crate::bucket::lifecycle::transition_transaction::TransitionTransaction::new( + crate::bucket::lifecycle::transition_transaction::TransitionTransactionInit { + deployment_id: uuid::Uuid::new_v4(), + transaction_id: uuid::Uuid::new_v4(), + owner_epoch: uuid::Uuid::new_v4(), + write_id: uuid::Uuid::new_v4(), + source: crate::bucket::lifecycle::transition_transaction::TransitionSourceIdentity { + bucket: "photos".to_string(), + object: "2026/transaction.jpg".to_string(), + version_id: None, + data_dir: uuid::Uuid::new_v4(), + mod_time_unix_nanos: 1, + size: 1, + etag: "etag".to_string(), + version_mode: crate::bucket::lifecycle::transition_transaction::TransitionSourceVersionMode::Unversioned, + }, + tier_name: "COLD-A".to_string(), + backend_fingerprint: current_identity, + not_after_unix_nanos: 2, + }, + ) + .expect("transaction record should initialize"); + transaction_store + .insert_config_object( + crate::bucket::lifecycle::transition_transaction::transition_transaction_record_object_name( + transaction.transaction_id, + ) + .expect("transaction record path should build"), + transaction.encode().expect("transaction record should encode"), + ) + .await; + let err = ensure_no_authoritative_tier_object_references(transaction_store, std::slice::from_ref(&target)) + .await + .expect_err("unfinished transition transaction for the old backend must block rebind"); + assert_eq!(err.code, ERR_TIER_BACKEND_IN_USE.code); + assert!(err.message.contains(TRANSITION_TRANSACTION_RECORD_PREFIX), "{}", err.message); + + let free_version_store = Arc::new(CasConfigStore::default()); + let mut free_version = transitioned_tier_object("photos", "2026/free-version.jpg", "COLD-A", Some(current_identity)); + free_version.transitioned_object.status = "pending".to_string(); + free_version.transitioned_object.free_version = true; + ensure_no_authoritative_tier_object_references(free_version_store.clone(), std::slice::from_ref(&target)) + .await + .expect("empty free-version fixture should permit rebind"); + free_version_store.add_listed_version(free_version); + let err = ensure_no_authoritative_tier_object_references(free_version_store, &[target]) + .await + .expect_err("recoverable free version for the old backend must block rebind"); + assert_eq!(err.code, ERR_TIER_BACKEND_IN_USE.code); + assert!(err.message.contains("photos/2026/free-version.jpg"), "{}", err.message); + } + #[tokio::test] async fn zero_reference_proof_fails_closed_when_version_listing_marker_is_missing() { let current = build_rustfs_tier("COLD-A");