diff --git a/crates/ecstore/src/bucket/lifecycle/durable_namespace.rs b/crates/ecstore/src/bucket/lifecycle/durable_namespace.rs index 786c39242..08c6c57ca 100644 --- a/crates/ecstore/src/bucket/lifecycle/durable_namespace.rs +++ b/crates/ecstore/src/bucket/lifecycle/durable_namespace.rs @@ -12,6 +12,8 @@ // See the License for the specific language governing permissions and // limitations under the License. +use rustfs_utils::crypto::hex_sha256; +use serde::{Deserialize, Serialize}; use uuid::Uuid; use super::{manual_transition_job, tier_delete_journal, transition_transaction}; @@ -89,6 +91,7 @@ pub(crate) struct ValidatedDurableIlmRecord { pub(crate) namespace: &'static str, pub(crate) id_kind: &'static str, pub(crate) id: String, + pub(crate) checkpoint: DurableIlmRecordCheckpoint, } impl ValidatedDurableIlmRecord { @@ -97,6 +100,180 @@ impl ValidatedDurableIlmRecord { } } +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)] +pub(crate) enum DurableIlmRecordCheckpoint { + TierDeleteJournal { + content_sha256: String, + identity_sha256: String, + committed: bool, + }, + TransitionTransaction { + content_sha256: String, + identity_sha256: String, + remote_version_sha256: String, + remote_version_known: bool, + revision: u64, + state: transition_transaction::TransitionTransactionState, + }, + ManualTransitionJob { + content_sha256: String, + identity_sha256: String, + updated_at_unix_nanos: i128, + state: manual_transition_job::ManualTransitionJobState, + scan_completed: bool, + cancel_requested: bool, + }, + ManualTransitionScope { + content_sha256: String, + identity_sha256: String, + updated_at_unix_nanos: i128, + }, + ManualTransitionTask { + content_sha256: String, + }, + ManualTransitionWorkerResult { + content_sha256: String, + }, +} + +impl DurableIlmRecordCheckpoint { + pub(crate) fn content_sha256(&self) -> &str { + match self { + Self::TierDeleteJournal { content_sha256, .. } + | Self::TransitionTransaction { content_sha256, .. } + | Self::ManualTransitionJob { content_sha256, .. } + | Self::ManualTransitionScope { content_sha256, .. } + | Self::ManualTransitionTask { content_sha256 } + | Self::ManualTransitionWorkerResult { content_sha256 } => content_sha256, + } + } + + pub(crate) fn validate_successor(&self, next: &Self) -> Result<()> { + if self == next { + return Ok(()); + } + + let valid = match (self, next) { + ( + Self::TierDeleteJournal { + identity_sha256: previous_identity, + committed: previous_committed, + .. + }, + Self::TierDeleteJournal { + identity_sha256: next_identity, + committed: next_committed, + .. + }, + ) => { + previous_identity == next_identity + && (previous_committed == next_committed || (!previous_committed && *next_committed)) + } + ( + Self::TransitionTransaction { + identity_sha256: previous_identity, + remote_version_sha256: previous_remote_version, + remote_version_known: previous_remote_version_known, + revision: previous_revision, + state: previous_state, + .. + }, + Self::TransitionTransaction { + identity_sha256: next_identity, + remote_version_sha256: next_remote_version, + revision: next_revision, + state: next_state, + .. + }, + ) => { + previous_identity == next_identity + && transition_state_distance(*previous_state, *next_state) + .and_then(|distance| previous_revision.checked_add(distance)) + .is_some_and(|expected_revision| *next_revision == expected_revision) + && (!previous_remote_version_known || previous_remote_version == next_remote_version) + } + ( + Self::ManualTransitionJob { + identity_sha256: previous_identity, + updated_at_unix_nanos: previous_updated_at, + state: previous_state, + scan_completed: previous_scan_completed, + cancel_requested: previous_cancel_requested, + .. + }, + Self::ManualTransitionJob { + identity_sha256: next_identity, + updated_at_unix_nanos: next_updated_at, + state: next_state, + scan_completed: next_scan_completed, + cancel_requested: next_cancel_requested, + .. + }, + ) => { + previous_identity == next_identity + && next_updated_at > previous_updated_at + && manual_job_state_reaches(*previous_state, *next_state) + && (!previous_scan_completed || *next_scan_completed) + && (!previous_cancel_requested || *next_cancel_requested) + } + ( + Self::ManualTransitionScope { + identity_sha256: previous_identity, + updated_at_unix_nanos: previous_updated_at, + .. + }, + Self::ManualTransitionScope { + identity_sha256: next_identity, + updated_at_unix_nanos: next_updated_at, + .. + }, + ) => previous_identity == next_identity && next_updated_at > previous_updated_at, + _ => false, + }; + + if valid { + Ok(()) + } else { + Err(Error::other("durable ILM record generation is not a monotonic successor")) + } + } +} + +fn transition_state_distance( + from: transition_transaction::TransitionTransactionState, + to: transition_transaction::TransitionTransactionState, +) -> Option { + use transition_transaction::TransitionTransactionState::{ + AbortedNoRemote, CleanupPending, Committed, LocalCommitStarted, UploadOutcomeUnknown, UploadStarted, Uploaded, + }; + + match (from, to) { + (UploadStarted, UploadOutcomeUnknown | AbortedNoRemote | Uploaded) => Some(1), + (UploadStarted, LocalCommitStarted | CleanupPending) => Some(2), + (UploadStarted, Committed) => Some(3), + (UploadOutcomeUnknown, Uploaded | CleanupPending) => Some(1), + (UploadOutcomeUnknown, LocalCommitStarted) => Some(2), + (UploadOutcomeUnknown, Committed) => Some(3), + (Uploaded, LocalCommitStarted | CleanupPending) => Some(1), + (Uploaded, Committed) => Some(2), + (LocalCommitStarted, Committed | CleanupPending) => Some(1), + _ => None, + } +} + +fn manual_job_state_reaches( + from: manual_transition_job::ManualTransitionJobState, + to: manual_transition_job::ManualTransitionJobState, +) -> bool { + from == to || from == manual_transition_job::ManualTransitionJobState::Running +} + +fn checkpoint_hash(value: &T) -> Result { + let encoded = serde_json::to_vec(value).map_err(Error::other)?; + Ok(hex_sha256(&encoded, ToOwned::to_owned)) +} + fn path_is_in_namespace(path: &str, namespace: &DurableIlmNamespace) -> bool { let Some(suffix) = path.strip_prefix(namespace.prefix) else { return false; @@ -163,7 +340,8 @@ pub(crate) fn validate_durable_ilm_record(path: &str, data: &[u8]) -> Result { let entry = tier_delete_journal::decode_tier_delete_journal_entry(data)?; if tier_delete_journal::tier_delete_journal_object_name(&entry) != path { @@ -173,12 +351,52 @@ pub(crate) fn validate_durable_ilm_record(path: &str, data: &[u8]) -> Result { let transaction = transition_transaction::decode_transition_transaction_record(path, data) .map_err(|err| Error::other(err.to_string()))?; - ("transaction_id", transaction.transaction_id.to_string()) + let identity_sha256 = checkpoint_hash(&( + transaction.deployment_id, + transaction.transaction_id, + transaction.owner_epoch, + transaction.write_id, + &transaction.source, + &transaction.tier_name, + transaction.backend_fingerprint, + &transaction.remote_object, + transaction.not_after_unix_nanos, + ))?; + let remote_version_sha256 = checkpoint_hash(&transaction.remote_version)?; + ( + "transaction_id", + transaction.transaction_id.to_string(), + DurableIlmRecordCheckpoint::TransitionTransaction { + content_sha256, + identity_sha256, + remote_version_sha256, + remote_version_known: !transaction.remote_version.is_unknown(), + revision: transaction.revision, + state: transaction.state, + }, + ) } DurableIlmRecordKind::ManualTransitionJob => { let job_id = manual_transition_job::manual_transition_job_id_from_record_object_name(path) @@ -188,9 +406,31 @@ pub(crate) fn validate_durable_ilm_record(path: &str, data: &[u8]) -> Result { let admission: manual_transition_job::ManualTransitionScopeAdmission = @@ -201,7 +441,24 @@ pub(crate) fn validate_durable_ilm_record(path: &str, data: &[u8]) -> Result { let (job_id, task_key) = parse_manual_sharded_record(path, namespace.prefix)?; @@ -212,7 +469,11 @@ pub(crate) fn validate_durable_ilm_record(path: &str, data: &[u8]) -> Result { let (job_id, task_key) = parse_manual_sharded_record(path, namespace.prefix)?; @@ -223,7 +484,11 @@ pub(crate) fn validate_durable_ilm_record(path: &str, data: &[u8]) -> Result Result, queue_snapshot: ManualTransitionQueueSnapshot) { @@ -384,7 +384,7 @@ impl ManualTransitionJobRecord { pub fn abandon_recovery_lease(&mut self, lease_id: Uuid) { if self.state == ManualTransitionJobState::Running && self.lease_id == lease_id { self.lease_expires_at_unix_nanos = 0; - self.updated_at_unix_nanos = OffsetDateTime::now_utc().unix_timestamp_nanos(); + self.advance_updated_at(); } } @@ -402,7 +402,7 @@ impl ManualTransitionJobRecord { pub fn renew_lease(&mut self, queue_snapshot: ManualTransitionQueueSnapshot) { let now = OffsetDateTime::now_utc().unix_timestamp_nanos(); - self.updated_at_unix_nanos = now; + self.updated_at_unix_nanos = self.updated_at_unix_nanos.saturating_add(1).max(now); self.lease_expires_at_unix_nanos = manual_transition_job_lease_expires_at(now); self.queue_snapshot = queue_snapshot; } @@ -467,9 +467,13 @@ impl ManualTransitionJobRecord { } fn mark_updated_terminal(&mut self) { + self.advance_updated_at(); + self.completed_at_unix_nanos = Some(self.updated_at_unix_nanos); + } + + fn advance_updated_at(&mut self) { let now = OffsetDateTime::now_utc().unix_timestamp_nanos(); - self.updated_at_unix_nanos = now; - self.completed_at_unix_nanos = Some(now); + self.updated_at_unix_nanos = self.updated_at_unix_nanos.saturating_add(1).max(now); } fn mark_terminal_if_worker_drained(&mut self) { @@ -1113,7 +1117,8 @@ pub fn manual_transition_scope_record_object_name(scope_key: &str) -> Result, job: &ManualTransitionJobRecord) -> EcstoreResult<()> { let object = manual_transition_job_record_object_name(job.job_id).map_err(manual_transition_job_store_error)?; let data = job.encode().map_err(manual_transition_job_store_error)?; - config_boundary::save_config(api, &object, data).await + config_boundary::save_config(api.clone(), &object, data.clone()).await?; + api.record_durable_ilm_decommission_progress(&object, &data).await } pub async fn load_manual_transition_job_record(api: Arc, job_id: Uuid) -> EcstoreResult { @@ -1146,9 +1151,9 @@ pub async fn save_manual_transition_job_record_if_current( let object = manual_transition_job_record_object_name(job.job_id).map_err(manual_transition_job_store_error)?; let data = job.encode().map_err(manual_transition_job_store_error)?; config_boundary::save_config_with_opts_quiet( - api, + api.clone(), &object, - data, + data.clone(), &ObjectOptions { max_parity: true, http_preconditions: Some(HTTPPreconditions { @@ -1158,7 +1163,8 @@ pub async fn save_manual_transition_job_record_if_current( ..Default::default() }, ) - .await + .await?; + api.record_durable_ilm_decommission_progress(&object, &data).await } /// Applies a job-record mutation with optimistic concurrency control. @@ -1596,9 +1602,9 @@ pub async fn save_manual_transition_scope_admission_if_absent( let object = manual_transition_scope_record_object_name(&admission.scope_key).map_err(manual_transition_job_store_error)?; let data = serde_json::to_vec(admission).map_err(Error::other)?; config_boundary::save_config_with_opts( - api, + api.clone(), &object, - data, + data.clone(), &ObjectOptions { max_parity: true, http_preconditions: Some(HTTPPreconditions { @@ -1608,7 +1614,8 @@ pub async fn save_manual_transition_scope_admission_if_absent( ..Default::default() }, ) - .await + .await?; + api.record_durable_ilm_decommission_progress(&object, &data).await } pub async fn load_manual_transition_scope_admission( @@ -1646,9 +1653,9 @@ pub async fn save_manual_transition_scope_admission_if_current( let object = manual_transition_scope_record_object_name(&admission.scope_key).map_err(manual_transition_job_store_error)?; let data = serde_json::to_vec(admission).map_err(Error::other)?; match config_boundary::save_config_with_opts( - api, + api.clone(), &object, - data, + data.clone(), &ObjectOptions { max_parity: true, http_preconditions: Some(HTTPPreconditions { @@ -1664,7 +1671,8 @@ pub async fn save_manual_transition_scope_admission_if_current( Err(Error::PreconditionFailed) } result => result, - } + }?; + api.record_durable_ilm_decommission_progress(&object, &data).await } pub async fn claim_manual_transition_scope_admission( @@ -1957,13 +1965,15 @@ pub async fn delete_manual_transition_scope_admission_if_current( job_id: Uuid, lease_id: Uuid, ) -> EcstoreResult { - let etag = match load_manual_transition_scope_admission_with_etag(api.clone(), scope_key).await { - Ok((admission, etag)) if admission.job_id == job_id && admission.lease_id == lease_id => etag, + let (admission, etag) = match load_manual_transition_scope_admission_with_etag(api.clone(), scope_key).await { + Ok((admission, etag)) if admission.job_id == job_id && admission.lease_id == lease_id => (admission, etag), Ok(_) => return Ok(false), Err(Error::ConfigNotFound) => return Ok(true), Err(err) => return Err(err), }; let object = manual_transition_scope_record_object_name(scope_key).map_err(manual_transition_job_store_error)?; + let data = serde_json::to_vec(&admission).map_err(Error::other)?; + api.record_durable_ilm_decommission_terminal(&object, &data).await?; match config_boundary::delete_config_if_match(api, &object, &etag).await { Ok(()) | Err(Error::ConfigNotFound) => Ok(true), Err(Error::PreconditionFailed) => Ok(false), diff --git a/crates/ecstore/src/bucket/lifecycle/tier_delete_journal.rs b/crates/ecstore/src/bucket/lifecycle/tier_delete_journal.rs index 48f79d5f6..b53ba33c1 100644 --- a/crates/ecstore/src/bucket/lifecycle/tier_delete_journal.rs +++ b/crates/ecstore/src/bucket/lifecycle/tier_delete_journal.rs @@ -433,9 +433,18 @@ async fn process_committed_tier_delete_journal_entry(api: Arc, je: &Jen ) .await?; } + record_tier_delete_journal_decommission_terminal(&api, je).await?; remove_tier_delete_journal_entry(api, je).await } +async fn record_tier_delete_journal_decommission_terminal(api: &Arc, je: &Jentry) -> std::io::Result<()> { + let path = tier_delete_journal_object_name(je); + let data = encode_tier_delete_journal_entry(je).map_err(std::io::Error::other)?; + api.record_durable_ilm_decommission_terminal(&path, &data) + .await + .map_err(std::io::Error::other) +} + fn object_info_references_tier_delete(info: &ObjectInfo, je: &Jentry) -> std::io::Result { if info.transitioned_object.status != rustfs_filemeta::TRANSITION_COMPLETE || info.transitioned_object.name != je.obj_name diff --git a/crates/ecstore/src/bucket/lifecycle/transition_transaction.rs b/crates/ecstore/src/bucket/lifecycle/transition_transaction.rs index ae216bc29..70bf4ed53 100644 --- a/crates/ecstore/src/bucket/lifecycle/transition_transaction.rs +++ b/crates/ecstore/src/bucket/lifecycle/transition_transaction.rs @@ -585,7 +585,8 @@ pub(crate) async fn save_transition_transaction_record( let object = transition_transaction_record_object_name(transaction.transaction_id).map_err(transition_transaction_store_error)?; let data = transaction.encode().map_err(transition_transaction_store_error)?; - config_boundary::save_config(api, &object, data).await + config_boundary::save_config(api.clone(), &object, data.clone()).await?; + api.record_durable_ilm_decommission_progress(&object, &data).await } pub(crate) async fn load_transition_transaction_record( @@ -597,8 +598,14 @@ pub(crate) async fn load_transition_transaction_record( TransitionTransaction::decode(transaction_id, &data).map_err(transition_transaction_store_error) } -pub(crate) async fn delete_transition_transaction_record(api: Arc, transaction_id: Uuid) -> EcstoreResult<()> { - let object = transition_transaction_record_object_name(transaction_id).map_err(transition_transaction_store_error)?; +pub(crate) async fn delete_transition_transaction_record( + api: Arc, + transaction: &TransitionTransaction, +) -> EcstoreResult<()> { + let object = + transition_transaction_record_object_name(transaction.transaction_id).map_err(transition_transaction_store_error)?; + let data = transaction.encode().map_err(transition_transaction_store_error)?; + api.record_durable_ilm_decommission_terminal(&object, &data).await?; match config_boundary::delete_config(api, &object).await { Ok(()) | Err(Error::ConfigNotFound) => Ok(()), Err(err) => Err(err), @@ -814,7 +821,7 @@ pub async fn finalize_missing_transition_transaction_for_operator( if probe != TransitionOperatorProbe::Missing { return Err(TransitionOperatorError::CandidateNotMissing(probe)); } - delete_transition_transaction_record(api, transaction_id) + delete_transition_transaction_record(api, &transaction) .await .map_err(TransitionOperatorError::Store) } @@ -850,22 +857,22 @@ pub async fn process_transition_transaction_record( match transaction.state { TransitionTransactionState::Uploaded => { delete_transition_remote_candidate(api.clone(), transaction).await?; - delete_transition_transaction_record(api, transaction.transaction_id).await?; + delete_transition_transaction_record(api, transaction).await?; Ok(TransitionTransactionRecoveryOutcome::RemoteCandidateDeleted) } TransitionTransactionState::CleanupPending => match local_commit_matches_transaction(api.clone(), transaction).await { Ok(true) => { - delete_transition_transaction_record(api, transaction.transaction_id).await?; + delete_transition_transaction_record(api, transaction).await?; Ok(TransitionTransactionRecoveryOutcome::RecordDeleted) } Ok(false) => { delete_transition_remote_candidate(api.clone(), transaction).await?; - delete_transition_transaction_record(api, transaction.transaction_id).await?; + delete_transition_transaction_record(api, transaction).await?; Ok(TransitionTransactionRecoveryOutcome::RemoteCandidateDeleted) } Err(err) if transition_source_is_missing(&err) => { delete_transition_remote_candidate(api.clone(), transaction).await?; - delete_transition_transaction_record(api, transaction.transaction_id).await?; + delete_transition_transaction_record(api, transaction).await?; Ok(TransitionTransactionRecoveryOutcome::RemoteCandidateDeleted) } Err(err) => Err(err), @@ -873,7 +880,7 @@ pub async fn process_transition_transaction_record( TransitionTransactionState::LocalCommitStarted => { match local_commit_matches_transaction(api.clone(), transaction).await { Ok(true) => { - delete_transition_transaction_record(api, transaction.transaction_id).await?; + delete_transition_transaction_record(api, transaction).await?; Ok(TransitionTransactionRecoveryOutcome::RecordDeleted) } Ok(false) => Ok(TransitionTransactionRecoveryOutcome::Retained), @@ -882,7 +889,7 @@ pub async fn process_transition_transaction_record( } } TransitionTransactionState::AbortedNoRemote | TransitionTransactionState::Committed => { - delete_transition_transaction_record(api, transaction.transaction_id).await?; + delete_transition_transaction_record(api, transaction).await?; Ok(TransitionTransactionRecoveryOutcome::RecordDeleted) } TransitionTransactionState::UploadOutcomeUnknown => recover_unknown_upload_outcome(api, transaction).await, @@ -908,7 +915,7 @@ async fn recover_unknown_upload_outcome( .map_err(Error::other)? { TransitionCandidateProbe::Missing => { - delete_transition_transaction_record(api, transaction.transaction_id).await?; + delete_transition_transaction_record(api, transaction).await?; Ok(TransitionTransactionRecoveryOutcome::RecordDeleted) } TransitionCandidateProbe::UnversionedPresent => { @@ -926,7 +933,7 @@ async fn recover_unknown_upload_outcome( ) .await .map_err(Error::other)?; - delete_transition_transaction_record(api, transaction.transaction_id).await?; + delete_transition_transaction_record(api, transaction).await?; Ok(TransitionTransactionRecoveryOutcome::RemoteCandidateDeleted) } TransitionCandidateProbe::VersionedPresent(version_id) => { @@ -959,7 +966,7 @@ async fn cleanup_recovered_unknown_upload_candidate( .map_err(transition_transaction_store_error)?; save_transition_transaction_record(api.clone(), &cleanup).await?; delete_transition_remote_candidate(api.clone(), &cleanup).await?; - delete_transition_transaction_record(api, cleanup.transaction_id).await?; + delete_transition_transaction_record(api, &cleanup).await?; Ok(TransitionTransactionRecoveryOutcome::RemoteCandidateDeleted) } diff --git a/crates/ecstore/src/config/com.rs b/crates/ecstore/src/config/com.rs index e03eace21..472c87e63 100644 --- a/crates/ecstore/src/config/com.rs +++ b/crates/ecstore/src/config/com.rs @@ -410,10 +410,21 @@ pub(crate) async fn read_config_limited_preserve_empty(api: Arc, file: &st where S: EcstoreObjectIO, { - let (data, _obj) = read_config_with_metadata_inner(api, file, &ObjectOptions::default(), true, Some(max_bytes)).await?; + let (data, _obj) = read_config_limited_preserve_empty_with_metadata(api, file, max_bytes).await?; Ok(data) } +pub(crate) async fn read_config_limited_preserve_empty_with_metadata( + api: Arc, + file: &str, + max_bytes: usize, +) -> Result<(Vec, ObjectInfo)> +where + S: EcstoreObjectIO, +{ + read_config_with_metadata_inner(api, file, &ObjectOptions::default(), true, Some(max_bytes)).await +} + /// Read an existing config object without treating an empty payload as absent. /// Callers that validate their own payload format need to distinguish corruption /// from `ConfigNotFound`. diff --git a/crates/ecstore/src/core/pools.rs b/crates/ecstore/src/core/pools.rs index 2b59acc66..08c5075f2 100644 --- a/crates/ecstore/src/core/pools.rs +++ b/crates/ecstore/src/core/pools.rs @@ -16,7 +16,7 @@ use crate::bucket::replication::replication_state_from_filemeta; use crate::bucket::versioning_sys::BucketVersioningSys; use crate::bucket::{ lifecycle::{ - ILM_META_PREFIX, LifecycleExpiryConfigs, ValidatedDurableIlmRecord, + DurableIlmRecordCheckpoint, ILM_META_PREFIX, LifecycleExpiryConfigs, ValidatedDurableIlmRecord, bucket_lifecycle_audit::LcEventSrc, bucket_lifecycle_ops::{ LifecycleOps, apply_expiry_on_transitioned_object, apply_expiry_rule_in, eval_action_from_lifecycle, @@ -30,8 +30,8 @@ use crate::bucket::{ }; use crate::cache_value::metacache_set::{ListPathRawOptions, list_path_raw}; use crate::config::com::{ - CONFIG_PREFIX, delete_config, read_config, read_config_limited_preserve_empty, read_config_no_lock, save_config, - save_config_with_opts, + CONFIG_PREFIX, delete_config, read_config, read_config_limited_preserve_empty, + read_config_limited_preserve_empty_with_metadata, read_config_no_lock, save_config, save_config_with_opts, }; use crate::data_movement; use crate::data_movement::backpressure::{self, DataMovementOperation}; @@ -54,7 +54,7 @@ use crate::storage_api_contracts::{ heal::HealOperations as _, list::ListOperations as _, namespace::NamespaceLocking as _, - object::{EcstoreObjectIO, ObjectIO as _, ObjectOperations as _}, + object::{EcstoreObjectIO, HTTPPreconditions, ObjectIO as _, ObjectOperations as _}, }; use crate::{core::sets::Sets, store::ECStore}; use byteorder::{ByteOrder, LittleEndian, WriteBytesExt}; @@ -104,8 +104,9 @@ const DECOMMISSION_TARGET_CAPACITY_OVERHEAD_PERCENT: usize = 30; const DECOMMISSION_LISTING_MAX_ATTEMPTS: usize = 3; const DECOMMISSION_LISTING_RETRY_DELAY: std::time::Duration = std::time::Duration::from_secs(5); const DECOMMISSION_DURABLE_ILM_RECEIPT_ROOT: &str = "decommission/ilm-receipts"; -const DECOMMISSION_DURABLE_ILM_RECEIPT_SCHEMA: &str = "v1"; +const DECOMMISSION_DURABLE_ILM_RECEIPT_SCHEMA: &str = "v2"; const DECOMMISSION_DURABLE_ILM_RECEIPT_MAX_SIZE: usize = 16 * 1024; +const DECOMMISSION_DURABLE_ILM_RECEIPT_CAS_ATTEMPTS: usize = 3; /// Background decommission walks must tolerate slow object migrations; the /// stall timeout is the drive-health bound, not the total listing duration. const DECOMMISSION_BACKGROUND_WALKDIR_STALL_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(60); @@ -857,22 +858,24 @@ async fn record_decommission_entry_error( fn validate_decommission_durable_ilm_copy( path: &str, source_record: &ValidatedDurableIlmRecord, - source: &[u8], target: &[u8], -) -> Result<()> { - validate_durable_ilm_record(path, target).map_err(|err| { +) -> Result { + let target_record = validate_durable_ilm_record(path, target).map_err(|err| { Error::other(format!( "target durable ILM record is invalid at path `{path}` {}: {err}", source_record.context() )) })?; - if source != target { - return Err(Error::other(format!( - "target durable ILM record content mismatch at path `{path}` {}", - source_record.context() - ))); - } - Ok(()) + source_record + .checkpoint + .validate_successor(&target_record.checkpoint) + .map_err(|err| { + Error::other(format!( + "target durable ILM record generation mismatch at path `{path}` {}: {err}", + source_record.context() + )) + })?; + Ok(target_record) } #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] @@ -882,17 +885,19 @@ struct DecommissionDurableIlmReceipt { namespace: String, id_kind: String, id: String, - target_content_sha256: String, + checkpoint: DurableIlmRecordCheckpoint, + terminal_checkpoint: Option, } impl DecommissionDurableIlmReceipt { - fn new(path: &str, record: &ValidatedDurableIlmRecord, target: &[u8]) -> Self { + fn new(path: &str, record: &ValidatedDurableIlmRecord) -> Self { Self { source_path: path.to_string(), namespace: record.namespace.to_string(), id_kind: record.id_kind.to_string(), id: record.id.clone(), - target_content_sha256: hex_sha256(target, ToOwned::to_owned), + checkpoint: record.checkpoint.clone(), + terminal_checkpoint: None, } } @@ -915,13 +920,22 @@ impl DecommissionDurableIlmReceipt { self.source_path ))); } - if !is_sha256_checksum(&self.target_content_sha256) { + if !is_sha256_checksum(self.checkpoint.content_sha256()) { return Err(Error::other(format!( "receipt target checksum is invalid for source path `{}` {}", self.source_path, self.context() ))); } + if let Some(terminal_checkpoint) = &self.terminal_checkpoint { + self.checkpoint.validate_successor(terminal_checkpoint).map_err(|err| { + Error::other(format!( + "receipt terminal checkpoint is invalid for source path `{}` {}: {err}", + self.source_path, + self.context() + )) + })?; + } Ok(()) } @@ -973,14 +987,80 @@ struct PersistedDecommissionDurableIlmReceipt { receipt: DecommissionDurableIlmReceipt, } -fn decommission_durable_ilm_receipt_prefix(cmd_line: &str) -> String { - let pool_key = hex_sha256(cmd_line.as_bytes(), ToOwned::to_owned); - format!("{DECOMMISSION_DURABLE_ILM_RECEIPT_ROOT}/{pool_key}/") +#[derive(Debug, Clone, PartialEq, Eq)] +struct DecommissionDurableIlmReceiptLocator { + run_token: String, + source_path: String, + id_kind: String, + id: String, } -fn decommission_durable_ilm_receipt_path(prefix: &str, source_path: &str) -> String { - let source_key = hex_sha256(source_path.as_bytes(), ToOwned::to_owned); - format!("{prefix}{source_key}.json") +impl DecommissionDurableIlmReceiptLocator { + fn context(&self) -> String { + format!("source path `{}` {} `{}`", self.source_path, self.id_kind, self.id) + } +} + +fn decommission_durable_ilm_receipt_run_token(cmd_line: &str) -> String { + hex_sha256(cmd_line.as_bytes(), ToOwned::to_owned) +} + +fn decommission_durable_ilm_receipt_run_prefix(run_token: &str) -> String { + format!("{DECOMMISSION_DURABLE_ILM_RECEIPT_ROOT}/{run_token}/") +} + +fn decommission_durable_ilm_receipt_path(run_token: &str, source_path: &str, id_kind: &str, id: &str) -> String { + format!( + "{}{}/{}/{}.json", + decommission_durable_ilm_receipt_run_prefix(run_token), + source_path, + id_kind, + id + ) +} + +fn parse_decommission_durable_ilm_receipt_path(path: &str) -> Result { + let prefix = format!("{DECOMMISSION_DURABLE_ILM_RECEIPT_ROOT}/"); + let suffix = path + .strip_prefix(&prefix) + .ok_or_else(|| Error::other(format!("durable ILM receipt path `{path}` has the wrong root")))?; + let (run_token, record_path) = suffix + .split_once('/') + .ok_or_else(|| Error::other(format!("durable ILM receipt path `{path}` is missing its record path")))?; + let mut parts = record_path.rsplitn(3, '/'); + let id = parts + .next() + .and_then(|file| file.strip_suffix(".json")) + .filter(|id| !id.is_empty()) + .ok_or_else(|| Error::other(format!("durable ILM receipt path `{path}` is missing its record id")))?; + let id_kind = parts + .next() + .filter(|id_kind| matches!(*id_kind, "operation_id" | "transaction_id" | "job_id")) + .ok_or_else(|| Error::other(format!("durable ILM receipt path `{path}` has an invalid id kind")))?; + let source_path = parts + .next() + .filter(|source_path| !source_path.is_empty()) + .ok_or_else(|| Error::other(format!("durable ILM receipt path `{path}` is missing its source path")))?; + if !is_sha256_checksum(run_token) { + return Err(Error::other(format!("durable ILM receipt path `{path}` has an invalid run token"))); + } + match id_kind { + "operation_id" if !is_sha256_checksum(id) => { + return Err(Error::other(format!("durable ILM receipt path `{path}` has an invalid operation id"))); + } + "transaction_id" | "job_id" if uuid::Uuid::parse_str(id).is_err() => { + return Err(Error::other(format!("durable ILM receipt path `{path}` has an invalid UUID"))); + } + _ => {} + } + classify_durable_ilm_record(source_path)? + .ok_or_else(|| Error::other(format!("durable ILM receipt path `{path}` does not identify a durable ILM source path")))?; + Ok(DecommissionDurableIlmReceiptLocator { + run_token: run_token.to_string(), + source_path: source_path.to_string(), + id_kind: id_kind.to_string(), + id: id.to_string(), + }) } fn resolve_decommission_pool_meta_reload_result(result: Result<()>, stage: &str) -> Result<()> { @@ -4543,7 +4623,7 @@ impl ECStore { Ok(ret) } - async fn durable_ilm_receipt_prefix(&self, source_pool_idx: usize) -> Result { + async fn durable_ilm_receipt_run_token(&self, source_pool_idx: usize) -> Result { let pool_meta = self.pool_meta.read().await; let cmd_line = pool_meta .pools @@ -4551,7 +4631,7 @@ impl ECStore { .ok_or_else(|| invalid_decommission_pool_index_error(pool_meta.pools.len(), source_pool_idx))? .cmd_line .clone(); - Ok(decommission_durable_ilm_receipt_prefix(&cmd_line)) + Ok(decommission_durable_ilm_receipt_run_token(&cmd_line)) } async fn load_decommissioned_durable_ilm_target( @@ -4560,7 +4640,7 @@ impl ECStore { path: &str, max_record_size: usize, record_context: &str, - ) -> Result<(usize, Vec)> { + ) -> Result)>> { let mut target = None::<(usize, Vec)>; let mut first_read_error = None; for (target_pool_idx, pool) in self.pools.iter().enumerate() { @@ -4592,40 +4672,62 @@ impl ECStore { } } - target.ok_or_else(|| { - first_read_error.unwrap_or_else(|| { - Error::other(format!("target durable ILM record is missing at path `{path}` {record_context}")) - }) - }) + match (target, first_read_error) { + (Some(target), _) => Ok(Some(target)), + (None, Some(err)) => Err(err), + (None, None) => Ok(None), + } + } + + async fn list_decommission_durable_ilm_receipt_paths_in_pool(&self, pool_idx: usize, prefix: &str) -> Result> { + let pool = self + .pools + .get(pool_idx) + .ok_or_else(|| invalid_decommission_pool_index_error(self.pools.len(), pool_idx))?; + let mut receipts = Vec::new(); + let mut continuation = None; + loop { + let page = pool + .clone() + .list_objects_v2(RUSTFS_META_BUCKET, prefix, continuation, None, 1000, false, None, false) + .await + .map_err(|err| { + Error::other(format!( + "failed to list durable ILM decommission receipts under `{prefix}` in pool {pool_idx}: {err}" + )) + })?; + receipts.extend(page.objects.into_iter().map(|object| object.name)); + if !page.is_truncated { + break; + } + continuation = Some(page.next_continuation_token.ok_or_else(|| { + Error::other(format!( + "durable ILM decommission receipt listing under `{prefix}` in pool {pool_idx} was truncated without a continuation token" + )) + })?); + } + Ok(receipts) } async fn list_decommission_durable_ilm_receipts(&self, source_pool_idx: usize) -> Result> { - let prefix = self.durable_ilm_receipt_prefix(source_pool_idx).await?; + let run_token = self.durable_ilm_receipt_run_token(source_pool_idx).await?; + let prefix = decommission_durable_ilm_receipt_run_prefix(&run_token); let mut receipts = Vec::new(); - for (pool_idx, pool) in self.pools.iter().enumerate() { + for pool_idx in 0..self.pools.len() { if pool_idx == source_pool_idx { continue; } - let mut continuation = None; - loop { - let page = pool - .clone() - .list_objects_v2(RUSTFS_META_BUCKET, &prefix, continuation, None, 1000, false, None, false) - .await - .map_err(|err| { - Error::other(format!( - "failed to list durable ILM decommission receipts under `{prefix}` in pool {pool_idx}: {err}" - )) - })?; - receipts.extend(page.objects.into_iter().map(|object| (pool_idx, object.name))); - if !page.is_truncated { - break; + for receipt_path in self + .list_decommission_durable_ilm_receipt_paths_in_pool(pool_idx, &prefix) + .await? + { + let locator = parse_decommission_durable_ilm_receipt_path(&receipt_path)?; + if locator.run_token != run_token { + return Err(Error::other(format!( + "durable ILM receipt path `{receipt_path}` has an unexpected run token" + ))); } - continuation = Some(page.next_continuation_token.ok_or_else(|| { - Error::other(format!( - "durable ILM decommission receipt listing under `{prefix}` in pool {pool_idx} was truncated without a continuation token" - )) - })?); + receipts.push((pool_idx, receipt_path)); } } Ok(receipts) @@ -4637,8 +4739,8 @@ impl ECStore { target_pool_idx: usize, receipt: &DecommissionDurableIlmReceipt, ) -> Result<()> { - let prefix = self.durable_ilm_receipt_prefix(source_pool_idx).await?; - let receipt_path = decommission_durable_ilm_receipt_path(&prefix, &receipt.source_path); + let run_token = self.durable_ilm_receipt_run_token(source_pool_idx).await?; + let receipt_path = decommission_durable_ilm_receipt_path(&run_token, &receipt.source_path, &receipt.id_kind, &receipt.id); let encoded = receipt.encode().map_err(|err| { Error::other(format!( "failed to encode durable ILM decommission receipt `{receipt_path}` for source path `{}` {}: {err}", @@ -4657,36 +4759,57 @@ impl ECStore { }) } + fn validate_decommission_durable_ilm_receipt_locator( + receipt_path: &str, + locator: &DecommissionDurableIlmReceiptLocator, + receipt: &DecommissionDurableIlmReceipt, + ) -> Result<()> { + if locator.source_path != receipt.source_path || locator.id_kind != receipt.id_kind || locator.id != receipt.id { + return Err(Error::other(format!( + "durable ILM decommission receipt path `{receipt_path}` identity {} does not match receipt {}", + locator.context(), + receipt.context() + ))); + } + Ok(()) + } + + async fn read_decommission_durable_ilm_receipt( + &self, + receipt_pool_idx: usize, + receipt_path: &str, + ) -> Result { + let locator = parse_decommission_durable_ilm_receipt_path(receipt_path)?; + let data = read_config_limited_preserve_empty( + self.pools[receipt_pool_idx].clone(), + receipt_path, + DECOMMISSION_DURABLE_ILM_RECEIPT_MAX_SIZE, + ) + .await + .map_err(|err| { + Error::other(format!( + "failed to read durable ILM decommission receipt `{receipt_path}` from pool {receipt_pool_idx} for {}: {err}", + locator.context() + )) + })?; + let receipt = DecommissionDurableIlmReceipt::decode(&data).map_err(|err| { + Error::other(format!( + "durable ILM decommission receipt `{receipt_path}` in pool {receipt_pool_idx} for {} is invalid: {err}", + locator.context() + )) + })?; + Self::validate_decommission_durable_ilm_receipt_locator(receipt_path, &locator, &receipt)?; + Ok(receipt) + } + async fn verify_decommission_durable_ilm_receipts(&self, source_pool_idx: usize) -> Result<()> { - let prefix = self.durable_ilm_receipt_prefix(source_pool_idx).await?; for (receipt_pool_idx, receipt_path) in self.list_decommission_durable_ilm_receipts(source_pool_idx).await? { - let data = read_config_limited_preserve_empty( - self.pools[receipt_pool_idx].clone(), - &receipt_path, - DECOMMISSION_DURABLE_ILM_RECEIPT_MAX_SIZE, - ) - .await - .map_err(|err| { - Error::other(format!( - "failed to read durable ILM decommission receipt `{receipt_path}` from pool {receipt_pool_idx}: {err}" - )) - })?; - let receipt = DecommissionDurableIlmReceipt::decode(&data).map_err(|err| { - Error::other(format!( - "durable ILM decommission receipt `{receipt_path}` in pool {receipt_pool_idx} is invalid: {err}" - )) - })?; - let expected_receipt_path = decommission_durable_ilm_receipt_path(&prefix, &receipt.source_path); - if receipt_path != expected_receipt_path { - return Err(Error::other(format!( - "durable ILM decommission receipt path `{receipt_path}` does not match source path `{}` {}", - receipt.source_path, - receipt.context() - ))); - } + let receipt = self + .read_decommission_durable_ilm_receipt(receipt_pool_idx, &receipt_path) + .await?; let namespace = classify_durable_ilm_record(&receipt.source_path)? .ok_or_else(|| Error::other(format!("path `{}` is not a durable ILM record", receipt.source_path)))?; - let (_, target) = self + let target = self .load_decommissioned_durable_ilm_target( source_pool_idx, &receipt.source_path, @@ -4694,28 +4817,50 @@ impl ECStore { &receipt.context(), ) .await?; - let target_record = validate_durable_ilm_record(&receipt.source_path, &target).map_err(|err| { - Error::other(format!( - "target durable ILM record is invalid at path `{}` {}: {err}", - receipt.source_path, - receipt.context() - )) - })?; - if target_record.namespace != receipt.namespace - || target_record.id_kind != receipt.id_kind - || target_record.id != receipt.id - { + if let Some((_, target)) = target { + let target_record = validate_durable_ilm_record(&receipt.source_path, &target).map_err(|err| { + Error::other(format!( + "target durable ILM record is invalid at path `{}` {}: {err}", + receipt.source_path, + receipt.context() + )) + })?; + let identity_matches = target_record.namespace == receipt.namespace + && target_record.id_kind == receipt.id_kind + && target_record.id == receipt.id; + let reused_manual_scope = receipt.terminal_checkpoint.is_some() + && matches!( + (&receipt.checkpoint, &target_record.checkpoint), + ( + DurableIlmRecordCheckpoint::ManualTransitionScope { .. }, + DurableIlmRecordCheckpoint::ManualTransitionScope { .. } + ) + ); + if !identity_matches && !reused_manual_scope { + return Err(Error::other(format!( + "target durable ILM record identity mismatch at path `{}` {}; decoded {}", + receipt.source_path, + receipt.context(), + target_record.context() + ))); + } + if identity_matches { + receipt + .terminal_checkpoint + .as_ref() + .unwrap_or(&receipt.checkpoint) + .validate_successor(&target_record.checkpoint) + .map_err(|err| { + Error::other(format!( + "target durable ILM record generation mismatch at path `{}` {}: {err}", + receipt.source_path, + receipt.context() + )) + })?; + } + } else if receipt.terminal_checkpoint.is_none() { return Err(Error::other(format!( - "target durable ILM record identity mismatch at path `{}` {}; decoded {}", - receipt.source_path, - receipt.context(), - target_record.context() - ))); - } - let target_content_sha256 = hex_sha256(&target, ToOwned::to_owned); - if target_content_sha256 != receipt.target_content_sha256 { - return Err(Error::other(format!( - "target durable ILM record content mismatch at path `{}` {}", + "target durable ILM record is missing at path `{}` {} without a recovery terminal checkpoint", receipt.source_path, receipt.context() ))); @@ -4724,6 +4869,180 @@ impl ECStore { Ok(()) } + async fn advance_durable_ilm_decommission_receipt( + &self, + pool_idx: usize, + receipt_path: &str, + record: &ValidatedDurableIlmRecord, + terminal: bool, + ) -> Result { + let stage = if terminal { "terminal" } else { "progress" }; + let locator = parse_decommission_durable_ilm_receipt_path(receipt_path)?; + let mut attempt = 1; + loop { + let (receipt_data, metadata) = match read_config_limited_preserve_empty_with_metadata( + self.pools[pool_idx].clone(), + receipt_path, + DECOMMISSION_DURABLE_ILM_RECEIPT_MAX_SIZE, + ) + .await + { + Ok(receipt) => receipt, + Err(err) + if matches!(&err, Error::ConfigNotFound | Error::FileNotFound | Error::FileVersionNotFound) + || is_err_object_not_found(&err) + || is_err_version_not_found(&err) => + { + return Ok(false); + } + Err(err) => { + return Err(Error::other(format!( + "failed to read durable ILM decommission receipt `{receipt_path}` from pool {pool_idx} for {}: {err}", + locator.context() + ))); + } + }; + let mut receipt = DecommissionDurableIlmReceipt::decode(&receipt_data).map_err(|err| { + Error::other(format!( + "durable ILM decommission receipt `{receipt_path}` in pool {pool_idx} for {} is invalid: {err}", + locator.context() + )) + })?; + Self::validate_decommission_durable_ilm_receipt_locator(receipt_path, &locator, &receipt)?; + receipt.checkpoint.validate_successor(&record.checkpoint).map_err(|err| { + Error::other(format!( + "{stage} durable ILM record generation mismatch at path `{}` {}: {err}", + receipt.source_path, + receipt.context() + )) + })?; + + if terminal { + if let Some(existing) = &receipt.terminal_checkpoint { + if existing == &record.checkpoint || record.checkpoint.validate_successor(existing).is_ok() { + return Ok(true); + } + existing.validate_successor(&record.checkpoint).map_err(|err| { + Error::other(format!( + "terminal durable ILM record checkpoint conflicts at path `{}` {}: {err}", + receipt.source_path, + receipt.context() + )) + })?; + } + receipt.terminal_checkpoint = Some(record.checkpoint.clone()); + } else { + if let Some(existing) = &receipt.terminal_checkpoint { + if existing == &record.checkpoint { + return Ok(true); + } + existing.validate_successor(&record.checkpoint).map_err(|err| { + Error::other(format!( + "progress durable ILM record conflicts with terminal checkpoint at path `{}` {}: {err}", + receipt.source_path, + receipt.context() + )) + })?; + receipt.terminal_checkpoint = None; + } + if receipt.checkpoint == record.checkpoint { + return Ok(true); + } + receipt.checkpoint = record.checkpoint.clone(); + } + + let etag = metadata.etag.filter(|etag| !etag.trim().is_empty()).ok_or_else(|| { + Error::other(format!( + "durable ILM decommission receipt `{receipt_path}` in pool {pool_idx} is missing an ETag" + )) + })?; + let encoded = receipt.encode()?; + match save_config_with_opts( + self.pools[pool_idx].clone(), + receipt_path, + encoded, + &ObjectOptions { + max_parity: true, + http_preconditions: Some(HTTPPreconditions { + if_match: Some(etag), + ..Default::default() + }), + ..Default::default() + }, + ) + .await + { + Ok(()) => return Ok(true), + Err(Error::PreconditionFailed) if attempt < DECOMMISSION_DURABLE_ILM_RECEIPT_CAS_ATTEMPTS => { + attempt += 1; + continue; + } + Err(Error::PreconditionFailed) => { + return Err(Error::other(format!( + "failed to persist {stage} durable ILM decommission receipt `{receipt_path}` for {} after concurrent updates", + locator.context() + ))); + } + Err(err) => { + return Err(Error::other(format!( + "failed to persist {stage} durable ILM decommission receipt `{receipt_path}` for {}: {err}", + locator.context() + ))); + } + } + } + } + + async fn advance_durable_ilm_decommission_receipts(&self, path: &str, data: &[u8], terminal: bool) -> Result<()> { + let active_runs = { + let pool_meta = self.pool_meta.read().await; + pool_meta + .pools + .iter() + .enumerate() + .filter(|(_, pool)| { + pool.decommission + .as_ref() + .is_some_and(|info| info.has_decommission_state() && !info.complete) + }) + .map(|(pool_idx, pool)| (pool_idx, decommission_durable_ilm_receipt_run_token(&pool.cmd_line))) + .collect::>() + }; + if active_runs.is_empty() { + return Ok(()); + } + + let stage = if terminal { "terminal" } else { "progress" }; + let record = validate_durable_ilm_record(path, data) + .map_err(|err| Error::other(format!("{stage} durable ILM record is invalid at path `{path}`: {err}")))?; + for (source_pool_idx, run_token) in active_runs { + let receipt_path = decommission_durable_ilm_receipt_path(&run_token, path, record.id_kind, &record.id); + let mut receipt_found = false; + for pool_idx in 0..self.pools.len() { + if pool_idx != source_pool_idx { + receipt_found |= self + .advance_durable_ilm_decommission_receipt(pool_idx, &receipt_path, &record, terminal) + .await?; + } + } + if terminal && !receipt_found { + return Err(Error::other(format!( + "terminal durable ILM record at path `{path}` {} is retained until its decommission receipt is committed", + record.context() + ))); + } + } + Ok(()) + } + + pub(crate) async fn record_durable_ilm_decommission_progress(&self, path: &str, data: &[u8]) -> Result<()> { + self.advance_durable_ilm_decommission_receipts(path, data, false).await + } + + pub(crate) async fn record_durable_ilm_decommission_terminal(&self, path: &str, data: &[u8]) -> Result<()> { + self.advance_durable_ilm_decommission_receipts(path, data, true).await + } + async fn cleanup_decommission_durable_ilm_receipts(&self, source_pool_idx: usize) -> Result<()> { for (pool_idx, receipt_path) in self.list_decommission_durable_ilm_receipts(source_pool_idx).await? { match delete_config(self.pools[pool_idx].clone(), &receipt_path).await { @@ -4759,9 +5078,15 @@ impl ECStore { .map_err(|err| Error::other(format!("source durable ILM record is invalid at path `{path}`: {err}")))?; let (target_pool_idx, target) = self .load_decommissioned_durable_ilm_target(source_pool_idx, path, namespace.max_record_size, &source_record.context()) - .await?; - validate_decommission_durable_ilm_copy(path, &source_record, &source, &target)?; - let receipt = DecommissionDurableIlmReceipt::new(path, &source_record, &target); + .await? + .ok_or_else(|| { + Error::other(format!( + "target durable ILM record is missing at path `{path}` {}", + source_record.context() + )) + })?; + let target_record = validate_decommission_durable_ilm_copy(path, &source_record, &target)?; + let receipt = DecommissionDurableIlmReceipt::new(path, &target_record); self.persist_decommission_durable_ilm_receipt(source_pool_idx, target_pool_idx, &receipt) .await?; @@ -4800,6 +5125,14 @@ impl ECStore { Ok(self.list_decommission_durable_ilm_receipts(source_pool_idx).await?.len()) } + #[cfg(test)] + pub(crate) async fn decommission_durable_ilm_receipt_paths_for_test( + &self, + source_pool_idx: usize, + ) -> Result> { + self.list_decommission_durable_ilm_receipts(source_pool_idx).await + } + #[cfg(test)] pub(crate) async fn cleanup_decommission_durable_ilm_receipts_for_test(&self, source_pool_idx: usize) -> Result<()> { self.cleanup_decommission_durable_ilm_receipts(source_pool_idx).await diff --git a/crates/ecstore/src/set_disk/ops/object.rs b/crates/ecstore/src/set_disk/ops/object.rs index cb8f70766..73463b6d3 100644 --- a/crates/ecstore/src/set_disk/ops/object.rs +++ b/crates/ecstore/src/set_disk/ops/object.rs @@ -37,7 +37,7 @@ use crate::bucket::lifecycle::{ transition_transaction::{ TransitionRemoteVersion, TransitionSourceIdentity, TransitionSourceVersionMode, TransitionTransaction, TransitionTransactionInit, TransitionTransactionState, delete_transition_transaction_record, - save_transition_transaction_record, + load_transition_transaction_record, save_transition_transaction_record, }, }; use crate::bucket::quota::reservation; @@ -4246,7 +4246,12 @@ fn record_transition_uploaded_save_attempt(transaction: &TransitionTransaction, async fn delete_transition_transaction_if_available(api: Option<&Arc>, transaction_id: Uuid) -> Result<()> { if let Some(api) = api { - return delete_transition_transaction_record(api.clone(), transaction_id).await; + let transaction = match load_transition_transaction_record(api.clone(), transaction_id).await { + Ok(transaction) => transaction, + Err(Error::ConfigNotFound) => return Ok(()), + Err(err) => return Err(err), + }; + return delete_transition_transaction_record(api.clone(), &transaction).await; } Ok(()) } diff --git a/crates/ecstore/src/store/init.rs b/crates/ecstore/src/store/init.rs index fb6dd9379..da97f3f63 100644 --- a/crates/ecstore/src/store/init.rs +++ b/crates/ecstore/src/store/init.rs @@ -3055,7 +3055,7 @@ mod tests { without_storage_class_env(build_isolated_test_store(temp_dir.path(), "durable-ilm-decommission", &[4, 4])).await; let tier_name = "DECOMMISSION-ILM"; - let backend = register_mock_tier(&ctx.tier_config_mgr(), tier_name).await; + let backend = register_transition_reconcile_test_tier(&ctx.tier_config_mgr(), tier_name).await; let backend_identity = TierConfigMgr::acquire_operation_lease(&ctx.tier_config_mgr(), tier_name) .await .expect("tier lease should resolve") @@ -3073,7 +3073,7 @@ mod tests { let tier_path = tier_delete_journal_object_name(&tier_entry); let tier_bytes = encode_tier_delete_journal_entry(&tier_entry).expect("tier journal should encode"); - let transaction = TransitionTransaction::new(TransitionTransactionInit { + let mut transaction = TransitionTransaction::new(TransitionTransactionInit { deployment_id: ctx.deployment_id().expect("test store should initialize deployment id"), transaction_id: uuid::Uuid::new_v4(), owner_epoch: uuid::Uuid::new_v4(), @@ -3093,6 +3093,10 @@ mod tests { not_after_unix_nanos: 1_780_000_000_000_000_000, }) .expect("transition transaction should build"); + let regressed_transaction = transaction.clone(); + transaction + .advance(transaction.fence(), TransitionTransactionState::UploadOutcomeUnknown, None) + .expect("transition transaction should advance before migration"); let transaction_path = transition_transaction_record_object_name(transaction.transaction_id) .expect("transition transaction path should build"); let transaction_bytes = transaction.encode().expect("transition transaction should encode"); @@ -3212,6 +3216,25 @@ mod tests { *expected ); } + assert_eq!( + store + .decommission_durable_ilm_receipt_count_for_test(0) + .await + .expect("no receipt should exist before the final sweep"), + 0 + ); + let isolated_tier_stats = recover_tier_delete_journal_entries(store.clone(), 100, None) + .await + .expect("tier recovery should retain a terminal record until its receipt is committed"); + assert!(isolated_tier_stats.scanned >= 1); + assert_eq!(isolated_tier_stats.deleted, 0); + assert!(isolated_tier_stats.failed >= 1); + assert_eq!( + com::read_config(store.pools[1].clone(), &tier_path) + .await + .expect("receipt isolation must retain the target tier journal"), + tier_bytes + ); com::delete_config(store.pools[1].clone(), &manual_job_path) .await @@ -3404,6 +3427,161 @@ mod tests { .await .expect("post-crash target transaction should restore"); + let mut wrong_manual_job = manual_job.clone(); + wrong_manual_job.job_id = uuid::Uuid::new_v4(); + com::save_config( + store.pools[1].clone(), + &manual_job_path, + wrong_manual_job.encode().expect("wrong-id job should encode"), + ) + .await + .expect("post-crash target manual job should accept the wrong-id fixture"); + let wrong_id_after_crash = store + .complete_decommission(0) + .await + .expect_err("completion must reject a target record with the wrong id") + .to_string(); + assert!(wrong_id_after_crash.contains(&manual_job_path)); + assert!(wrong_id_after_crash.contains(&manual_job_id.to_string())); + com::save_config(store.pools[1].clone(), &manual_job_path, manual_job_bytes.clone()) + .await + .expect("post-crash target manual job should restore after the wrong-id check"); + + com::save_config( + store.pools[1].clone(), + &transaction_path, + regressed_transaction + .encode() + .expect("regressed transition transaction should encode"), + ) + .await + .expect("post-crash target transaction should accept the regression fixture"); + let regression_after_crash = store + .complete_decommission(0) + .await + .expect_err("completion must reject a lower transition transaction revision") + .to_string(); + assert!(regression_after_crash.contains("generation mismatch")); + assert!(regression_after_crash.contains(&transaction_path)); + assert!(regression_after_crash.contains(&transaction.transaction_id.to_string())); + com::save_config(store.pools[1].clone(), &transaction_path, transaction_bytes.clone()) + .await + .expect("post-crash target transaction should restore after the regression check"); + + let (manual_task_receipt_pool, manual_task_receipt_path) = store + .decommission_durable_ilm_receipt_paths_for_test(0) + .await + .expect("durable ILM receipt paths should be listable") + .into_iter() + .find(|(_, path)| path.contains(&manual_task_path)) + .expect("manual task receipt should retain its reversible source path"); + let manual_task_receipt_bytes = + com::read_config(store.pools[manual_task_receipt_pool].clone(), &manual_task_receipt_path) + .await + .expect("manual task receipt should be readable before corruption"); + com::save_config( + store.pools[manual_task_receipt_pool].clone(), + &manual_task_receipt_path, + b"{corrupt".to_vec(), + ) + .await + .expect("manual task receipt should corrupt deterministically"); + let corrupt_receipt = store + .complete_decommission(0) + .await + .expect_err("completion must fail closed on a corrupt receipt") + .to_string(); + assert!(corrupt_receipt.contains(&manual_task_path)); + assert!(corrupt_receipt.contains(&manual_job_id.to_string())); + com::save_config( + store.pools[manual_task_receipt_pool].clone(), + &manual_task_receipt_path, + manual_task_receipt_bytes, + ) + .await + .expect("manual task receipt should restore after the corruption check"); + + let tier_stats = recover_tier_delete_journal_entries(store.clone(), 100, None) + .await + .expect("tier journal recovery should consume the migrated record before completion"); + assert_eq!((tier_stats.scanned, tier_stats.deleted, tier_stats.failed), (1, 1, 0)); + assert!(matches!(com::read_config(store.clone(), &tier_path).await, Err(Error::ConfigNotFound))); + + let recovered_transition_version = "recovered-transition-version".to_string(); + backend + .set_transition_candidate_probe_override(Some(TransitionCandidateProbe::VersionedPresent( + recovered_transition_version.clone(), + ))) + .await; + let transaction_stats = recover_transition_transaction_records(store.clone(), 100, None) + .await + .expect("transition recovery should advance and consume the migrated transaction before completion"); + backend.set_transition_candidate_probe_override(None).await; + assert_eq!( + ( + transaction_stats.scanned, + transaction_stats.recovered, + transaction_stats.retained, + transaction_stats.failed, + ), + (1, 1, 0, 0) + ); + assert!(matches!( + com::read_config(store.clone(), &transaction_path).await, + Err(Error::ConfigNotFound) + )); + com::save_config( + store.pools[1].clone(), + &transaction_path, + regressed_transaction + .encode() + .expect("post-terminal transition rollback should encode"), + ) + .await + .expect("target should accept the post-terminal rollback fixture"); + let post_terminal_regression = store + .complete_decommission(0) + .await + .expect_err("terminal proof must not mask a lower transition revision") + .to_string(); + assert!(post_terminal_regression.contains("generation mismatch")); + assert!(post_terminal_regression.contains(&transaction_path)); + assert!(post_terminal_regression.contains(&transaction.transaction_id.to_string())); + com::delete_config(store.pools[1].clone(), &transaction_path) + .await + .expect("post-terminal rollback fixture should be removed"); + + let manual_stats = recover_manual_transition_jobs_once(store.clone(), 100, None) + .await + .expect("manual recovery should advance the migrated job and consume its scope before completion"); + assert_eq!( + (manual_stats.scanned, manual_stats.resumed, manual_stats.skipped, manual_stats.failed,), + (1, 1, 0, 0) + ); + assert!(matches!( + com::read_config(store.clone(), &manual_scope_path).await, + Err(Error::ConfigNotFound) + )); + let recovered_manual_job_bytes = com::read_config(store.pools[1].clone(), &manual_job_path) + .await + .expect("manual recovery should retain the advanced job record"); + assert_ne!(recovered_manual_job_bytes, manual_job_bytes); + + com::save_config(store.pools[1].clone(), &manual_job_path, manual_job_bytes.clone()) + .await + .expect("target manual job should accept the rollback fixture"); + let manual_regression = store + .complete_decommission(0) + .await + .expect_err("completion must reject a manual job generation rollback") + .to_string(); + assert!(manual_regression.contains("generation mismatch")); + assert!(manual_regression.contains(&manual_job_path)); + assert!(manual_regression.contains(&manual_job_id.to_string())); + com::save_config(store.pools[1].clone(), &manual_job_path, recovered_manual_job_bytes) + .await + .expect("target manual job should restore its recovered generation"); + store .complete_decommission(0) .await @@ -3426,38 +3604,9 @@ mod tests { .cleanup_decommission_durable_ilm_receipts_for_test(0) .await .expect("receipt cleanup should be idempotent"); - - let tier_stats = recover_tier_delete_journal_entries(store.clone(), 100, None) - .await - .expect("tier journal recovery should consume the migrated record"); - assert_eq!((tier_stats.scanned, tier_stats.deleted, tier_stats.failed), (1, 1, 0)); - assert_eq!( - backend.remove_versions().await, - vec![(tier_entry.obj_name.clone(), tier_entry.version_id.clone())] - ); - let transaction_stats = recover_transition_transaction_records(store.clone(), 100, None) - .await - .expect("transition transaction recovery should read the migrated record"); - assert_eq!( - ( - transaction_stats.scanned, - transaction_stats.recovered, - transaction_stats.retained, - transaction_stats.failed, - ), - (1, 0, 1, 0) - ); - let manual_stats = recover_manual_transition_jobs_once(store.clone(), 100, None) - .await - .expect("manual recovery should reconcile migrated job, scope, task, and result records"); - assert_eq!( - (manual_stats.scanned, manual_stats.resumed, manual_stats.skipped, manual_stats.failed,), - (1, 1, 0, 0) - ); - assert!(matches!( - com::read_config(store.clone(), &manual_scope_path).await, - Err(Error::ConfigNotFound) - )); + let removed_versions = backend.remove_versions().await; + assert!(removed_versions.contains(&(tier_entry.obj_name.clone(), tier_entry.version_id.clone()))); + assert!(removed_versions.contains(&(transaction.remote_object.clone(), recovered_transition_version))); } #[cfg(feature = "test-util")]