From 15b8e8860fd9ba146deaca25a646f69bdbdf4cab Mon Sep 17 00:00:00 2001 From: overtrue Date: Sat, 22 Aug 2026 03:00:29 +0800 Subject: [PATCH] fix(ecstore): close ILM receipt recovery gaps --- .config/nextest.toml | 8 +- .../src/bucket/lifecycle/durable_namespace.rs | 284 +++++++++++++- crates/ecstore/src/core/pools.rs | 358 +++++++++++++++--- crates/ecstore/src/store/init.rs | 176 +++++++++ 4 files changed, 778 insertions(+), 48 deletions(-) diff --git a/.config/nextest.toml b/.config/nextest.toml index 4e1637705..a65ebe9fe 100644 --- a/.config/nextest.toml +++ b/.config/nextest.toml @@ -78,10 +78,10 @@ test-group = 'embedded-test-ports' filter = 'package(rustfs-ecstore) & test(manual_transition_page_checkpoint_persists_durable_job_progress)' test-group = 'ecstore-serial-flaky' -# The durable ILM decommission regression builds an isolated 8-disk store and -# deliberately takes target disks offline while checking read-quorum fencing. +# The durable ILM decommission regressions build isolated multi-pool stores and +# deliberately take source or target disks offline while checking fencing. [[profile.default.overrides]] -filter = 'package(rustfs-ecstore) & test(decommission_migrates_and_verifies_registered_durable_ilm_records)' +filter = 'package(rustfs-ecstore) & (test(decommission_migrates_and_verifies_registered_durable_ilm_records) | test(decommission_durable_ilm_target_read_error_is_not_masked_by_peer_success) | test(decommission_durable_ilm_terminal_receipt_recovers_failed_source_cleanup))' test-group = 'ecstore-serial-flaky' # Serialize the bucket-incarnation / lifecycle-fence tests. They drive @@ -197,7 +197,7 @@ filter = 'package(rustfs-ecstore) & test(manual_transition_page_checkpoint_persi test-group = 'ecstore-serial-flaky' [[profile.ci.overrides]] -filter = 'package(rustfs-ecstore) & test(decommission_migrates_and_verifies_registered_durable_ilm_records)' +filter = 'package(rustfs-ecstore) & (test(decommission_migrates_and_verifies_registered_durable_ilm_records) | test(decommission_durable_ilm_target_read_error_is_not_masked_by_peer_success) | test(decommission_durable_ilm_terminal_receipt_recovers_failed_source_cleanup))' test-group = 'ecstore-serial-flaky' # Serialize the bucket-incarnation / lifecycle-fence tests under the ci profile diff --git a/crates/ecstore/src/bucket/lifecycle/durable_namespace.rs b/crates/ecstore/src/bucket/lifecycle/durable_namespace.rs index 08c6c57ca..e9d296c2f 100644 --- a/crates/ecstore/src/bucket/lifecycle/durable_namespace.rs +++ b/crates/ecstore/src/bucket/lifecycle/durable_namespace.rs @@ -16,7 +16,12 @@ use rustfs_utils::crypto::hex_sha256; use serde::{Deserialize, Serialize}; use uuid::Uuid; -use super::{manual_transition_job, tier_delete_journal, transition_transaction}; +use super::{ + bucket_lifecycle_ops::{ + ManualTransitionQueueSnapshot, ManualTransitionRunReport, decode_manual_transition_continuation_token, + }, + manual_transition_job, tier_delete_journal, transition_transaction, +}; use crate::error::{Error, Result}; pub(crate) const ILM_META_PREFIX: &str = "ilm"; @@ -94,6 +99,13 @@ pub(crate) struct ValidatedDurableIlmRecord { pub(crate) checkpoint: DurableIlmRecordCheckpoint, } +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub(crate) struct ManualTransitionJobProgressCheckpoint { + report: ManualTransitionRunReport, + queue_snapshot: ManualTransitionQueueSnapshot, +} + impl ValidatedDurableIlmRecord { pub(crate) fn context(&self) -> String { format!("namespace `{}` {} `{}`", self.namespace, self.id_kind, self.id) @@ -123,6 +135,8 @@ pub(crate) enum DurableIlmRecordCheckpoint { state: manual_transition_job::ManualTransitionJobState, scan_completed: bool, cancel_requested: bool, + #[serde(default, skip_serializing_if = "Option::is_none")] + progress: Option>, }, ManualTransitionScope { content_sha256: String, @@ -151,6 +165,14 @@ impl DurableIlmRecordCheckpoint { pub(crate) fn validate_successor(&self, next: &Self) -> Result<()> { if self == next { + if let Self::ManualTransitionJob { + progress: Some(progress), + .. + } = self + && !manual_job_progress_is_valid(progress) + { + return Err(Error::other("durable ILM manual transition checkpoint is invalid")); + } return Ok(()); } @@ -200,6 +222,7 @@ impl DurableIlmRecordCheckpoint { state: previous_state, scan_completed: previous_scan_completed, cancel_requested: previous_cancel_requested, + progress: previous_progress, .. }, Self::ManualTransitionJob { @@ -208,6 +231,7 @@ impl DurableIlmRecordCheckpoint { state: next_state, scan_completed: next_scan_completed, cancel_requested: next_cancel_requested, + progress: next_progress, .. }, ) => { @@ -216,6 +240,7 @@ impl DurableIlmRecordCheckpoint { && manual_job_state_reaches(*previous_state, *next_state) && (!previous_scan_completed || *next_scan_completed) && (!previous_cancel_requested || *next_cancel_requested) + && manual_job_progress_reaches(previous_progress.as_deref(), next_progress.as_deref(), *next_scan_completed) } ( Self::ManualTransitionScope { @@ -269,6 +294,129 @@ fn manual_job_state_reaches( from == to || from == manual_transition_job::ManualTransitionJobState::Running } +fn manual_job_progress_reaches( + previous: Option<&ManualTransitionJobProgressCheckpoint>, + next: Option<&ManualTransitionJobProgressCheckpoint>, + next_scan_completed: bool, +) -> bool { + let (previous, next) = match (previous, next) { + (None, Some(next)) => return manual_job_progress_is_valid(next), + (Some(previous), Some(next)) => (previous, next), + _ => return false, + }; + let previous_report = &previous.report; + let next_report = &next.report; + + macro_rules! counters_do_not_regress { + ($($field:ident),+ $(,)?) => { + $(previous_report.$field <= next_report.$field)&&+ + }; + } + + let counters_monotonic = counters_do_not_regress!( + scanned, + eligible, + enqueued, + dry_run_eligible, + skipped_not_transition, + skipped_tier, + skipped_delete_marker, + skipped_directory, + skipped_replication, + skipped_already_transitioned, + skipped_already_in_flight, + skipped_queue_full, + skipped_queue_closed, + skipped_queue_timeout, + transition_completed, + transition_failed, + tier_failure, + ); + let failure_reasons_monotonic = previous_report.tier_failure_by_reason.iter().all(|(reason, previous_count)| { + next_report + .tier_failure_by_reason + .get(reason) + .is_some_and(|next_count| next_count >= previous_count) + }); + let flags_monotonic = (!previous_report.lifecycle_config_found || next_report.lifecycle_config_found) + && (!previous_report.truncated_by_limit || next_report.truncated_by_limit) + && (!previous_report.truncated_by_duration || next_report.truncated_by_duration) + && (!previous_report.cancelled || next_report.cancelled); + let cursor_monotonic = manual_job_cursor_reaches(previous_report, next_report, next_scan_completed); + let progress_valid = manual_job_progress_is_valid(previous) && manual_job_progress_is_valid(next); + + previous_report.bucket == next_report.bucket + && previous_report.prefix == next_report.prefix + && previous_report.tier == next_report.tier + && previous_report.dry_run == next_report.dry_run + && counters_monotonic + && failure_reasons_monotonic + && flags_monotonic + && cursor_monotonic + && progress_valid +} + +fn manual_job_progress_is_valid(progress: &ManualTransitionJobProgressCheckpoint) -> bool { + manual_job_worker_results_are_valid(&progress.report) + && manual_job_queue_snapshot_is_valid(&progress.queue_snapshot) + && manual_job_cursor_is_valid(progress.report.continuation_token.as_deref()) +} + +fn manual_job_worker_results_are_valid(report: &ManualTransitionRunReport) -> bool { + let reason_total = report + .tier_failure_by_reason + .values() + .try_fold(0u64, |total, count| total.checked_add(*count)); + report + .transition_completed + .checked_add(report.transition_failed) + .is_some_and(|total| total <= report.enqueued) + && report.transition_failed <= report.tier_failure + && reason_total.is_some_and(|total| total <= report.tier_failure) +} + +fn manual_job_cursor_reaches( + previous: &ManualTransitionRunReport, + next: &ManualTransitionRunReport, + next_scan_completed: bool, +) -> bool { + if previous.continuation_token == next.continuation_token { + return manual_job_cursor_is_valid(previous.continuation_token.as_deref()); + } + match (&previous.continuation_token, &next.continuation_token) { + (None, Some(next_token)) => next.scanned > previous.scanned && manual_job_cursor_is_valid(Some(next_token)), + (Some(_), None) => next_scan_completed, + (Some(previous_token), Some(next_token)) if next.scanned > previous.scanned => { + let (Ok((Some(previous_marker), previous_version)), Ok((Some(next_marker), next_version))) = ( + decode_manual_transition_continuation_token(previous_token), + decode_manual_transition_continuation_token(next_token), + ) else { + return false; + }; + next_marker > previous_marker + || (next_marker == previous_marker + && previous_version.is_some() + && next_version.is_some() + && previous_version != next_version) + } + _ => false, + } +} + +fn manual_job_cursor_is_valid(token: Option<&str>) -> bool { + let Some(token) = token else { + return true; + }; + matches!(decode_manual_transition_continuation_token(token), Ok((Some(_), _))) +} + +fn manual_job_queue_snapshot_is_valid(snapshot: &ManualTransitionQueueSnapshot) -> bool { + (snapshot.queue_capacity > 0 || snapshot.queued == 0) + && (snapshot.queue_capacity == 0 || snapshot.queued <= snapshot.queue_capacity) + && (snapshot.workers > 0 || snapshot.active == 0) + && (snapshot.workers == 0 || snapshot.active <= snapshot.workers) +} + fn checkpoint_hash(value: &T) -> Result { let encoded = serde_json::to_vec(value).map_err(Error::other)?; Ok(hex_sha256(&encoded, ToOwned::to_owned)) @@ -429,6 +577,10 @@ pub(crate) fn validate_durable_ilm_record(path: &str, data: &[u8]) -> Result Result DurableIlmRecordCheckpoint { + let path = + manual_transition_job::manual_transition_job_record_object_name(job.job_id).expect("manual job path should build"); + let encoded = job.encode().expect("manual job should encode"); + validate_durable_ilm_record(&path, &encoded) + .expect("manual job checkpoint should validate") + .checkpoint + } + + fn continuation_token(marker: &str) -> String { + let encoded = serde_json::to_vec(&serde_json::json!({ "marker": marker, "version_marker": null })) + .expect("continuation token should encode"); + base64_simd::URL_SAFE_NO_PAD.encode_to_string(&encoded) + } + #[test] fn unknown_ilm_record_requires_namespace_registration() { let err = classify_durable_ilm_record("ilm/future-durable/jobs/one.json") @@ -524,4 +691,119 @@ mod tests { } } } + + #[test] + fn manual_transition_job_checkpoint_rejects_progress_poison() { + let options = super::super::bucket_lifecycle_ops::ManualTransitionRunOptions::default(); + let mut initial = + manual_transition_job::ManualTransitionJobRecord::new(Uuid::new_v4(), "manual-checkpoint-bucket", &options, "owner"); + let initial_checkpoint = manual_job_checkpoint(&initial); + initial.updated_at_unix_nanos += 1; + initial.report.scanned = 1; + initial.report.continuation_token = Some(continuation_token("logs/a")); + let first_page_checkpoint = manual_job_checkpoint(&initial); + initial_checkpoint + .validate_successor(&first_page_checkpoint) + .expect("the first durable cursor should advance from no cursor"); + + let mut legacy_checkpoint = initial_checkpoint; + let DurableIlmRecordCheckpoint::ManualTransitionJob { progress, .. } = &mut legacy_checkpoint else { + panic!("manual job should produce a manual checkpoint"); + }; + *progress = None; + legacy_checkpoint + .validate_successor(&first_page_checkpoint) + .expect("legacy checkpoints should upgrade to validated progress"); + + let mut previous = initial; + previous.updated_at_unix_nanos += 1; + previous.report.scanned = 10; + previous.report.eligible = 8; + previous.report.enqueued = 2; + previous.report.transition_completed = 1; + previous.report.continuation_token = Some(continuation_token("logs/b")); + previous.queue_snapshot = ManualTransitionQueueSnapshot { + queue_capacity: 10, + queued: 1, + active: 1, + workers: 2, + queue_full: 2, + queue_send_timeout: 1, + ..Default::default() + }; + let previous_checkpoint = manual_job_checkpoint(&previous); + + let mut next = previous.clone(); + next.updated_at_unix_nanos += 1; + next.report.scanned = 11; + next.report.eligible = 9; + next.report.transition_completed = 2; + next.report.continuation_token = Some(continuation_token("logs/c")); + next.queue_snapshot.queued = 0; + next.queue_snapshot.active = 0; + next.queue_snapshot.queue_full = 3; + let next_checkpoint = manual_job_checkpoint(&next); + previous_checkpoint + .validate_successor(&next_checkpoint) + .expect("forward job progress should validate"); + + let mut counter_rollback = next.clone(); + counter_rollback.updated_at_unix_nanos += 1; + counter_rollback.report.scanned = 9; + assert!( + previous_checkpoint + .validate_successor(&manual_job_checkpoint(&counter_rollback)) + .is_err() + ); + + let mut cursor_rollback = next.clone(); + cursor_rollback.updated_at_unix_nanos += 1; + cursor_rollback.report.scanned = previous.report.scanned; + cursor_rollback.report.scanned += 1; + cursor_rollback.report.continuation_token = Some(continuation_token("logs/a")); + assert!( + previous_checkpoint + .validate_successor(&manual_job_checkpoint(&cursor_rollback)) + .is_err() + ); + + let mut worker_result_rollback = next.clone(); + worker_result_rollback.updated_at_unix_nanos += 1; + worker_result_rollback.report.transition_completed = 0; + assert!( + previous_checkpoint + .validate_successor(&manual_job_checkpoint(&worker_result_rollback)) + .is_err() + ); + + let mut worker_result_overflow = next.clone(); + worker_result_overflow.updated_at_unix_nanos += 1; + worker_result_overflow.report.enqueued = u64::MAX; + worker_result_overflow.report.transition_completed = u64::MAX; + worker_result_overflow.report.transition_failed = 1; + worker_result_overflow.report.tier_failure = 1; + assert!( + previous_checkpoint + .validate_successor(&manual_job_checkpoint(&worker_result_overflow)) + .is_err() + ); + + let mut invalid_cursor = next.clone(); + invalid_cursor.updated_at_unix_nanos += 1; + invalid_cursor.report.continuation_token = Some("not-base64".to_string()); + assert!( + previous_checkpoint + .validate_successor(&manual_job_checkpoint(&invalid_cursor)) + .is_err() + ); + + let mut queue_state_poison = next; + queue_state_poison.updated_at_unix_nanos += 1; + queue_state_poison.queue_snapshot.queued = queue_state_poison.queue_snapshot.queue_capacity + 1; + assert!( + previous_checkpoint + .validate_successor(&manual_job_checkpoint(&queue_state_poison)) + .is_err() + ); + } } diff --git a/crates/ecstore/src/core/pools.rs b/crates/ecstore/src/core/pools.rs index d12bad195..d826c2d5b 100644 --- a/crates/ecstore/src/core/pools.rs +++ b/crates/ecstore/src/core/pools.rs @@ -924,6 +924,65 @@ impl DecommissionDurableIlmReceipt { } } +fn merge_decommission_durable_ilm_receipts( + existing: &DecommissionDurableIlmReceipt, + incoming: &DecommissionDurableIlmReceipt, +) -> Result { + if existing.source_path != incoming.source_path + || existing.namespace != incoming.namespace + || existing.id_kind != incoming.id_kind + || existing.id != incoming.id + { + return Err(Error::other(format!( + "durable ILM receipt identity conflict for source path `{}` {}; incoming {}", + existing.source_path, + existing.context(), + incoming.context() + ))); + } + + let checkpoint = + if existing.checkpoint == incoming.checkpoint || incoming.checkpoint.validate_successor(&existing.checkpoint).is_ok() { + existing.checkpoint.clone() + } else { + existing.checkpoint.validate_successor(&incoming.checkpoint).map_err(|err| { + Error::other(format!( + "durable ILM receipt checkpoint conflict for source path `{}` {}: {err}", + existing.source_path, + existing.context() + )) + })?; + incoming.checkpoint.clone() + }; + let terminal_checkpoint = match (&existing.terminal_checkpoint, &incoming.terminal_checkpoint) { + (Some(existing), Some(incoming)) if existing == incoming => Some(existing.clone()), + (Some(existing), Some(incoming)) if incoming.validate_successor(existing).is_ok() => Some(existing.clone()), + (Some(existing), Some(incoming)) => { + existing.validate_successor(incoming).map_err(|err| { + Error::other(format!( + "durable ILM receipt terminal checkpoint conflict for source path `{}` {}: {err}", + existing.source_path, + existing.context() + )) + })?; + Some(incoming.clone()) + } + (Some(existing), None) => Some(existing.clone()), + (None, Some(incoming)) => Some(incoming.clone()), + (None, None) => None, + }; + let merged = DecommissionDurableIlmReceipt { + source_path: existing.source_path.clone(), + namespace: existing.namespace.clone(), + id_kind: existing.id_kind.clone(), + id: existing.id.clone(), + checkpoint, + terminal_checkpoint, + }; + merged.validate()?; + Ok(merged) +} + #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] #[serde(deny_unknown_fields)] struct PersistedDecommissionDurableIlmReceipt { @@ -946,8 +1005,9 @@ impl DecommissionDurableIlmReceiptLocator { } } -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_token(cmd_line: &str, start_time: OffsetDateTime) -> String { + let identity = format!("{cmd_line}\0{}", start_time.unix_timestamp_nanos()); + hex_sha256(identity.as_bytes(), ToOwned::to_owned) } fn decommission_durable_ilm_receipt_run_prefix(run_token: &str) -> String { @@ -4475,13 +4535,16 @@ impl ECStore { 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 + let pool = pool_meta .pools .get(source_pool_idx) - .ok_or_else(|| invalid_decommission_pool_index_error(pool_meta.pools.len(), source_pool_idx))? - .cmd_line - .clone(); - Ok(decommission_durable_ilm_receipt_run_token(&cmd_line)) + .ok_or_else(|| invalid_decommission_pool_index_error(pool_meta.pools.len(), source_pool_idx))?; + let start_time = pool + .decommission + .as_ref() + .and_then(|info| info.start_time) + .ok_or_else(|| Error::other(format!("decommission run identity is missing for pool {source_pool_idx}")))?; + Ok(decommission_durable_ilm_receipt_run_token(&pool.cmd_line, start_time)) } async fn load_decommissioned_durable_ilm_target( @@ -4522,11 +4585,10 @@ impl ECStore { } } - match (target, first_read_error) { - (Some(target), _) => Ok(Some(target)), - (None, Some(err)) => Err(err), - (None, None) => Ok(None), + if let Some(err) = first_read_error { + return Err(err); } + Ok(target) } async fn list_decommission_durable_ilm_receipt_paths_in_pool(&self, pool_idx: usize, prefix: &str) -> Result> { @@ -4591,22 +4653,98 @@ impl ECStore { ) -> Result<()> { 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}", - receipt.source_path, - receipt.context() - )) - })?; - save_config(self.pools[target_pool_idx].clone(), &receipt_path, encoded) + let locator = parse_decommission_durable_ilm_receipt_path(&receipt_path)?; + let mut attempt = 1; + loop { + let (merged, http_preconditions) = match read_config_limited_preserve_empty_with_metadata( + self.pools[target_pool_idx].clone(), + &receipt_path, + DECOMMISSION_DURABLE_ILM_RECEIPT_MAX_SIZE, + ) .await - .map_err(|err| { + { + Ok((existing_data, metadata)) => { + let existing = DecommissionDurableIlmReceipt::decode(&existing_data).map_err(|err| { + Error::other(format!( + "durable ILM decommission receipt `{receipt_path}` in pool {target_pool_idx} for {} is invalid: {err}", + locator.context() + )) + })?; + Self::validate_decommission_durable_ilm_receipt_locator(&receipt_path, &locator, &existing)?; + let merged = merge_decommission_durable_ilm_receipts(&existing, receipt)?; + if merged == existing { + return Ok(()); + } + let etag = metadata.etag.filter(|etag| !etag.trim().is_empty()).ok_or_else(|| { + Error::other(format!( + "durable ILM decommission receipt `{receipt_path}` in pool {target_pool_idx} is missing an ETag" + )) + })?; + ( + merged, + HTTPPreconditions { + if_match: Some(etag), + ..Default::default() + }, + ) + } + Err(err) + if matches!(&err, Error::ConfigNotFound | Error::FileNotFound | Error::FileVersionNotFound) + || is_err_object_not_found(&err) + || is_err_version_not_found(&err) => + { + ( + receipt.clone(), + HTTPPreconditions { + if_none_match: Some("*".to_string()), + ..Default::default() + }, + ) + } + Err(err) => { + return Err(Error::other(format!( + "failed to read durable ILM decommission receipt `{receipt_path}` from pool {target_pool_idx} for {}: {err}", + locator.context() + ))); + } + }; + let encoded = merged.encode().map_err(|err| { Error::other(format!( - "failed to persist durable ILM decommission receipt `{receipt_path}` for source path `{}` {}: {err}", + "failed to encode durable ILM decommission receipt `{receipt_path}` for source path `{}` {}: {err}", receipt.source_path, receipt.context() )) - }) + })?; + match save_config_with_opts( + self.pools[target_pool_idx].clone(), + &receipt_path, + encoded, + &ObjectOptions { + max_parity: true, + http_preconditions: Some(http_preconditions), + ..Default::default() + }, + ) + .await + { + Ok(()) => return Ok(()), + Err(Error::PreconditionFailed) if attempt < DECOMMISSION_DURABLE_ILM_RECEIPT_CAS_ATTEMPTS => { + attempt += 1; + } + Err(Error::PreconditionFailed) => { + return Err(Error::other(format!( + "failed to persist durable ILM decommission receipt `{receipt_path}` for {} after concurrent updates", + locator.context() + ))); + } + Err(err) => { + return Err(Error::other(format!( + "failed to persist durable ILM decommission receipt `{receipt_path}` for {}: {err}", + locator.context() + ))); + } + } + } } fn validate_decommission_durable_ilm_receipt_locator( @@ -4652,6 +4790,78 @@ impl ECStore { Ok(receipt) } + async fn load_decommission_durable_ilm_terminal_receipt( + &self, + source_pool_idx: usize, + path: &str, + source_record: &ValidatedDurableIlmRecord, + ) -> Result> { + let run_token = self.durable_ilm_receipt_run_token(source_pool_idx).await?; + let receipt_path = decommission_durable_ilm_receipt_path(&run_token, path, source_record.id_kind, &source_record.id); + let locator = parse_decommission_durable_ilm_receipt_path(&receipt_path)?; + let mut proof = None::; + for pool_idx in 0..self.pools.len() { + if pool_idx == source_pool_idx { + continue; + } + let data = match read_config_limited_preserve_empty( + self.pools[pool_idx].clone(), + &receipt_path, + DECOMMISSION_DURABLE_ILM_RECEIPT_MAX_SIZE, + ) + .await + { + Ok(data) => data, + Err(err) + if matches!(&err, Error::ConfigNotFound | Error::FileNotFound | Error::FileVersionNotFound) + || is_err_object_not_found(&err) + || is_err_version_not_found(&err) => + { + continue; + } + Err(err) => { + return Err(Error::other(format!( + "failed to read terminal durable ILM decommission receipt `{receipt_path}` from pool {pool_idx} for {}: {err}", + source_record.context() + ))); + } + }; + let receipt = DecommissionDurableIlmReceipt::decode(&data).map_err(|err| { + Error::other(format!( + "terminal durable ILM decommission receipt `{receipt_path}` in pool {pool_idx} for {} is invalid: {err}", + source_record.context() + )) + })?; + Self::validate_decommission_durable_ilm_receipt_locator(&receipt_path, &locator, &receipt)?; + if receipt.namespace != source_record.namespace + || receipt.id_kind != source_record.id_kind + || receipt.id != source_record.id + { + return Err(Error::other(format!( + "terminal durable ILM decommission receipt identity mismatch at path `{path}` {}; receipt {}", + source_record.context(), + receipt.context() + ))); + } + source_record + .checkpoint + .validate_successor(&receipt.checkpoint) + .map_err(|err| { + Error::other(format!( + "terminal durable ILM decommission receipt does not cover source at path `{path}` {}: {err}", + source_record.context() + )) + })?; + if receipt.terminal_checkpoint.is_some() { + proof = Some(match proof { + Some(existing) => merge_decommission_durable_ilm_receipts(&existing, &receipt)?, + None => receipt, + }); + } + } + Ok(proof) + } + async fn verify_decommission_durable_ilm_receipts(&self, source_pool_idx: usize) -> Result<()> { for (receipt_pool_idx, receipt_path) in self.list_decommission_durable_ilm_receipts(source_pool_idx).await? { let receipt = self @@ -4850,12 +5060,13 @@ impl ECStore { .pools .iter() .enumerate() - .filter(|(_, pool)| { + .filter_map(|(pool_idx, pool)| { pool.decommission .as_ref() - .is_some_and(|info| info.has_decommission_state() && !info.complete) + .filter(|info| info.has_decommission_state() && !info.complete) + .and_then(|info| info.start_time) + .map(|start_time| (pool_idx, decommission_durable_ilm_receipt_run_token(&pool.cmd_line, start_time))) }) - .map(|(pool_idx, pool)| (pool_idx, decommission_durable_ilm_receipt_run_token(&pool.cmd_line))) .collect::>() }; if active_runs.is_empty() { @@ -4926,19 +5137,24 @@ impl ECStore { .map_err(|err| Error::other(format!("failed to read source durable ILM record at path `{path}`: {err}")))?; let source_record = validate_durable_ilm_record(path, &source) .map_err(|err| Error::other(format!("source durable ILM record is invalid at path `{path}`: {err}")))?; - let (target_pool_idx, target) = self + let target = self .load_decommissioned_durable_ilm_target(source_pool_idx, path, namespace.max_record_size, &source_record.context()) - .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?; + if let Some((target_pool_idx, target)) = target { + 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?; + } else { + self.load_decommission_durable_ilm_terminal_receipt(source_pool_idx, path, &source_record) + .await? + .ok_or_else(|| { + Error::other(format!( + "target durable ILM record is missing at path `{path}` {} without a matching terminal receipt", + source_record.context() + )) + })?; + } let cleanup_result = data_movement::cleanup_source_entry_if_unchanged( source_set, @@ -6125,10 +6341,12 @@ pub(crate) fn fallback_free_capacity_dedup(disks: &[rustfs_madmin::Disk]) -> usi mod pools_tests { use super::{ DECOMMISSION_META_PREFIXES, DECOMMISSION_PROGRESS_SAVE_INTERVAL, DECOMMISSION_PROGRESS_SAVE_ITEM_THRESHOLD, - DecomBucketInfo, DecommissionStartPoolState, DecommissionTerminalState, ListCallback, PoolDecommissionInfo, PoolMeta, - PoolSpaceInfo, PoolStatus, apply_decommission_status_space_info, bind_decommission_cancelers, - bind_missing_decommission_cancelers, cancel_decommission_canceler, classify_decommission_terminal_state, - count_decommission_item, decommission_cancel_signal_result, decommission_item_size, decommission_meta_bucket_options, + DecomBucketInfo, DecommissionDurableIlmReceipt, DecommissionStartPoolState, DecommissionTerminalState, ListCallback, + PoolDecommissionInfo, PoolMeta, PoolSpaceInfo, PoolStatus, apply_decommission_status_space_info, + bind_decommission_cancelers, bind_missing_decommission_cancelers, cancel_decommission_canceler, + classify_decommission_terminal_state, count_decommission_item, decommission_cancel_signal_result, + decommission_durable_ilm_receipt_path, decommission_durable_ilm_receipt_run_prefix, + decommission_durable_ilm_receipt_run_token, decommission_item_size, decommission_meta_bucket_options, decommission_start_pool_state, dedup_indices, default_decommission_bucket_concurrency, ensure_decommission_cancel_allowed, ensure_decommission_clear_allowed, ensure_decommission_listing_disks_available, ensure_decommission_not_rebalancing, ensure_decommission_start_allowed, ensure_decommission_start_keeps_active_pool, @@ -6138,9 +6356,9 @@ mod pools_tests { ensure_valid_decommission_pool_index, first_resumable_decommission_queue_indices, get_by_index, has_active_decommission_canceler, is_decommission_active, is_decommission_cancel_requested, load_decommission_entry_versions, local_decommission_queue_prefix, mark_decommission_bucket_done, - merge_pool_status_refresh, missing_decommission_worker_prefix, observe_decommission_terminal_reload_result, - pool_meta_has_active_decommission, reconcile_decommission_meta_buckets, require_decommission_store, - resolve_decommission_bucket_done_save_result, resolve_decommission_bucket_state, + merge_decommission_durable_ilm_receipts, merge_pool_status_refresh, missing_decommission_worker_prefix, + observe_decommission_terminal_reload_result, pool_meta_has_active_decommission, reconcile_decommission_meta_buckets, + require_decommission_store, resolve_decommission_bucket_done_save_result, resolve_decommission_bucket_state, resolve_decommission_check_after_list_result, resolve_decommission_entry_cleanup_delete_result, resolve_decommission_entry_exact_versions, resolve_decommission_entry_reload_result, resolve_decommission_listing_worker_result, resolve_decommission_optional_bucket_config_result, @@ -6157,6 +6375,7 @@ mod pools_tests { validate_start_decommission_request, wait_decommission_listing_retry, wait_decommission_worker_drain, with_decommission_entry_context, }; + use crate::bucket::lifecycle::DurableIlmRecordCheckpoint; use crate::data_movement; use crate::disk::endpoint::Endpoint; use crate::error::{Error, StorageError}; @@ -6203,6 +6422,59 @@ mod pools_tests { } } + #[test] + fn decommission_receipt_run_token_changes_with_persisted_start_time() { + let first = OffsetDateTime::from_unix_timestamp(1_000).expect("first run timestamp should be valid"); + let second = OffsetDateTime::from_unix_timestamp(2_000).expect("second run timestamp should be valid"); + let first_token = decommission_durable_ilm_receipt_run_token("pool-0", first); + let second_token = decommission_durable_ilm_receipt_run_token("pool-0", second); + + assert_ne!(first_token, second_token); + assert_eq!(first_token, decommission_durable_ilm_receipt_run_token("pool-0", first)); + let operation_id = "a".repeat(64); + let old_receipt = decommission_durable_ilm_receipt_path( + &first_token, + &format!("ilm/tier-delete-journal/{operation_id}.json"), + "operation_id", + &operation_id, + ); + assert!(!old_receipt.starts_with(&decommission_durable_ilm_receipt_run_prefix(&second_token))); + } + + #[test] + fn decommission_receipt_merge_preserves_terminal_proof() { + let operation_id = "a".repeat(64); + let source_path = format!("ilm/tier-delete-journal/{operation_id}.json"); + let checkpoint = DurableIlmRecordCheckpoint::TierDeleteJournal { + content_sha256: "b".repeat(64), + identity_sha256: "c".repeat(64), + committed: false, + }; + let terminal_checkpoint = DurableIlmRecordCheckpoint::TierDeleteJournal { + content_sha256: "d".repeat(64), + identity_sha256: "c".repeat(64), + committed: true, + }; + let incoming = DecommissionDurableIlmReceipt { + source_path, + namespace: "tier-delete-journal".to_string(), + id_kind: "operation_id".to_string(), + id: operation_id, + checkpoint: checkpoint.clone(), + terminal_checkpoint: None, + }; + let existing = DecommissionDurableIlmReceipt { + terminal_checkpoint: Some(terminal_checkpoint.clone()), + ..incoming.clone() + }; + + let merged = merge_decommission_durable_ilm_receipts(&existing, &incoming) + .expect("retry receipt must merge with a terminal receipt"); + + assert_eq!(merged.checkpoint, checkpoint); + assert_eq!(merged.terminal_checkpoint, Some(terminal_checkpoint)); + } + #[test] fn test_apply_decommission_status_space_info_adds_idle_pool_usage() { let status = apply_decommission_status_space_info( diff --git a/crates/ecstore/src/store/init.rs b/crates/ecstore/src/store/init.rs index da97f3f63..47fb33a2b 100644 --- a/crates/ecstore/src/store/init.rs +++ b/crates/ecstore/src/store/init.rs @@ -585,6 +585,7 @@ mod tests { client::transition_api::ReaderImpl, config::com, core::pools::DecomBucketInfo, + data_movement::SourceCleanupDeleteBarrier, disk::{BUCKET_META_PREFIX, RUSTFS_META_BUCKET, STORAGE_FORMAT_FILE}, runtime::{global::set_object_store_resolver, sources as runtime_sources}, services::tier::{ @@ -3046,6 +3047,181 @@ mod tests { )); } + #[cfg(feature = "test-util")] + #[tokio::test] + #[serial_test::serial(storage_class_env)] + async fn decommission_durable_ilm_target_read_error_is_not_masked_by_peer_success() { + let temp_dir = tempfile::tempdir().expect("create temp store dir"); + let (_ctx, store, _shutdown) = + without_storage_class_env(build_isolated_test_store(temp_dir.path(), "durable-ilm-target-read-error", &[4, 4, 4])) + .await; + let job_id = uuid::Uuid::new_v4(); + let job = + ManualTransitionJobRecord::new(job_id, "manual-target-read-error", &ManualTransitionRunOptions::default(), "owner"); + let path = manual_transition_job_record_object_name(job_id).expect("manual job path should build"); + let data = job.encode().expect("manual job should encode"); + for pool in &store.pools { + com::save_config(pool.clone(), &path, data.clone()) + .await + .expect("manual job fixture should persist in every pool"); + } + store.pool_meta.write().await.pools[0].decommission = Some(PoolDecommissionInfo { + start_time: Some(OffsetDateTime::now_utc()), + ..Default::default() + }); + + let failing_target = store.pools[2].get_disks_by_key(&path); + let original_disks = { + let mut disks = failing_target.disks.write().await; + let original = disks.clone(); + for disk in disks.iter_mut().take(3) { + *disk = None; + } + original + }; + let error = store + .verify_and_cleanup_decommissioned_durable_ilm_record_for_test(0, store.pools[0].get_disks_by_key(&path), &path) + .await + .expect_err("one target read-quorum error must fail closed despite another target success") + .to_string(); + *failing_target.disks.write().await = original_disks; + + assert!(error.contains(&path)); + assert!(error.contains("pool 2")); + assert_eq!( + com::read_config(store.pools[0].clone(), &path) + .await + .expect("target read error must retain the source"), + data + ); + assert_eq!( + store + .decommission_durable_ilm_receipt_count_for_test(0) + .await + .expect("failed target verification should not create a receipt"), + 0 + ); + } + + #[cfg(feature = "test-util")] + #[tokio::test] + #[serial_test::serial(storage_class_env)] + async fn decommission_durable_ilm_terminal_receipt_recovers_failed_source_cleanup() { + let temp_dir = tempfile::tempdir().expect("create temp store dir"); + let (ctx, store, _shutdown) = + without_storage_class_env(build_isolated_test_store(temp_dir.path(), "durable-ilm-terminal-receipt", &[4, 4])).await; + let tier_name = "DECOMMISSION-RECEIPT"; + 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") + .backend_identity(); + let entry = Jentry { + obj_name: "receipt-recovery-object".to_string(), + version_id: "receipt-recovery-version".to_string(), + tier_name: tier_name.to_string(), + backend_identity: Some(backend_identity), + version_id_exact: true, + version_state: rustfs_filemeta::TransitionVersionState::Exact, + state: TierDeleteJournalState::Committed, + source: None, + }; + let path = tier_delete_journal_object_name(&entry); + let data = encode_tier_delete_journal_entry(&entry).expect("tier journal should encode"); + com::save_config(store.pools[0].clone(), &path, data.clone()) + .await + .expect("source tier journal should persist"); + com::save_config(store.pools[1].clone(), &path, data.clone()) + .await + .expect("target tier journal should persist"); + let active_pool_meta = { + let mut pool_meta = store.pool_meta.write().await; + pool_meta.pools[0].decommission = Some(PoolDecommissionInfo { + start_time: Some(OffsetDateTime::now_utc()), + ..Default::default() + }); + pool_meta.clone() + }; + active_pool_meta + .save(store.pools.clone()) + .await + .expect("active decommission run identity should persist"); + + let source_set = store.pools[0].get_disks_by_key(&path); + let barrier = SourceCleanupDeleteBarrier::install(RUSTFS_META_BUCKET, &path); + let cleanup_store = store.clone(); + let cleanup_set = source_set.clone(); + let cleanup_path = path.clone(); + let cleanup = tokio::spawn(async move { + cleanup_store + .verify_and_cleanup_decommissioned_durable_ilm_record_for_test(0, cleanup_set, &cleanup_path) + .await + }); + barrier.wait_until_paused().await; + let original_source_disks = { + let mut disks = source_set.disks.write().await; + let original = disks.clone(); + for disk in disks.iter_mut().take(3) { + *disk = None; + } + original + }; + barrier.release(); + let cleanup_error = cleanup + .await + .expect("source cleanup task should not panic") + .expect_err("injected source delete quorum failure must fail cleanup") + .to_string(); + *source_set.disks.write().await = original_source_disks; + drop(barrier); + + assert!(cleanup_error.contains("source durable ILM cleanup failed")); + assert_eq!( + store + .decommission_durable_ilm_receipt_count_for_test(0) + .await + .expect("receipt should persist before source cleanup"), + 1 + ); + assert_eq!( + com::read_config(store.pools[0].clone(), &path) + .await + .expect("failed cleanup must retain the source"), + data + ); + + let mut restarted_pool_meta = PoolMeta::default(); + restarted_pool_meta + .load(store.pools[0].clone(), store.pools.clone()) + .await + .expect("decommission run identity should reload after restart"); + *store.pool_meta.write().await = restarted_pool_meta; + let stats = recover_tier_delete_journal_entries(store.clone(), 100, None) + .await + .expect("target recovery should commit terminal proof and delete the target"); + assert_eq!((stats.scanned, stats.deleted, stats.failed), (1, 1, 0)); + assert!(matches!( + com::read_config(store.pools[1].clone(), &path).await, + Err(Error::ConfigNotFound) + )); + assert_eq!( + com::read_config(store.pools[0].clone(), &path) + .await + .expect("target recovery must not delete the decommission source"), + data + ); + + store + .verify_and_cleanup_decommissioned_durable_ilm_record_for_test(0, source_set, &path) + .await + .expect("terminal receipt should authorize cleanup after target deletion"); + assert!(matches!( + com::read_config(store.pools[0].clone(), &path).await, + Err(Error::ConfigNotFound) + )); + assert!(backend.remove_versions().await.contains(&(entry.obj_name, entry.version_id))); + } + #[cfg(feature = "test-util")] #[tokio::test] #[serial_test::serial(storage_class_env)]