From 36fd4c9fc2bb918178615c64bf5c4b527183d600 Mon Sep 17 00:00:00 2001 From: cxymds Date: Sun, 6 Sep 2026 23:27:32 +0800 Subject: [PATCH] feat(ilm): retry retained transition recovery (#7308) --- crates/ecstore/src/api/mod.rs | 5 +- .../src/bucket/lifecycle/recovery_control.rs | 107 ++++++++ .../lifecycle/transition_transaction.rs | 231 +++++++++++++++++- crates/ecstore/src/store/init.rs | 80 +++++- .../ilm-tiering-persistence-contracts.md | 4 +- docs/operations/tier-ilm-debugging.md | 35 ++- rustfs/src/admin/handlers/ilm_transition.rs | 206 +++++++++++++++- rustfs/src/admin/storage_api.rs | 6 +- 8 files changed, 652 insertions(+), 22 deletions(-) diff --git a/crates/ecstore/src/api/mod.rs b/crates/ecstore/src/api/mod.rs index 450e585aa..c79450396 100644 --- a/crates/ecstore/src/api/mod.rs +++ b/crates/ecstore/src/api/mod.rs @@ -93,8 +93,9 @@ pub mod bucket { pub mod transition_transaction { pub use crate::bucket::lifecycle::transition_transaction::{ TransitionOperatorDeleteResult, TransitionOperatorError, TransitionOperatorProbe, TransitionOperatorStatus, - delete_transition_candidate_for_operator, finalize_missing_transition_transaction_for_operator, - inspect_transition_transaction_for_operator, + TransitionRecoveryRetryResult, TransitionRecoveryRetryStatus, delete_transition_candidate_for_operator, + finalize_missing_transition_transaction_for_operator, inspect_transition_recovery_retry_for_operator, + inspect_transition_transaction_for_operator, retry_transition_recovery_for_operator, }; #[cfg(feature = "test-util")] pub use crate::bucket::lifecycle::transition_transaction::{ diff --git a/crates/ecstore/src/bucket/lifecycle/recovery_control.rs b/crates/ecstore/src/bucket/lifecycle/recovery_control.rs index f773d497c..4328f179b 100644 --- a/crates/ecstore/src/bucket/lifecycle/recovery_control.rs +++ b/crates/ecstore/src/bucket/lifecycle/recovery_control.rs @@ -499,6 +499,26 @@ impl IlmRecoveryControl { self.validate() } + pub fn retry_for_operator(&mut self, expected_source_generation: &IlmRecoverySourceGeneration) -> Result<()> { + if self.owner.is_some() + || !matches!( + self.classification, + IlmRecoveryClassification::RetainedAmbiguous | IlmRecoveryClassification::OperatorRequired + ) + || self.attempt_count == u64::MAX + || &self.observed_source_generation != expected_source_generation + { + return Err(IlmRecoveryControlError::InvalidSuccessor( + "operator retry requires the exact ownerless retained source generation", + )); + } + self.bump_revision()?; + self.classification = IlmRecoveryClassification::Retrying; + self.consecutive_failure_count = 0; + self.next_attempt_at_unix_nanos = None; + self.validate() + } + pub fn validate_successor(&self, next: &Self) -> Result<()> { self.validate()?; next.validate()?; @@ -528,6 +548,14 @@ impl IlmRecoveryControl { { self.validate_operator_abandon_successor(next) } + (None, None) + if matches!( + self.classification, + IlmRecoveryClassification::RetainedAmbiguous | IlmRecoveryClassification::OperatorRequired + ) && next.classification == IlmRecoveryClassification::Retrying => + { + self.validate_operator_retry_successor(next) + } (None, None) => Err(IlmRecoveryControlError::InvalidSuccessor( "ownerless control cannot advance without a claim", )), @@ -550,6 +578,22 @@ impl IlmRecoveryControl { Ok(()) } + fn validate_operator_retry_successor(&self, next: &Self) -> Result<()> { + if next.observed_source_generation != self.observed_source_generation + || next.attempt_count != self.attempt_count + || next.consecutive_failure_count != 0 + || next.first_failure_at_unix_nanos != self.first_failure_at_unix_nanos + || next.last_failure_at_unix_nanos != self.last_failure_at_unix_nanos + || next.next_attempt_at_unix_nanos.is_some() + || next.last_error_code != self.last_error_code + { + return Err(IlmRecoveryControlError::InvalidSuccessor( + "operator retry changed recovery history or source generation", + )); + } + Ok(()) + } + fn validate_claim_successor(&self, next: &Self) -> Result<()> { if self.classification != IlmRecoveryClassification::Retrying || next.classification != IlmRecoveryClassification::Retrying @@ -1347,6 +1391,69 @@ mod tests { assert!(previous.validate_successor(&mutated_history).is_err()); } + #[test] + fn operator_retry_rearms_exact_retained_generation_without_resetting_history() { + for classification in [ + IlmRecoveryClassification::RetainedAmbiguous, + IlmRecoveryClassification::OperatorRequired, + ] { + let mut retained = control(); + retained + .claim("node-a", Uuid::new_v4(), 2_000_000_000, 1) + .expect("attempt should claim"); + retained + .record_retryable_failure(2_000_000_001, IlmRecoveryErrorCode::BackendTimeout) + .expect("failure should persist"); + retained.classification = classification; + retained.next_attempt_at_unix_nanos = None; + if classification == IlmRecoveryClassification::OperatorRequired { + retained.attempt_count = u64::from(MAX_RECOVERY_ATTEMPTS); + retained.consecutive_failure_count = MAX_RECOVERY_ATTEMPTS; + } + retained.validate().expect("retained control should remain valid"); + + let previous = retained.clone(); + retained + .retry_for_operator(&previous.observed_source_generation) + .expect("exact retained generation should be retryable"); + previous + .validate_successor(&retained) + .expect("operator retry should be a valid successor"); + assert_eq!(retained.classification, IlmRecoveryClassification::Retrying); + assert_eq!(retained.revision, previous.revision + 1); + assert_eq!(retained.attempt_count, previous.attempt_count); + assert_eq!(retained.first_failure_at_unix_nanos, previous.first_failure_at_unix_nanos); + assert_eq!(retained.last_failure_at_unix_nanos, previous.last_failure_at_unix_nanos); + assert_eq!(retained.last_error_code, previous.last_error_code); + assert_eq!(retained.consecutive_failure_count, 0); + assert_eq!(retained.next_attempt_at_unix_nanos, None); + assert!(retained.should_attempt_at(2_000_000_002)); + + if classification == IlmRecoveryClassification::OperatorRequired { + retained + .claim("node-b", Uuid::new_v4(), 2_000_000_002, 1) + .expect("operator retry should authorize one new bounded attempt"); + retained + .record_retryable_failure(2_000_000_003, IlmRecoveryErrorCode::BackendTimeout) + .expect("the bounded attempt failure should persist"); + assert_eq!(retained.classification, IlmRecoveryClassification::OperatorRequired); + assert_eq!(retained.attempt_count, u64::from(MAX_RECOVERY_ATTEMPTS) + 1); + assert_eq!(retained.consecutive_failure_count, 1); + } + + let mut wrong_generation = previous; + let mut changed_generation = wrong_generation.observed_source_generation.clone(); + changed_generation.source_etag = "changed".to_string(); + assert!(wrong_generation.retry_for_operator(&changed_generation).is_err()); + } + + let mut exhausted = control(); + exhausted.classification = IlmRecoveryClassification::OperatorRequired; + exhausted.attempt_count = u64::MAX; + let generation = exhausted.observed_source_generation.clone(); + assert!(exhausted.retry_for_operator(&generation).is_err()); + } + #[test] fn recovery_control_view_redacts_source_and_owner_details() { let mut control = control(); diff --git a/crates/ecstore/src/bucket/lifecycle/transition_transaction.rs b/crates/ecstore/src/bucket/lifecycle/transition_transaction.rs index 426956d13..a950a6a5b 100644 --- a/crates/ecstore/src/bucket/lifecycle/transition_transaction.rs +++ b/crates/ecstore/src/bucket/lifecycle/transition_transaction.rs @@ -34,7 +34,7 @@ use crate::bucket::lifecycle::tier_sweeper::{ }; use crate::disk::RUSTFS_META_BUCKET; use crate::error::{Error, Result as EcstoreResult}; -use crate::object_api::ObjectOptions; +use crate::object_api::{ObjectInfo, ObjectOptions}; use crate::services::tier::{tier::TierConfigMgr, warm_backend::TransitionCandidateProbe}; use crate::storage_api_contracts::{ list::ListOperations as _, @@ -954,6 +954,10 @@ pub enum TransitionOperatorError { expected: String, actual: TransitionOperatorProbe, }, + #[error("transition recovery control is stale")] + StaleRecoveryControl, + #[error("transition recovery control is not eligible for operator retry")] + RetryNotAllowed, #[error("transition transaction store failed: {0}")] Store(#[source] Error), #[error("remote tier reconciliation failed: {0}")] @@ -962,6 +966,179 @@ pub enum TransitionOperatorError { type TransitionOperatorResult = std::result::Result; +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +pub struct TransitionRecoveryRetryStatus { + pub control_id: String, + pub transaction_id: Uuid, + pub state: TransitionTransactionState, + pub classification: IlmRecoveryClassification, + pub control_revision: u64, + pub attempt_count: u64, + pub consecutive_failure_count: u32, + pub last_error_code: IlmRecoveryErrorCode, + pub source_generation_sha256: String, + pub copy_set_sha256: String, + pub retry_ready: bool, + #[serde(skip_serializing_if = "Option::is_none")] + pub retry_not_ready_reason: Option<&'static str>, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +pub struct TransitionRecoveryRetryResult { + pub control_id: String, + pub transaction_id: Uuid, + pub previous_revision: u64, + pub revision: u64, + pub classification: IlmRecoveryClassification, + pub attempt_count: u64, + pub source_generation_sha256: String, +} + +struct TransitionRecoveryRetryContext { + observed: ObservedIlmRecoveryControl, + transaction: TransitionTransaction, + source_generation_sha256: String, +} + +fn transition_recovery_retry_readiness(control: &IlmRecoveryControl) -> (bool, Option<&'static str>) { + if control.owner.is_some() { + return (false, Some("attempt_owned")); + } + match control.classification { + IlmRecoveryClassification::RetainedAmbiguous | IlmRecoveryClassification::OperatorRequired => (true, None), + IlmRecoveryClassification::Retrying => (false, Some("already_retrying")), + IlmRecoveryClassification::Corrupt => (false, Some("source_corrupt")), + IlmRecoveryClassification::Abandoned => (false, Some("source_abandoned")), + IlmRecoveryClassification::Terminal => (false, Some("source_terminal")), + } +} + +async fn load_transition_recovery_retry_context( + api: Arc, + control_id: &str, +) -> TransitionOperatorResult { + let observed = match load_recovery_control(api.clone(), IlmRecoveryProtocol::TransitionTransaction, control_id).await { + Ok(observed) => observed, + Err(Error::ConfigNotFound) => return Err(TransitionOperatorError::NotFound), + Err(err) => return Err(TransitionOperatorError::Store(err)), + }; + let transaction_id = Uuid::parse_str(&observed.control.identity.stable_operation_identity) + .ok() + .filter(|transaction_id| !transaction_id.is_nil()) + .ok_or(TransitionOperatorError::StaleRecoveryControl)?; + let canonical_path = transition_transaction_record_object_name(transaction_id) + .map_err(|err| TransitionOperatorError::Store(Error::other(err)))?; + if observed.control.identity.canonical_source_path != canonical_path + || observed.control.identity.record_class != "transition_transaction_v1" + { + return Err(TransitionOperatorError::StaleRecoveryControl); + } + let transaction = match load_transition_transaction_record(api.clone(), transaction_id).await { + Ok(transaction) => transaction, + Err(Error::ConfigNotFound) => return Err(TransitionOperatorError::NotFound), + Err(err) => return Err(TransitionOperatorError::Store(err)), + }; + let source = observe_recovery_source(api, &canonical_path, TRANSITION_TRANSACTION_SCHEMA) + .await + .map_err(TransitionOperatorError::Store)?; + let exact_source = source.is_consistent() + && source.generation == observed.control.observed_source_generation + && source + .canonical_data + .as_deref() + .is_some_and(|data| TransitionTransaction::decode(transaction_id, data).is_ok_and(|decoded| decoded == transaction)); + if !exact_source { + return Err(TransitionOperatorError::StaleRecoveryControl); + } + let generation = serde_json::to_vec(&observed.control.observed_source_generation) + .map_err(|err| TransitionOperatorError::Store(Error::other(err)))?; + Ok(TransitionRecoveryRetryContext { + observed, + transaction, + source_generation_sha256: hex_sha256(&generation, ToOwned::to_owned), + }) +} + +pub async fn inspect_transition_recovery_retry_for_operator( + api: Arc, + control_id: &str, +) -> TransitionOperatorResult { + let context = load_transition_recovery_retry_context(api, control_id).await?; + let (retry_ready, retry_not_ready_reason) = transition_recovery_retry_readiness(&context.observed.control); + Ok(TransitionRecoveryRetryStatus { + control_id: control_id.to_string(), + transaction_id: context.transaction.transaction_id, + state: context.transaction.state, + classification: context.observed.control.classification, + control_revision: context.observed.control.revision, + attempt_count: context.observed.control.attempt_count, + consecutive_failure_count: context.observed.control.consecutive_failure_count, + last_error_code: context.observed.control.last_error_code, + source_generation_sha256: context.source_generation_sha256, + copy_set_sha256: context.observed.control.observed_source_generation.copy_set_sha256.clone(), + retry_ready, + retry_not_ready_reason, + }) +} + +pub async fn retry_transition_recovery_for_operator( + api: Arc, + control_id: &str, + expected_control_revision: u64, + expected_source_generation_sha256: &str, +) -> TransitionOperatorResult { + let control_object = recovery_control_record_object_name(IlmRecoveryProtocol::TransitionTransaction, control_id) + .map_err(|err| TransitionOperatorError::Store(Error::other(err)))?; + let retry_lock = api + .new_ns_lock(RUSTFS_META_BUCKET, &format!("{control_object}.recovery-lock")) + .await + .map_err(TransitionOperatorError::Store)?; + let retry_guard = retry_lock + .get_write_lock(crate::set_disk::get_lock_acquire_timeout()) + .await + .map_err(|err| TransitionOperatorError::Store(Error::other(err)))?; + let context = load_transition_recovery_retry_context(api.clone(), control_id).await?; + let (retry_ready, _) = transition_recovery_retry_readiness(&context.observed.control); + if !retry_ready { + return Err(TransitionOperatorError::RetryNotAllowed); + } + if retry_guard.is_lock_lost() + || expected_control_revision == 0 + || context.observed.control.revision != expected_control_revision + || context.source_generation_sha256 != expected_source_generation_sha256 + { + return Err(TransitionOperatorError::StaleRecoveryControl); + } + let previous_revision = context.observed.control.revision; + let mut next = context.observed.control.clone(); + next.retry_for_operator(&context.observed.control.observed_source_generation) + .map_err(|_| TransitionOperatorError::RetryNotAllowed)?; + if retry_guard.is_lock_lost() { + return Err(TransitionOperatorError::StaleRecoveryControl); + } + save_recovery_control_if_current(api.clone(), &context.observed, &next) + .await + .map_err(|err| match err { + Error::PreconditionFailed => TransitionOperatorError::StaleRecoveryControl, + err => TransitionOperatorError::Store(err), + })?; + let persisted = load_recovery_control(api, IlmRecoveryProtocol::TransitionTransaction, control_id) + .await + .map_err(TransitionOperatorError::Store)?; + if retry_guard.is_lock_lost() || persisted.control != next { + return Err(TransitionOperatorError::StaleRecoveryControl); + } + Ok(TransitionRecoveryRetryResult { + control_id: control_id.to_string(), + transaction_id: context.transaction.transaction_id, + previous_revision, + revision: persisted.control.revision, + classification: persisted.control.classification, + attempt_count: persisted.control.attempt_count, + source_generation_sha256: context.source_generation_sha256, + }) +} + fn validate_operator_reconcile_transaction( transaction: &TransitionTransaction, now_unix_nanos: i128, @@ -1706,12 +1883,27 @@ async fn local_commit_matches_transaction(api: Arc, transaction: &Trans .get_object_info(&transaction.source.bucket, &transaction.source.object, &opts) .await?; let transitioned = &object.transitioned_object; - Ok(transitioned.status == TRANSITION_COMPLETE + Ok(local_object_matches_transition_source(&object, &transaction.source) + && transitioned.status == TRANSITION_COMPLETE && transitioned.name == transaction.remote_object && transitioned.tier == transaction.tier_name && transitioned.version_id == transaction.remote_version.tier_delete_version_id().unwrap_or_default()) } +fn local_object_matches_transition_source(object: &ObjectInfo, source: &TransitionSourceIdentity) -> bool { + let observed_version_id = object.version_id.filter(|version_id| !version_id.is_nil()); + let observed_mod_time = object + .mod_time + .and_then(|mod_time| i64::try_from(mod_time.unix_timestamp_nanos()).ok()); + object.bucket == source.bucket + && object.name == source.object + && observed_version_id == source.version_id + && object.data_dir == Some(source.data_dir) + && observed_mod_time == Some(source.mod_time_unix_nanos) + && object.size == source.size + && object.etag.as_deref() == Some(source.etag.as_str()) +} + fn transition_source_lookup_options(transaction: &TransitionTransaction) -> ObjectOptions { ObjectOptions { version_id: match transaction.source.version_mode { @@ -2183,6 +2375,41 @@ mod tests { } } + #[test] + fn local_commit_proof_requires_the_complete_source_identity() { + let source = source_identity(TransitionSourceVersionMode::Versioned); + let exact = ObjectInfo { + bucket: source.bucket.clone(), + name: source.object.clone(), + version_id: source.version_id, + data_dir: Some(source.data_dir), + mod_time: Some( + time::OffsetDateTime::from_unix_timestamp_nanos(i128::from(source.mod_time_unix_nanos)) + .expect("source timestamp should be valid"), + ), + size: source.size, + etag: Some(source.etag.clone()), + ..Default::default() + }; + assert!(local_object_matches_transition_source(&exact, &source)); + + let mut changed = exact.clone(); + changed.version_id = Some(Uuid::new_v4()); + assert!(!local_object_matches_transition_source(&changed, &source)); + changed = exact.clone(); + changed.data_dir = Some(Uuid::new_v4()); + assert!(!local_object_matches_transition_source(&changed, &source)); + changed = exact.clone(); + changed.mod_time = changed.mod_time.map(|value| value + Duration::from_nanos(1)); + assert!(!local_object_matches_transition_source(&changed, &source)); + changed = exact.clone(); + changed.size += 1; + assert!(!local_object_matches_transition_source(&changed, &source)); + changed = exact; + changed.etag = Some("different-etag".to_string()); + assert!(!local_object_matches_transition_source(&changed, &source)); + } + fn cleanup_proof(transaction: &TransitionTransaction, decision: TransitionCleanupDecision) -> TransitionCleanupProof { TransitionCleanupProof { transaction_id: transaction.transaction_id, diff --git a/crates/ecstore/src/store/init.rs b/crates/ecstore/src/store/init.rs index 5c0e16a66..a41093a07 100644 --- a/crates/ecstore/src/store/init.rs +++ b/crates/ecstore/src/store/init.rs @@ -862,9 +862,10 @@ mod tests { TransitionOperatorProbe, TransitionRecoveryClaimBarrier, TransitionRecoveryTerminalBarrier, TransitionRemoteVersion, TransitionSourceIdentity, TransitionSourceVersionMode, TransitionTransaction, TransitionTransactionInit, TransitionTransactionState, delete_transition_candidate_for_operator, - finalize_missing_transition_transaction_for_operator, inspect_transition_transaction_for_operator, - load_transition_transaction_record, recover_transition_transaction_records, - recover_transition_transaction_records_at, save_transition_transaction_record, + finalize_missing_transition_transaction_for_operator, inspect_transition_recovery_retry_for_operator, + inspect_transition_transaction_for_operator, load_transition_transaction_record, + recover_transition_transaction_records, recover_transition_transaction_records_at, + retry_transition_recovery_for_operator, save_transition_transaction_record, save_transition_transaction_record_if_current, transition_recovery_control_id, transition_transaction_record_object_name, }, @@ -20629,7 +20630,7 @@ mod tests { bucket: bucket.to_string(), object: object.to_string(), version_id: None, - data_dir: uuid::Uuid::new_v4(), + data_dir: original.data_dir.expect("source object should have data_dir"), mod_time_unix_nanos: original .mod_time .expect("source object should have mod_time") @@ -20763,7 +20764,7 @@ mod tests { bucket: bucket.to_string(), object: object.to_string(), version_id: None, - data_dir: uuid::Uuid::new_v4(), + data_dir: original.data_dir.expect("source object should have data_dir"), mod_time_unix_nanos: original .mod_time .expect("source object should have mod_time") @@ -20904,10 +20905,77 @@ mod tests { IlmRecoveryClassification::RetainedAmbiguous ); let local_commit_control = - load_recovery_control(store, IlmRecoveryProtocol::TransitionTransaction, &local_commit_control_id) + load_recovery_control(store.clone(), IlmRecoveryProtocol::TransitionTransaction, &local_commit_control_id) .await .expect("local-commit control should persist"); assert_eq!(local_commit_control.control.classification, IlmRecoveryClassification::OperatorRequired); + + let upload_status = inspect_transition_recovery_retry_for_operator(store.clone(), &upload_started_control_id) + .await + .expect("retained upload should be inspectable for a bounded retry"); + let local_status = inspect_transition_recovery_retry_for_operator(store.clone(), &local_commit_control_id) + .await + .expect("operator-required local commit should be inspectable for a bounded retry"); + assert!(upload_status.retry_ready); + assert!(local_status.retry_ready); + assert!(matches!( + retry_transition_recovery_for_operator( + store.clone(), + &upload_started_control_id, + upload_status.control_revision + 1, + &upload_status.source_generation_sha256, + ) + .await, + Err(TransitionOperatorError::StaleRecoveryControl) + )); + + let put_count_before_retry = backend.put_count().await; + let get_count_before_retry = backend.get_count().await; + let remove_count_before_retry = backend.remove_count().await; + let upload_retry = retry_transition_recovery_for_operator( + store.clone(), + &upload_started_control_id, + upload_status.control_revision, + &upload_status.source_generation_sha256, + ) + .await + .expect("exact retained upload generation should be rearmed"); + let local_retry = retry_transition_recovery_for_operator( + store.clone(), + &local_commit_control_id, + local_status.control_revision, + &local_status.source_generation_sha256, + ) + .await + .expect("exact operator-required local commit generation should be rearmed"); + assert_eq!(upload_retry.classification, IlmRecoveryClassification::Retrying); + assert_eq!(local_retry.classification, IlmRecoveryClassification::Retrying); + assert_eq!(upload_retry.attempt_count, upload_status.attempt_count); + assert_eq!(local_retry.attempt_count, local_status.attempt_count); + assert_eq!(backend.put_count().await, put_count_before_retry); + assert_eq!(backend.get_count().await, get_count_before_retry); + assert_eq!(backend.remove_count().await, remove_count_before_retry); + assert_eq!(backend.exact_remove_count(), 0, "operator retry must not directly issue remote DELETE"); + + let retried = recover_transition_transaction_records(store.clone(), 100, None) + .await + .expect("rearmed records should be re-evaluated through normal recovery"); + assert_eq!((retried.scanned, retried.recovered, retried.retained, retried.failed), (2, 0, 2, 0)); + let upload_retained = + load_recovery_control(store.clone(), IlmRecoveryProtocol::TransitionTransaction, &upload_started_control_id) + .await + .expect("upload retry result should persist"); + let local_retained = load_recovery_control(store, IlmRecoveryProtocol::TransitionTransaction, &local_commit_control_id) + .await + .expect("local commit retry result should persist"); + assert_eq!(upload_retained.control.classification, IlmRecoveryClassification::RetainedAmbiguous); + assert_eq!(local_retained.control.classification, IlmRecoveryClassification::OperatorRequired); + assert_eq!(upload_retained.control.attempt_count, upload_status.attempt_count + 1); + assert_eq!(local_retained.control.attempt_count, local_status.attempt_count + 1); + assert_eq!(backend.put_count().await, put_count_before_retry); + assert_eq!(backend.get_count().await, get_count_before_retry); + assert_eq!(backend.remove_count().await, remove_count_before_retry); + assert_eq!(backend.exact_remove_count(), 0); } #[cfg(feature = "test-util")] diff --git a/docs/architecture/ilm-tiering-persistence-contracts.md b/docs/architecture/ilm-tiering-persistence-contracts.md index 4234edd9a..0becd45af 100644 --- a/docs/architecture/ilm-tiering-persistence-contracts.md +++ b/docs/architecture/ilm-tiering-persistence-contracts.md @@ -28,7 +28,7 @@ These are approved-target invariants. A protocol's explicitly labeled current ex | Phase | Authoritative owner | May issue remote DELETE? | Ownership transfer evidence | |---|---|---:|---| | Remote PUT is in flight or its response is unknown | Transition transaction | Only cleanup of its own canonical candidate, subject to the transaction recovery predicate | Durable transaction identity plus a known remote-version state; the approved target also requires expiry and durable takeover of the creator fence | -| Local transition commit is complete | Exact transitioned version in `xl.meta` | No | Current recovery finds the transaction's logical bucket/object/version and checks `TRANSITION_COMPLETE` plus the same remote object, tier, and remote version. It does not compare the recorded data directory, modification time, size, or ETag; the approved target adds that full source comparison | +| Local transition commit is complete | Exact transitioned version in `xl.meta` | No | Recovery finds the transaction's logical bucket/object/version and requires the complete recorded source identity (version ID, data directory, modification time, size, and ETag), `TRANSITION_COMPLETE`, and the same remote object, tier, and remote version before removing only the transaction record | | An ordinary delete removes that transitioned version | Hidden `xl.meta` free-version | Yes | Metadata quorum atomically removes the visible version and preserves its exact tier tuple in the free-version | | A recursive prefix/delete-all operation cannot preserve per-object markers | v6 journal bound to an immutable single dispatch manifest or a chunk-parent-bound child manifest | Yes, but only after child/manifest completion and all-pool absence proof | `DispatchAuthorized`, exact local destructive mutation, every journal `Committed`, then child/manifest `Completed`; a chunk parent advances only after that child completion | | Tier configuration mutation, manual job, or decommission receipt | Intent/admission/copy proof only | No | These records gate configuration, scheduling, or migration; they never become remote-object cleanup owners | @@ -134,7 +134,7 @@ The creator owns the canonical remote candidate until local metadata commits the | `UploadOutcomeUnknown`; probe returns `VersionedPresent` whose identifier is a nil UUID | Transaction recovery retains ownership evidence | Retain | A nil identifier is invalid exact-version evidence and never becomes unversioned or remote-delete authority | | `UploadOutcomeUnknown`; probe ambiguous, unsupported, or errors | Transaction recovery retains ownership evidence | Retain | No destructive action; operator reconcile may inspect after expiry | | `Uploaded` | Originating transition attempt until expiry; after expiry, the worker that wins `Uploaded -> CleanupPending` by exact ETag CAS | Retain while active; after expiry, persist `CleanupPending`, then recheck and delete the unreferenced candidate or record | Current CAS fences the predecessor, but approved v2 also requires a durable recovery lease, full all-pool source/free-version proof, and before/after fence checks | -| `LocalCommitStarted`; logical source lookup returns `TRANSITION_COMPLETE` with the same remote object, tier, and remote version | Transition committer until ownership transfers to `xl.meta` | Delete transaction record | Current recovery treats this tuple as ownership transfer. The approved target additionally compares recorded source version ID, data directory, modification time, size, and ETag before conditional terminal cleanup | +| `LocalCommitStarted`; logical source lookup returns the complete recorded source identity and `TRANSITION_COMPLETE` with the same remote object, tier, and remote version | Transition committer until ownership transfers to `xl.meta` | Delete transaction record | Recovery treats only the full source and remote tuple as ownership transfer; a mismatch remains operator-required and cannot authorize remote deletion | | `LocalCommitStarted`; logical source is missing, its transition tuple differs, or the read is uncertain | Transaction record/recovery | Retain | No remote delete without a separate durable cleanup proof | | `CleanupPending`; logical source lookup returns the same current transition predicate | `xl.meta` is remote reachability owner; recovery owns only record cleanup | Delete transaction record | `xl.meta` is owner; do not delete remote. The approved target adds the full recorded source comparison | | `CleanupPending`; logical source is absent or its transition tuple differs | Transaction recovery | Delete exact candidate, then record | Cleanup proof, known version state, exact backend lease, durable owner fence, and before/after identity checks | diff --git a/docs/operations/tier-ilm-debugging.md b/docs/operations/tier-ilm-debugging.md index d5da40af3..52e6c680b 100644 --- a/docs/operations/tier-ilm-debugging.md +++ b/docs/operations/tier-ilm-debugging.md @@ -173,7 +173,7 @@ Historical transition transactions in `upload_outcome_unknown` state can use an ## Inspect and disposition retained recovery records -This section describes an **approved target that is not implemented yet**. Current servers do not expose the routes below and continue to quarantine tier-delete journal v1/v2 records. Do not remove internal metadata objects by hand: that loses ETag, all-pool, decommission, export, and audit guarantees. +Current servers expose the routes below for retained recovery controls. Do not remove internal metadata objects by hand: that loses ETag, all-pool, decommission, export, and audit guarantees. The approved read-only inventory is bounded and paginated: @@ -223,6 +223,39 @@ Malformed/unsupported records and journal v3-v6 cannot use abandon. Known-versio Automatic retry state survives restart. Retryable transport/quorum failures use a 60-second exponential base capped at one hour and a deterministic 80-to-100-percent multiplier, so jitter never increases the capped delay. After 32 consecutive failures or seven days from the first persisted failure, automatic work stops at `operator_required`. Unsupported or ambiguous evidence goes directly to `retained_ambiguous`/`operator_required`; age alone never deletes it. Resolved controls, immutable exports, and completed disposition receipts have minimum 30-day, 90-day, and 365-day retention respectively, and are collected only after exact source absence, decommission, successor, and audit checks. +### Retry a retained transition transaction + +For a `transition_transaction` control, inspect returns an additional `transition_retry` object when the exact transaction source and recovery-control generation are still consistent. It contains `retry_ready`, `control_revision`, `source_generation_sha256`, the current classification and counters, and a bounded refusal reason. A missing `transition_retry` with `transition_retry_not_ready_reason=source_or_control_not_ready` means the server could not reconstruct exact live evidence; do not retry from an older response. + +First perform a dry-run with the exact revision and source-generation digest returned by the latest inspect: + +```json +POST /rustfs/admin/v3/ilm/recovery/records/ +{ + "action": "retry_transition_recovery", + "mode": "dry_run", + "expected_control_revision": 7, + "expected_source_generation_sha256": "" +} +``` + +After repairing the reported storage, tier, or capability problem, repeat inspect and dry-run, then execute with the newly observed values: + +```json +POST /rustfs/admin/v3/ilm/recovery/records/ +{ + "action": "retry_transition_recovery", + "mode": "execute", + "expected_control_revision": 7, + "expected_source_generation_sha256": "", + "confirm": true +} +``` + +Execution performs one ETag-CAS update of the exact ownerless `retained_ambiguous` or `operator_required` control to `retrying`. It preserves the lifetime attempt count and failure history, clears only the consecutive-failure backoff, and does not mutate the transaction source or issue a tier PUT, GET, probe, or DELETE. The normal recovery worker then acquires a fresh bounded owner lease and repeats every source and remote proof before any side effect. + +A historical v1 `UploadStarted` record can return to `retained_ambiguous` because its bytes do not prove whether PUT reached the provider. `LocalCommitStarted` becomes terminal only when the local object still matches the recorded version ID, data directory, modification time, size, ETag, and exact transitioned remote tuple; otherwise it returns to `operator_required`. Retrying is therefore a bounded re-evaluation after an underlying repair, not an override of missing evidence. + The full schema, lease, mixed-version, retry, privacy, and metric requirements are in [../architecture/ilm-tiering-persistence-contracts.md](../architecture/ilm-tiering-persistence-contracts.md#bounded-recovery-control-and-operator-disposition). ## Reconcile legacy transition-version metadata diff --git a/rustfs/src/admin/handlers/ilm_transition.rs b/rustfs/src/admin/handlers/ilm_transition.rs index eceb763a2..313e1e22a 100644 --- a/rustfs/src/admin/handlers/ilm_transition.rs +++ b/rustfs/src/admin/handlers/ilm_transition.rs @@ -22,15 +22,17 @@ use crate::admin::storage_api::lifecycle::{ IlmRecoveryDispositionState, IlmRecoveryExportObservation, IlmRecoveryProtocol, ManualTransitionCancelCheck, ManualTransitionJobRecord, ManualTransitionJobState, ManualTransitionProgressSink, ManualTransitionQueueSnapshot, ManualTransitionRunOptions, ManualTransitionRunReport, ManualTransitionScopeAdmission, ManualTransitionScopeAdmissionClaim, - TransitionOperatorDeleteResult, TransitionOperatorError, claim_manual_transition_scope_admission, create_recovery_export, - delete_manual_transition_scope_admission_if_current, delete_transition_candidate_for_operator, dry_run_recovery_disposition, - enqueue_transition_for_existing_objects_scoped, execute_recovery_disposition, - finalize_missing_transition_transaction_for_operator, inspect_recovery_control, inspect_recovery_export_observation, + TransitionOperatorDeleteResult, TransitionOperatorError, TransitionRecoveryRetryResult, TransitionRecoveryRetryStatus, + claim_manual_transition_scope_admission, create_recovery_export, delete_manual_transition_scope_admission_if_current, + delete_transition_candidate_for_operator, dry_run_recovery_disposition, enqueue_transition_for_existing_objects_scoped, + execute_recovery_disposition, finalize_missing_transition_transaction_for_operator, inspect_recovery_control, + inspect_recovery_export_observation, inspect_transition_recovery_retry_for_operator, inspect_transition_transaction_for_operator, list_recovery_controls, load_manual_transition_job_record, load_manual_transition_scope_admission, load_recovery_export, manual_transition_job_lease_expired, manual_transition_queue_snapshot, manual_transition_scope_admission_lease_expired, persist_manual_transition_job_progress_if_owned, renew_manual_transition_job_lease_if_owned, - request_manual_transition_job_cancel, save_manual_transition_job_record, update_manual_transition_job_record, + request_manual_transition_job_cancel, retry_transition_recovery_for_operator, save_manual_transition_job_record, + update_manual_transition_job_record, }; use crate::admin::storage_api::runtime::ECStore; use crate::admin::storage_api::s3::{S3ErrorCode as AdminS3ErrorCode, error as admin_s3_error}; @@ -615,6 +617,10 @@ struct IlmRecoveryControlInspectResponse { disposition_dry_run_receipt: Option, #[serde(skip_serializing_if = "Option::is_none")] disposition_dry_run_receipt_expires_at_unix_nanos: Option, + #[serde(skip_serializing_if = "Option::is_none")] + transition_retry: Option, + #[serde(skip_serializing_if = "Option::is_none")] + transition_retry_not_ready_reason: Option<&'static str>, } #[derive(Debug, Deserialize)] @@ -634,6 +640,13 @@ enum IlmRecoveryRecordMutationRequest { #[serde(default)] acknowledge_remote_cleanup_abandoned: Option, }, + RetryTransitionRecovery { + mode: IlmRecoveryReceiptMode, + expected_control_revision: u64, + expected_source_generation_sha256: String, + #[serde(default)] + confirm: Option, + }, } fn parse_recovery_record_mutation_request(body: &[u8]) -> S3Result { @@ -656,6 +669,19 @@ fn parse_recovery_record_mutation_request(body: &[u8]) -> S3Result { export_sha256: &'a str, reason_code: IlmRecoveryDispositionReasonCode, }, + RetryTransitionDryRun { + expected_control_revision: u64, + expected_source_generation_sha256: &'a str, + }, + RetryTransitionExecute { + expected_control_revision: u64, + expected_source_generation_sha256: &'a str, + }, } fn validate_recovery_record_mutation_request( @@ -730,9 +764,62 @@ fn validate_recovery_record_mutation_request( )), } } + IlmRecoveryRecordMutationRequest::RetryTransitionRecovery { + mode, + expected_control_revision, + expected_source_generation_sha256, + confirm, + } => { + if *expected_control_revision == 0 { + return Err(admin_s3_error( + AdminS3ErrorCode::InvalidArgument, + "transition recovery retry requires a nonzero expected control revision", + )); + } + validate_recovery_sha256( + expected_source_generation_sha256, + "invalid transition recovery source generation checksum", + )?; + match mode { + IlmRecoveryReceiptMode::DryRun if confirm.is_none() => { + Ok(ValidatedIlmRecoveryRecordMutation::RetryTransitionDryRun { + expected_control_revision: *expected_control_revision, + expected_source_generation_sha256, + }) + } + IlmRecoveryReceiptMode::DryRun => Err(admin_s3_error( + AdminS3ErrorCode::InvalidArgument, + "ILM recovery retry dry-run must not include confirm", + )), + IlmRecoveryReceiptMode::Execute if *confirm == Some(true) => { + Ok(ValidatedIlmRecoveryRecordMutation::RetryTransitionExecute { + expected_control_revision: *expected_control_revision, + expected_source_generation_sha256, + }) + } + IlmRecoveryReceiptMode::Execute => Err(admin_s3_error( + AdminS3ErrorCode::InvalidRequest, + "transition recovery retry requires confirm=true", + )), + } + } } } +#[derive(Debug, Serialize)] +struct IlmTransitionRecoveryRetryDryRunResponse { + action: &'static str, + mode: IlmRecoveryReceiptMode, + status: TransitionRecoveryRetryStatus, +} + +#[derive(Debug, Serialize)] +struct IlmTransitionRecoveryRetryExecuteResponse { + action: &'static str, + mode: IlmRecoveryReceiptMode, + result: TransitionRecoveryRetryResult, +} + #[derive(Debug, Serialize)] struct IlmRecoveryExportCreateResponse { export_id: String, @@ -929,6 +1016,12 @@ fn map_transition_operator_error(err: TransitionOperatorError) -> S3Error { TransitionOperatorError::CandidateVersionMismatch { .. } => { s3_error!(OperationAborted, "remote candidate version does not match requested exact version") } + TransitionOperatorError::StaleRecoveryControl => { + s3_error!(OperationAborted, "transition recovery control or source generation changed") + } + TransitionOperatorError::RetryNotAllowed => { + s3_error!(OperationAborted, "transition recovery control is not eligible for operator retry") + } TransitionOperatorError::Store(_) | TransitionOperatorError::Remote(_) => { s3_error!(InternalError, "transition reconciliation failed") } @@ -1549,6 +1642,15 @@ impl Operation for IlmRecoveryControlInspectHandler { let control = inspect_recovery_control(store.clone(), &control_id) .await .map_err(map_recovery_control_error)?; + let (transition_retry, transition_retry_not_ready_reason) = + if control.protocol == IlmRecoveryProtocol::TransitionTransaction { + match inspect_transition_recovery_retry_for_operator(store.clone(), &control_id).await { + Ok(status) => (Some(status), None), + Err(_) => (None, Some("source_or_control_not_ready")), + } + } else { + (None, None) + }; let now = OffsetDateTime::now_utc(); let ( export_ready, @@ -1602,6 +1704,8 @@ impl Operation for IlmRecoveryControlInspectHandler { observation_receipt_expires_at_unix_nanos, disposition_dry_run_receipt, disposition_dry_run_receipt_expires_at_unix_nanos, + transition_retry, + transition_retry_not_ready_reason, }, ) } @@ -1625,10 +1729,9 @@ impl Operation for IlmRecoveryRecordMutationHandler { let now = OffsetDateTime::now_utc(); let now_unix_nanos = i64::try_from(now.unix_timestamp_nanos()) .map_err(|_| admin_s3_error(AdminS3ErrorCode::InternalError, "ILM recovery receipt timestamp is invalid"))?; - let receipt_credentials = recovery_receipt_credentials()?; - match mutation { ValidatedIlmRecoveryRecordMutation::Export { observation_receipt } => { + let receipt_credentials = recovery_receipt_credentials()?; let receipt = decode_recovery_receipt(observation_receipt, &receipt_credentials)?; let observation = validate_recovery_observation_receipt( receipt, @@ -1655,6 +1758,7 @@ impl Operation for IlmRecoveryRecordMutationHandler { export_sha256, reason_code: IlmRecoveryDispositionReasonCode::LegacyRemoteCleanupAbandoned, } => { + let receipt_credentials = recovery_receipt_credentials()?; let receipt = decode_recovery_receipt(observation_receipt, &receipt_credentials)?; let observation = validate_recovery_observation_receipt( receipt, @@ -1699,6 +1803,7 @@ impl Operation for IlmRecoveryRecordMutationHandler { export_sha256, reason_code: IlmRecoveryDispositionReasonCode::LegacyRemoteCleanupAbandoned, } => { + let receipt_credentials = recovery_receipt_credentials()?; let receipt = decode_recovery_receipt(observation_receipt, &receipt_credentials)?; let observation = validate_recovery_observation_receipt( receipt, @@ -1726,6 +1831,51 @@ impl Operation for IlmRecoveryRecordMutationHandler { }, ) } + ValidatedIlmRecoveryRecordMutation::RetryTransitionDryRun { + expected_control_revision, + expected_source_generation_sha256, + } => { + let status = inspect_transition_recovery_retry_for_operator(store, &control_id) + .await + .map_err(map_transition_operator_error)?; + if !status.retry_ready { + return Err(map_transition_operator_error(TransitionOperatorError::RetryNotAllowed)); + } + if status.control_revision != expected_control_revision + || status.source_generation_sha256 != expected_source_generation_sha256 + { + return Err(map_transition_operator_error(TransitionOperatorError::StaleRecoveryControl)); + } + json_response( + StatusCode::OK, + &IlmTransitionRecoveryRetryDryRunResponse { + action: "retry_transition_recovery", + mode: IlmRecoveryReceiptMode::DryRun, + status, + }, + ) + } + ValidatedIlmRecoveryRecordMutation::RetryTransitionExecute { + expected_control_revision, + expected_source_generation_sha256, + } => { + let result = retry_transition_recovery_for_operator( + store, + &control_id, + expected_control_revision, + expected_source_generation_sha256, + ) + .await + .map_err(map_transition_operator_error)?; + json_response( + StatusCode::OK, + &IlmTransitionRecoveryRetryExecuteResponse { + action: "retry_transition_recovery", + mode: IlmRecoveryReceiptMode::Execute, + result, + }, + ) + } } } } @@ -2145,6 +2295,48 @@ mod tests { }) if observed_export_id == export_id && observed_export_sha256 == export_sha256 )); + let source_generation_sha256 = "ef".repeat(32); + let retry_dry_run_json = format!( + r#"{{"action":"retry_transition_recovery","mode":"dry_run","expected_control_revision":7,"expected_source_generation_sha256":"{source_generation_sha256}"}}"# + ); + let retry_dry_run = parse_recovery_record_mutation_request(retry_dry_run_json.as_bytes()).unwrap(); + assert!(matches!( + validate_recovery_record_mutation_request(&retry_dry_run), + Ok(ValidatedIlmRecoveryRecordMutation::RetryTransitionDryRun { + expected_control_revision: 7, + expected_source_generation_sha256: observed, + }) if observed == source_generation_sha256 + )); + let retry_execute_json = retry_dry_run_json.replace(r#""mode":"dry_run""#, r#""mode":"execute","confirm":true"#); + let retry_execute = parse_recovery_record_mutation_request(retry_execute_json.as_bytes()).unwrap(); + assert!(matches!( + validate_recovery_record_mutation_request(&retry_execute), + Ok(ValidatedIlmRecoveryRecordMutation::RetryTransitionExecute { + expected_control_revision: 7, + expected_source_generation_sha256: observed, + }) if observed == source_generation_sha256 + )); + assert!( + parse_recovery_record_mutation_request( + retry_dry_run_json + .replace(r#""mode":"dry_run""#, r#""mode":"dry_run","confirm":false"#) + .as_bytes() + ) + .is_err() + ); + let retry_without_confirmation = retry_execute_json.replace(r#""confirm":true,"#, ""); + if let Ok(request) = parse_recovery_record_mutation_request(retry_without_confirmation.as_bytes()) { + assert!(validate_recovery_record_mutation_request(&request).is_err()); + } + for invalid in [ + retry_execute_json.replace(r#""expected_control_revision":7"#, r#""expected_control_revision":0"#), + retry_execute_json.replace(source_generation_sha256.as_str(), "EF".repeat(32).as_str()), + retry_execute_json.replace(source_generation_sha256.as_str(), "too-short"), + ] { + let request = parse_recovery_record_mutation_request(invalid.as_bytes()).unwrap(); + assert!(validate_recovery_record_mutation_request(&request).is_err()); + } + let dry_run_with_confirmation = dry_run_json.replace(r#""mode":"dry_run""#, r#""mode":"dry_run","confirm":false"#); assert!(parse_recovery_record_mutation_request(dry_run_with_confirmation.as_bytes()).is_err()); assert!(parse_recovery_record_mutation_request(dry_run_json.replace('}', r#","confirm":null}"#).as_bytes()).is_err()); diff --git a/rustfs/src/admin/storage_api.rs b/rustfs/src/admin/storage_api.rs index 1f2dc7582..d729fb24a 100644 --- a/rustfs/src/admin/storage_api.rs +++ b/rustfs/src/admin/storage_api.rs @@ -243,8 +243,10 @@ pub(crate) mod lifecycle { IlmRecoveryExportObservation, create_recovery_export, inspect_recovery_export_observation, load_recovery_export, }; pub(crate) use super::ecstore_bucket::lifecycle::transition_transaction::{ - TransitionOperatorDeleteResult, TransitionOperatorError, delete_transition_candidate_for_operator, - finalize_missing_transition_transaction_for_operator, inspect_transition_transaction_for_operator, + TransitionOperatorDeleteResult, TransitionOperatorError, TransitionRecoveryRetryResult, TransitionRecoveryRetryStatus, + delete_transition_candidate_for_operator, finalize_missing_transition_transaction_for_operator, + inspect_transition_recovery_retry_for_operator, inspect_transition_transaction_for_operator, + retry_transition_recovery_for_operator, }; pub(crate) async fn enqueue_transition_for_existing_objects_scoped(