From 619fae2512c4944e068a6b5b68ce5ddea6a5fd8b Mon Sep 17 00:00:00 2001 From: overtrue Date: Sat, 22 Aug 2026 01:22:57 +0800 Subject: [PATCH] fix(ecstore): verify ILM metadata before decommission --- .config/nextest.toml | 10 + .../src/bucket/lifecycle/durable_namespace.rs | 261 +++++++++ .../bucket/lifecycle/manual_transition_job.rs | 12 +- crates/ecstore/src/bucket/lifecycle/mod.rs | 5 +- .../bucket/lifecycle/tier_delete_journal.rs | 3 +- .../lifecycle/transition_transaction.rs | 3 +- crates/ecstore/src/config/com.rs | 8 + crates/ecstore/src/core/pools.rs | 553 +++++++++++++++++- crates/ecstore/src/store/init.rs | 430 +++++++++++++- 9 files changed, 1246 insertions(+), 39 deletions(-) create mode 100644 crates/ecstore/src/bucket/lifecycle/durable_namespace.rs diff --git a/.config/nextest.toml b/.config/nextest.toml index 097c79e3e..4e1637705 100644 --- a/.config/nextest.toml +++ b/.config/nextest.toml @@ -78,6 +78,12 @@ 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. +[[profile.default.overrides]] +filter = 'package(rustfs-ecstore) & test(decommission_migrates_and_verifies_registered_durable_ilm_records)' +test-group = 'ecstore-serial-flaky' + # Serialize the bucket-incarnation / lifecycle-fence tests. They drive # init_bucket_metadata_sys and bucket_metadata_sys_of, i.e. process-global # OnceLock state that serial_test's #[serial] cannot protect across nextest's @@ -190,6 +196,10 @@ test-group = 'embedded-test-ports' filter = 'package(rustfs-ecstore) & test(manual_transition_page_checkpoint_persists_durable_job_progress)' test-group = 'ecstore-serial-flaky' +[[profile.ci.overrides]] +filter = 'package(rustfs-ecstore) & test(decommission_migrates_and_verifies_registered_durable_ilm_records)' +test-group = 'ecstore-serial-flaky' + # Serialize the bucket-incarnation / lifecycle-fence tests under the ci profile # too (see the matching default-profile override near the top). No retries. [[profile.ci.overrides]] diff --git a/crates/ecstore/src/bucket/lifecycle/durable_namespace.rs b/crates/ecstore/src/bucket/lifecycle/durable_namespace.rs new file mode 100644 index 000000000..786c39242 --- /dev/null +++ b/crates/ecstore/src/bucket/lifecycle/durable_namespace.rs @@ -0,0 +1,261 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use uuid::Uuid; + +use super::{manual_transition_job, tier_delete_journal, transition_transaction}; +use crate::error::{Error, Result}; + +pub(crate) const ILM_META_PREFIX: &str = "ilm"; +const ILM_META_OBJECT_PREFIX: &str = "ilm/"; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum DurableIlmRecordKind { + TierDeleteJournal, + TransitionTransaction, + ManualTransitionJob, + ManualTransitionScope, + ManualTransitionTask, + ManualTransitionWorkerResult, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) struct DurableIlmNamespace { + pub(crate) name: &'static str, + pub(crate) prefix: &'static str, + pub(crate) max_record_size: usize, + kind: DurableIlmRecordKind, +} + +pub(crate) const TIER_DELETE_JOURNAL_NAMESPACE: DurableIlmNamespace = DurableIlmNamespace { + name: "tier-delete-journal", + prefix: "ilm/tier-delete-journal/", + max_record_size: 64 * 1024, + kind: DurableIlmRecordKind::TierDeleteJournal, +}; +pub(crate) const TRANSITION_TRANSACTION_NAMESPACE: DurableIlmNamespace = DurableIlmNamespace { + name: "transition-transaction", + prefix: "ilm/transition-transactions/records", + max_record_size: transition_transaction::MAX_TRANSITION_TRANSACTION_SIZE, + kind: DurableIlmRecordKind::TransitionTransaction, +}; +pub(crate) const MANUAL_TRANSITION_JOB_NAMESPACE: DurableIlmNamespace = DurableIlmNamespace { + name: "manual-transition-job", + prefix: "ilm/manual-transition/jobs", + max_record_size: manual_transition_job::MAX_MANUAL_TRANSITION_JOB_RECORD_SIZE, + kind: DurableIlmRecordKind::ManualTransitionJob, +}; +pub(crate) const MANUAL_TRANSITION_SCOPE_NAMESPACE: DurableIlmNamespace = DurableIlmNamespace { + name: "manual-transition-scope", + prefix: "ilm/manual-transition/scopes", + max_record_size: manual_transition_job::MAX_MANUAL_TRANSITION_JOB_RECORD_SIZE, + kind: DurableIlmRecordKind::ManualTransitionScope, +}; +pub(crate) const MANUAL_TRANSITION_TASK_NAMESPACE: DurableIlmNamespace = DurableIlmNamespace { + name: "manual-transition-task", + prefix: "ilm/manual-transition/tasks", + max_record_size: manual_transition_job::MAX_MANUAL_TRANSITION_TASK_RECORD_SIZE, + kind: DurableIlmRecordKind::ManualTransitionTask, +}; +pub(crate) const MANUAL_TRANSITION_WORKER_RESULT_NAMESPACE: DurableIlmNamespace = DurableIlmNamespace { + name: "manual-transition-worker-result", + prefix: "ilm/manual-transition/results", + max_record_size: manual_transition_job::MAX_MANUAL_TRANSITION_WORKER_RESULT_RECORD_SIZE, + kind: DurableIlmRecordKind::ManualTransitionWorkerResult, +}; + +pub(crate) const DURABLE_ILM_NAMESPACES: [DurableIlmNamespace; 6] = [ + TIER_DELETE_JOURNAL_NAMESPACE, + TRANSITION_TRANSACTION_NAMESPACE, + MANUAL_TRANSITION_JOB_NAMESPACE, + MANUAL_TRANSITION_SCOPE_NAMESPACE, + MANUAL_TRANSITION_TASK_NAMESPACE, + MANUAL_TRANSITION_WORKER_RESULT_NAMESPACE, +]; + +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) struct ValidatedDurableIlmRecord { + pub(crate) namespace: &'static str, + pub(crate) id_kind: &'static str, + pub(crate) id: String, +} + +impl ValidatedDurableIlmRecord { + pub(crate) fn context(&self) -> String { + format!("namespace `{}` {} `{}`", self.namespace, self.id_kind, self.id) + } +} + +fn path_is_in_namespace(path: &str, namespace: &DurableIlmNamespace) -> bool { + let Some(suffix) = path.strip_prefix(namespace.prefix) else { + return false; + }; + if namespace.prefix.ends_with('/') { + !suffix.is_empty() + } else { + suffix.starts_with('/') && suffix.len() > 1 + } +} + +pub(crate) fn classify_durable_ilm_record(path: &str) -> Result> { + if path != ILM_META_PREFIX && !path.starts_with(ILM_META_OBJECT_PREFIX) { + return Ok(None); + } + + DURABLE_ILM_NAMESPACES + .iter() + .find(|namespace| path_is_in_namespace(path, namespace)) + .map(Some) + .ok_or_else(|| Error::other(format!("unregistered durable ILM namespace for path `{path}`"))) +} + +fn parse_manual_sharded_record(path: &str, prefix: &str) -> Result<(Uuid, String)> { + let suffix = path + .strip_prefix(prefix) + .and_then(|suffix| suffix.strip_prefix('/')) + .ok_or_else(|| Error::other("manual transition record path has wrong prefix"))?; + let mut parts = suffix.split('/'); + let first = parts + .next() + .ok_or_else(|| Error::other("manual transition record first shard is missing"))?; + let second = parts + .next() + .ok_or_else(|| Error::other("manual transition record second shard is missing"))?; + let job_key = parts + .next() + .ok_or_else(|| Error::other("manual transition record job id is missing"))?; + let task_key = parts + .next() + .and_then(|file| file.strip_suffix(".json")) + .ok_or_else(|| Error::other("manual transition record task key is missing"))?; + if parts.next().is_some() + || job_key.len() != 32 + || first != &job_key[..2] + || second != &job_key[2..4] + || !job_key + .bytes() + .all(|byte| byte.is_ascii_hexdigit() && !byte.is_ascii_uppercase()) + { + return Err(Error::other("manual transition record job id or shards are invalid")); + } + let job_id = Uuid::parse_str(job_key).map_err(|_| Error::other("manual transition record job id is invalid"))?; + Ok((job_id, task_key.to_string())) +} + +pub(crate) fn validate_durable_ilm_record(path: &str, data: &[u8]) -> Result { + let namespace = + classify_durable_ilm_record(path)?.ok_or_else(|| Error::other(format!("path `{path}` is not a durable ILM record")))?; + if data.len() > namespace.max_record_size { + return Err(Error::other(format!( + "durable ILM record exceeds {} byte limit", + namespace.max_record_size + ))); + } + + let (id_kind, id) = match namespace.kind { + DurableIlmRecordKind::TierDeleteJournal => { + let entry = tier_delete_journal::decode_tier_delete_journal_entry(data)?; + if tier_delete_journal::tier_delete_journal_object_name(&entry) != path { + return Err(Error::other("tier delete journal content does not match its path")); + } + let operation_id = path + .strip_prefix(namespace.prefix) + .and_then(|suffix| suffix.strip_suffix(".json")) + .ok_or_else(|| Error::other("tier delete journal path is invalid"))?; + ("operation_id", operation_id.to_string()) + } + DurableIlmRecordKind::TransitionTransaction => { + 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()) + } + DurableIlmRecordKind::ManualTransitionJob => { + let job_id = manual_transition_job::manual_transition_job_id_from_record_object_name(path) + .map_err(|err| Error::other(err.to_string()))?; + let canonical = manual_transition_job::manual_transition_job_record_object_name(job_id) + .map_err(|err| Error::other(err.to_string()))?; + if canonical != path { + return Err(Error::other("manual transition job path is not canonical")); + } + manual_transition_job::ManualTransitionJobRecord::decode(job_id, data) + .map_err(|err| Error::other(err.to_string()))?; + ("job_id", job_id.to_string()) + } + DurableIlmRecordKind::ManualTransitionScope => { + let admission: manual_transition_job::ManualTransitionScopeAdmission = + serde_json::from_slice(data).map_err(Error::other)?; + admission.validate().map_err(|err| Error::other(err.to_string()))?; + let canonical = manual_transition_job::manual_transition_scope_record_object_name(&admission.scope_key) + .map_err(|err| Error::other(err.to_string()))?; + if canonical != path { + return Err(Error::other("manual transition scope content does not match its path")); + } + ("job_id", admission.job_id.to_string()) + } + DurableIlmRecordKind::ManualTransitionTask => { + let (job_id, task_key) = parse_manual_sharded_record(path, namespace.prefix)?; + let canonical = manual_transition_job::manual_transition_task_object_name(job_id, &task_key) + .map_err(|err| Error::other(err.to_string()))?; + if canonical != path { + return Err(Error::other("manual transition task path is not canonical")); + } + manual_transition_job::ManualTransitionTaskRecord::decode(job_id, &task_key, data) + .map_err(|err| Error::other(err.to_string()))?; + ("job_id", job_id.to_string()) + } + DurableIlmRecordKind::ManualTransitionWorkerResult => { + let (job_id, task_key) = parse_manual_sharded_record(path, namespace.prefix)?; + let canonical = manual_transition_job::manual_transition_worker_result_object_name(job_id, &task_key) + .map_err(|err| Error::other(err.to_string()))?; + if canonical != path { + return Err(Error::other("manual transition worker result path is not canonical")); + } + manual_transition_job::ManualTransitionWorkerResultRecord::decode(job_id, &task_key, data) + .map_err(|err| Error::other(err.to_string()))?; + ("job_id", job_id.to_string()) + } + }; + + Ok(ValidatedDurableIlmRecord { + namespace: namespace.name, + id_kind, + id, + }) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn unknown_ilm_record_requires_namespace_registration() { + let err = classify_durable_ilm_record("ilm/future-durable/jobs/one.json") + .expect_err("unknown durable ILM path must fail closed"); + + assert!(err.to_string().contains("ilm/future-durable/jobs/one.json")); + } + + #[test] + fn durable_ilm_registry_has_unique_non_overlapping_prefixes() { + for (index, namespace) in DURABLE_ILM_NAMESPACES.iter().enumerate() { + assert!(namespace.prefix.starts_with(ILM_META_OBJECT_PREFIX)); + assert!(namespace.max_record_size > 0); + for other in DURABLE_ILM_NAMESPACES.iter().skip(index + 1) { + assert_ne!(namespace.prefix, other.prefix); + assert!(!path_is_in_namespace(namespace.prefix, other)); + assert!(!path_is_in_namespace(other.prefix, namespace)); + } + } + } +} diff --git a/crates/ecstore/src/bucket/lifecycle/manual_transition_job.rs b/crates/ecstore/src/bucket/lifecycle/manual_transition_job.rs index b1fe45cad..c8d4d2e76 100644 --- a/crates/ecstore/src/bucket/lifecycle/manual_transition_job.rs +++ b/crates/ecstore/src/bucket/lifecycle/manual_transition_job.rs @@ -24,6 +24,10 @@ use crate::bucket::lifecycle::bucket_lifecycle_ops::{ ManualTransitionQueueSnapshot, ManualTransitionRunOptions, ManualTransitionRunReport, }; use crate::bucket::lifecycle::config_boundary; +use crate::bucket::lifecycle::durable_namespace::{ + MANUAL_TRANSITION_JOB_NAMESPACE, MANUAL_TRANSITION_SCOPE_NAMESPACE, MANUAL_TRANSITION_TASK_NAMESPACE, + MANUAL_TRANSITION_WORKER_RESULT_NAMESPACE, +}; use crate::disk::RUSTFS_META_BUCKET; use crate::error::{Error, Result as EcstoreResult}; use crate::object_api::ObjectOptions; @@ -34,10 +38,10 @@ use crate::store::ECStore; pub const MANUAL_TRANSITION_JOB_SCHEMA: &str = "rustfs-manual-transition-job-v1"; pub const MANUAL_TRANSITION_TASK_SCHEMA: &str = "rustfs-manual-transition-task-v1"; pub const MANUAL_TRANSITION_WORKER_RESULT_SCHEMA: &str = "rustfs-manual-transition-worker-result-v1"; -pub const MANUAL_TRANSITION_JOB_RECORD_PREFIX: &str = "ilm/manual-transition/jobs"; -pub const MANUAL_TRANSITION_SCOPE_RECORD_PREFIX: &str = "ilm/manual-transition/scopes"; -pub const MANUAL_TRANSITION_TASK_PREFIX: &str = "ilm/manual-transition/tasks"; -pub const MANUAL_TRANSITION_WORKER_RESULT_PREFIX: &str = "ilm/manual-transition/results"; +pub const MANUAL_TRANSITION_JOB_RECORD_PREFIX: &str = MANUAL_TRANSITION_JOB_NAMESPACE.prefix; +pub const MANUAL_TRANSITION_SCOPE_RECORD_PREFIX: &str = MANUAL_TRANSITION_SCOPE_NAMESPACE.prefix; +pub const MANUAL_TRANSITION_TASK_PREFIX: &str = MANUAL_TRANSITION_TASK_NAMESPACE.prefix; +pub const MANUAL_TRANSITION_WORKER_RESULT_PREFIX: &str = MANUAL_TRANSITION_WORKER_RESULT_NAMESPACE.prefix; pub const MAX_MANUAL_TRANSITION_JOB_RECORD_SIZE: usize = 64 * 1024; pub const MAX_MANUAL_TRANSITION_TASK_RECORD_SIZE: usize = 16 * 1024; pub const MAX_MANUAL_TRANSITION_WORKER_RESULT_RECORD_SIZE: usize = 8 * 1024; diff --git a/crates/ecstore/src/bucket/lifecycle/mod.rs b/crates/ecstore/src/bucket/lifecycle/mod.rs index b0cac4d0b..43b7fac04 100644 --- a/crates/ecstore/src/bucket/lifecycle/mod.rs +++ b/crates/ecstore/src/bucket/lifecycle/mod.rs @@ -16,6 +16,7 @@ pub mod bucket_lifecycle_audit; pub mod bucket_lifecycle_ops; mod config_boundary; pub mod core; +mod durable_namespace; pub mod evaluator; pub mod manual_transition_job; mod metadata_boundary; @@ -32,4 +33,6 @@ pub mod tier_last_day_stats; pub mod tier_sweeper; pub mod transition_transaction; -pub(crate) const ILM_META_PREFIX: &str = "ilm"; +pub(crate) use durable_namespace::{ + ILM_META_PREFIX, ValidatedDurableIlmRecord, classify_durable_ilm_record, validate_durable_ilm_record, +}; diff --git a/crates/ecstore/src/bucket/lifecycle/tier_delete_journal.rs b/crates/ecstore/src/bucket/lifecycle/tier_delete_journal.rs index 6308b3767..48f79d5f6 100644 --- a/crates/ecstore/src/bucket/lifecycle/tier_delete_journal.rs +++ b/crates/ecstore/src/bucket/lifecycle/tier_delete_journal.rs @@ -20,6 +20,7 @@ use tokio_util::sync::CancellationToken; use tracing::{debug, warn}; use crate::bucket::lifecycle::config_boundary; +use crate::bucket::lifecycle::durable_namespace::TIER_DELETE_JOURNAL_NAMESPACE; use crate::bucket::lifecycle::runtime_boundary; use crate::bucket::lifecycle::tier_sweeper::{ Jentry, TierDeleteJournalState, TierDeleteSourceIdentity, @@ -49,7 +50,7 @@ const TIER_DELETE_JOURNAL_VERSION: u8 = 2; const TIER_DELETE_JOURNAL_EXACT_VERSION: u8 = 3; const TIER_DELETE_JOURNAL_STATE_VERSION: u8 = 4; const TIER_DELETE_JOURNAL_TRANSACTION_VERSION: u8 = 5; -pub(crate) const TIER_DELETE_JOURNAL_PREFIX: &str = "ilm/tier-delete-journal/"; +pub(crate) const TIER_DELETE_JOURNAL_PREFIX: &str = TIER_DELETE_JOURNAL_NAMESPACE.prefix; #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] #[serde(deny_unknown_fields)] diff --git a/crates/ecstore/src/bucket/lifecycle/transition_transaction.rs b/crates/ecstore/src/bucket/lifecycle/transition_transaction.rs index f9831d394..ae216bc29 100644 --- a/crates/ecstore/src/bucket/lifecycle/transition_transaction.rs +++ b/crates/ecstore/src/bucket/lifecycle/transition_transaction.rs @@ -21,6 +21,7 @@ use tracing::{debug, warn}; use uuid::Uuid; use crate::bucket::lifecycle::config_boundary; +use crate::bucket::lifecycle::durable_namespace::TRANSITION_TRANSACTION_NAMESPACE; use crate::bucket::lifecycle::lifecycle::TRANSITION_COMPLETE; use crate::bucket::lifecycle::tier_sweeper::{ delete_confirmed_transition_candidate_exact_with_lease_idempotent, @@ -42,7 +43,7 @@ const TRANSITION_TRANSACTION_RECOVERY_INTERVAL: Duration = Duration::from_secs(6 const TRANSITION_TRANSACTION_RECOVERY_TIMEOUT: Duration = Duration::from_secs(300); pub const TRANSITION_TRANSACTION_SCHEMA: &str = "rustfs-transition-transaction-v1"; pub const TRANSITION_TRANSACTION_PREFIX: &str = "ilm/transition-transactions"; -pub const TRANSITION_TRANSACTION_RECORD_PREFIX: &str = "ilm/transition-transactions/records"; +pub const TRANSITION_TRANSACTION_RECORD_PREFIX: &str = TRANSITION_TRANSACTION_NAMESPACE.prefix; pub const MAX_TRANSITION_TRANSACTION_SIZE: usize = 64 * 1024; pub type Result = std::result::Result; diff --git a/crates/ecstore/src/config/com.rs b/crates/ecstore/src/config/com.rs index 3b2729b8b..e03eace21 100644 --- a/crates/ecstore/src/config/com.rs +++ b/crates/ecstore/src/config/com.rs @@ -406,6 +406,14 @@ where Ok(data) } +pub(crate) async fn read_config_limited_preserve_empty(api: Arc, file: &str, max_bytes: usize) -> Result> +where + S: EcstoreObjectIO, +{ + let (data, _obj) = read_config_with_metadata_inner(api, file, &ObjectOptions::default(), true, Some(max_bytes)).await?; + Ok(data) +} + /// 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 51bf7b658..2b59acc66 100644 --- a/crates/ecstore/src/core/pools.rs +++ b/crates/ecstore/src/core/pools.rs @@ -16,19 +16,23 @@ use crate::bucket::replication::replication_state_from_filemeta; use crate::bucket::versioning_sys::BucketVersioningSys; use crate::bucket::{ lifecycle::{ - ILM_META_PREFIX, LifecycleExpiryConfigs, + 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, lifecycle_delete_all_versions_blocked_by_replication, }, - get_expiry_configs, + classify_durable_ilm_record, get_expiry_configs, lifecycle::IlmAction, + validate_durable_ilm_record, }, metadata_sys, }; use crate::cache_value::metacache_set::{ListPathRawOptions, list_path_raw}; -use crate::config::com::{CONFIG_PREFIX, read_config, read_config_no_lock, save_config, save_config_with_opts}; +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, +}; use crate::data_movement; use crate::data_movement::backpressure::{self, DataMovementOperation}; use crate::data_usage::DATA_USAGE_CACHE_NAME; @@ -48,6 +52,7 @@ use crate::storage_api_contracts::{ admin::StorageAdminApi, bucket::{BucketOperations, BucketOptions, MakeBucketOptions}, heal::HealOperations as _, + list::ListOperations as _, namespace::NamespaceLocking as _, object::{EcstoreObjectIO, ObjectIO as _, ObjectOperations as _}, }; @@ -61,6 +66,7 @@ use rmp_serde::Serializer; use rustfs_common::defer; use rustfs_common::heal_channel::HealOpts; use rustfs_filemeta::{FileInfoVersions, MetaCacheEntries, MetaCacheEntry, MetadataResolutionParams}; +use rustfs_utils::crypto::{hex_sha256, is_sha256_checksum}; use rustfs_utils::path::{encode_dir_object, path_join, path_to_bucket_object, path_to_bucket_object_with_base_path}; use s3s::dto::{BucketLifecycleConfiguration, ObjectLockConfiguration, ReplicationConfiguration}; use serde::{Deserialize, Serialize}; @@ -97,6 +103,9 @@ const DECOMMISSION_META_PREFIXES: [&str; 3] = [CONFIG_PREFIX, BUCKET_META_PREFIX 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_MAX_SIZE: usize = 16 * 1024; /// 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); @@ -845,6 +854,135 @@ 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| { + 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(()) +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +struct DecommissionDurableIlmReceipt { + source_path: String, + namespace: String, + id_kind: String, + id: String, + target_content_sha256: String, +} + +impl DecommissionDurableIlmReceipt { + fn new(path: &str, record: &ValidatedDurableIlmRecord, target: &[u8]) -> 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), + } + } + + fn context(&self) -> String { + format!("namespace `{}` {} `{}`", self.namespace, self.id_kind, self.id) + } + + fn validate(&self) -> Result<()> { + let namespace = classify_durable_ilm_record(&self.source_path)? + .ok_or_else(|| Error::other(format!("receipt source path `{}` is not a durable ILM record", self.source_path)))?; + if namespace.name != self.namespace { + return Err(Error::other(format!( + "receipt namespace `{}` does not match source path `{}`", + self.namespace, self.source_path + ))); + } + if self.id_kind.is_empty() || self.id.is_empty() { + return Err(Error::other(format!( + "receipt identity is missing for source path `{}`", + self.source_path + ))); + } + if !is_sha256_checksum(&self.target_content_sha256) { + return Err(Error::other(format!( + "receipt target checksum is invalid for source path `{}` {}", + self.source_path, + self.context() + ))); + } + Ok(()) + } + + fn encode(&self) -> Result> { + self.validate()?; + let receipt_bytes = serde_json::to_vec(self)?; + let persisted = PersistedDecommissionDurableIlmReceipt { + schema: DECOMMISSION_DURABLE_ILM_RECEIPT_SCHEMA.to_string(), + content_sha256: hex_sha256(&receipt_bytes, ToOwned::to_owned), + receipt: self.clone(), + }; + let encoded = serde_json::to_vec(&persisted)?; + if encoded.len() > DECOMMISSION_DURABLE_ILM_RECEIPT_MAX_SIZE { + return Err(Error::other(format!( + "durable ILM receipt exceeds maximum size for source path `{}` {}", + self.source_path, + self.context() + ))); + } + Ok(encoded) + } + + fn decode(data: &[u8]) -> Result { + if data.len() > DECOMMISSION_DURABLE_ILM_RECEIPT_MAX_SIZE { + return Err(Error::other("durable ILM receipt exceeds maximum size")); + } + let persisted: PersistedDecommissionDurableIlmReceipt = serde_json::from_slice(data)?; + if persisted.schema != DECOMMISSION_DURABLE_ILM_RECEIPT_SCHEMA { + return Err(Error::other(format!("unsupported durable ILM receipt schema `{}`", persisted.schema))); + } + if !is_sha256_checksum(&persisted.content_sha256) { + return Err(Error::other("durable ILM receipt checksum is invalid")); + } + let receipt_bytes = serde_json::to_vec(&persisted.receipt)?; + let actual_checksum = hex_sha256(&receipt_bytes, ToOwned::to_owned); + if persisted.content_sha256 != actual_checksum { + return Err(Error::other("durable ILM receipt checksum mismatch")); + } + persisted.receipt.validate()?; + Ok(persisted.receipt) + } +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +struct PersistedDecommissionDurableIlmReceipt { + schema: String, + content_sha256: String, + 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}/") +} + +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") +} + fn resolve_decommission_pool_meta_reload_result(result: Result<()>, stage: &str) -> Result<()> { result.map_err(|err| Error::other(format!("decommission pool meta reload failed during {stage}: {err}"))) } @@ -2916,6 +3054,11 @@ impl ECStore { Ok(()) } + #[cfg(test)] + pub(crate) async fn promote_queued_decommission_for_test(&self, idx: usize) -> Result<()> { + self.promote_queued_decommission(idx).await + } + async fn record_decommission_terminal_reload_failure(&self, idx: usize, stage: &str, err: Error) -> Result<()> { let changed = { let mut pool_meta = self.pool_meta.write().await; @@ -3103,6 +3246,12 @@ impl ECStore { ); return Ok(()); } + let durable_ilm_record = if bucket == RUSTFS_META_BUCKET { + classify_durable_ilm_record(&entry.name) + .map_err(|err| with_decommission_entry_context("durable_ilm_namespace", &bucket, &entry.name, err))? + } else { + None + }; if self.decommission_cancel_requested(idx, &rx).await { rx.cancel(); } @@ -3415,7 +3564,8 @@ impl ECStore { } } - if should_cleanup_decommission_source_entry(decommissioned, fivs.versions.len(), expired) { + if should_cleanup_decommission_source_entry(decommissioned, fivs.versions.len(), expired) && durable_ilm_record.is_none() + { if bucket_incarnation_fence.as_ref().is_some_and(|guard| guard.is_lock_lost()) { return Err(Error::other("decommission bucket incarnation fence was lost before source cleanup")); } @@ -3460,6 +3610,17 @@ impl ECStore { data_movement::SourceCleanupError::Storage(err) => err, }); resolve_decommission_entry_cleanup_delete_result(cleanup_result, bucket.as_str(), entry.name.as_str())? + } else if durable_ilm_record.is_some() { + debug!( + event = EVENT_DECOMMISSION_ENTRY, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_POOLS, + pool_index = idx, + bucket = %bucket, + object = %entry.name, + state = "retained_for_final_verification", + "Decommission durable ILM source retained for final verification" + ); } else if decommissioned != fivs.versions.len() || expired > 0 { warn!( event = EVENT_DECOMMISSION_ENTRY, @@ -3787,6 +3948,17 @@ impl ECStore { Ok(()) } + #[cfg(test)] + pub(crate) async fn decommission_pool_for_test( + self: &Arc, + rx: CancellationToken, + idx: usize, + pool: Arc, + bucket: DecomBucketInfo, + ) -> Result<()> { + self.decommission_pool(rx, idx, pool, bucket).await + } + #[tracing::instrument(skip(self, rx))] pub async fn do_decommission_in_routine(self: &Arc, rx: CancellationToken, idx: usize) -> Result<()> { defer!(|| async { @@ -3935,7 +4107,6 @@ impl ECStore { "failed to finalize decommission for pool {cmd_line}: post-check failed: {err}" ))); } - info!( event = EVENT_DECOMMISSION_STATE, component = LOG_COMPONENT_ECSTORE, @@ -3945,7 +4116,10 @@ impl ECStore { state = "marking_completed", "Decommission marking completed state" ); - resolve_decommission_terminal_mark_result(self.complete_decommission(idx).await, "completed", &cmd_line)?; + if let Err(err) = self.complete_decommission(idx).await { + resolve_decommission_terminal_mark_result(self.decommission_failed(idx).await, "failed", &cmd_line)?; + return Err(Error::other(format!("failed to finalize decommission for pool {cmd_line}: {err}"))); + } } DecommissionFinalState::Failed => { warn!( @@ -4041,12 +4215,19 @@ impl ECStore { #[tracing::instrument(skip(self))] pub async fn complete_decommission(&self, idx: usize) -> Result<()> { ensure_decommission_terminal_operation_supported(self.single_pool(), "complete decommission")?; + ensure_valid_decommission_pool_index(self.pools.len(), idx)?; + self.verify_decommission_durable_ilm_receipts(idx).await?; - let (should_reload_pool_meta, previous_pool_meta) = { + let (should_reload_pool_meta, completed, previous_pool_meta) = { let mut pool_meta = self.pool_meta.write().await; let previous_pool_meta = pool_meta.clone(); let changed = pool_meta.decommission_complete(idx); - (changed, changed.then_some(previous_pool_meta)) + let completed = pool_meta + .pools + .get(idx) + .and_then(|pool| pool.decommission.as_ref()) + .is_some_and(|decommission| decommission.complete); + (changed, completed, changed.then_some(previous_pool_meta)) }; { @@ -4100,6 +4281,18 @@ impl ECStore { } } + if completed && let Err(err) = self.cleanup_decommission_durable_ilm_receipts(idx).await { + warn!( + event = EVENT_DECOMMISSION_STATE, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_POOLS, + pool_index = idx, + state = "receipt_cleanup_failed", + error = %err, + "Decommission durable ILM receipt cleanup failed" + ); + } + Ok(()) } @@ -4350,6 +4543,268 @@ impl ECStore { Ok(ret) } + async fn durable_ilm_receipt_prefix(&self, source_pool_idx: usize) -> Result { + let pool_meta = self.pool_meta.read().await; + let cmd_line = 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_prefix(&cmd_line)) + } + + async fn load_decommissioned_durable_ilm_target( + &self, + source_pool_idx: usize, + path: &str, + max_record_size: usize, + record_context: &str, + ) -> Result<(usize, Vec)> { + let mut target = None::<(usize, Vec)>; + let mut first_read_error = None; + for (target_pool_idx, pool) in self.pools.iter().enumerate() { + if target_pool_idx == source_pool_idx { + continue; + } + match read_config_limited_preserve_empty(pool.clone(), path, max_record_size).await { + Ok(data) => { + if let Some((existing_pool_idx, existing)) = target.as_ref() + && existing != &data + { + return Err(Error::other(format!( + "divergent target durable ILM records at path `{path}` {record_context} in pools {existing_pool_idx} and {target_pool_idx}" + ))); + } + target = Some((target_pool_idx, data)); + } + Err(err) + if matches!(&err, Error::ConfigNotFound | Error::FileNotFound | Error::FileVersionNotFound) + || is_err_object_not_found(&err) + || is_err_version_not_found(&err) => {} + Err(err) => { + first_read_error.get_or_insert_with(|| { + Error::other(format!( + "failed to read target durable ILM record at path `{path}` {record_context} from pool {target_pool_idx}: {err}" + )) + }); + } + } + } + + target.ok_or_else(|| { + first_read_error.unwrap_or_else(|| { + Error::other(format!("target durable ILM record is missing at path `{path}` {record_context}")) + }) + }) + } + + 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 mut receipts = Vec::new(); + for (pool_idx, pool) in self.pools.iter().enumerate() { + 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; + } + 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 persist_decommission_durable_ilm_receipt( + &self, + source_pool_idx: usize, + 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 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) + .await + .map_err(|err| { + Error::other(format!( + "failed to persist durable ILM decommission receipt `{receipt_path}` for source path `{}` {}: {err}", + receipt.source_path, + receipt.context() + )) + }) + } + + 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 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 + .load_decommissioned_durable_ilm_target( + source_pool_idx, + &receipt.source_path, + namespace.max_record_size, + &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 + { + 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 `{}` {}", + receipt.source_path, + receipt.context() + ))); + } + } + Ok(()) + } + + 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 { + Ok(()) | Err(Error::ConfigNotFound | Error::FileNotFound | Error::FileVersionNotFound) => {} + Err(err) if is_err_object_not_found(&err) || is_err_version_not_found(&err) => {} + Err(err) => { + return Err(Error::other(format!( + "failed to clean durable ILM decommission receipt `{receipt_path}` from pool {pool_idx}: {err}" + ))); + } + } + } + Ok(()) + } + + async fn verify_and_cleanup_decommissioned_durable_ilm_record( + &self, + source_pool_idx: usize, + source_set: Arc, + path: &str, + ) -> Result<()> { + let namespace = classify_durable_ilm_record(path)? + .ok_or_else(|| Error::other(format!("path `{path}` is not a durable ILM record")))?; + let source_versions = source_set + .load_file_info_versions_exact(RUSTFS_META_BUCKET, path) + .await + .map_err(|err| Error::other(format!("failed to load source durable ILM versions at path `{path}`: {err}")))? + .ok_or_else(|| Error::other(format!("source durable ILM record is missing at path `{path}`")))?; + let source = read_config_limited_preserve_empty(source_set.clone(), path, namespace.max_record_size) + .await + .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 + .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); + self.persist_decommission_durable_ilm_receipt(source_pool_idx, target_pool_idx, &receipt) + .await?; + + let cleanup_result = data_movement::cleanup_source_entry_if_unchanged( + source_set, + RUSTFS_META_BUCKET, + path, + &source_versions, + &[], + data_movement::SourceCleanupBucketFence::default(), + "decommission durable ILM final sweep", + ) + .await + .map_err(|err| { + Error::other(format!( + "source durable ILM cleanup failed at path `{path}` {}: {err}", + source_record.context() + )) + }); + resolve_decommission_entry_cleanup_delete_result(cleanup_result, RUSTFS_META_BUCKET, path) + } + + #[cfg(test)] + pub(crate) async fn verify_and_cleanup_decommissioned_durable_ilm_record_for_test( + &self, + source_pool_idx: usize, + source_set: Arc, + path: &str, + ) -> Result<()> { + self.verify_and_cleanup_decommissioned_durable_ilm_record(source_pool_idx, source_set, path) + .await + } + + #[cfg(test)] + pub(crate) async fn decommission_durable_ilm_receipt_count_for_test(&self, source_pool_idx: usize) -> Result { + Ok(self.list_decommission_durable_ilm_receipts(source_pool_idx).await?.len()) + } + + #[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 + } + async fn check_after_decommission(self: &Arc, idx: usize) -> Result<()> { let buckets = self.get_buckets_to_decommission().await?; let pool = self.pools[idx].clone(); @@ -4366,22 +4821,27 @@ impl ECStore { let versions_found = Arc::new(AtomicUsize::new(0)); let entry_error = Arc::new(tokio::sync::Mutex::new(None::)); + let first_remaining_path = Arc::new(tokio::sync::Mutex::new(None::)); let callback_rx = CancellationToken::new(); let versions_found_cb = versions_found.clone(); let entry_error_cb = entry_error.clone(); + let first_remaining_path_cb = first_remaining_path.clone(); let bucket_name = bucket_info.name.clone(); let lifecycle_config_cb = lifecycle_config.clone(); let object_lock_config_cb = object_lock_config.clone(); let store = Arc::clone(self); + let source_set = set.clone(); let callback_rx_cb = callback_rx.clone(); let callback: ListCallback = Arc::new(move |entry: MetaCacheEntry| { let versions_found = versions_found_cb.clone(); let entry_error = entry_error_cb.clone(); + let first_remaining_path = first_remaining_path_cb.clone(); let bucket_name = bucket_name.clone(); let lifecycle_config = lifecycle_config_cb.clone(); let object_lock_config = object_lock_config_cb.clone(); let store = Arc::clone(&store); + let source_set = source_set.clone(); let callback_rx = callback_rx_cb.clone(); Box::pin(async move { if callback_rx.is_cancelled() { @@ -4396,6 +4856,41 @@ impl ECStore { return; } + let durable_ilm_record = if bucket_name == RUSTFS_META_BUCKET { + match classify_durable_ilm_record(&entry.name) { + Ok(record) => record, + Err(err) => { + let mut first_err = entry_error.lock().await; + if first_err.is_none() { + *first_err = Some(with_decommission_entry_context( + "check_after_decommission.durable_ilm_namespace", + &bucket_name, + &entry.name, + err, + )); + callback_rx.cancel(); + } + return; + } + } + } else { + None + }; + + if durable_ilm_record.is_some() { + if let Err(err) = store + .verify_and_cleanup_decommissioned_durable_ilm_record(idx, source_set, &entry.name) + .await + { + let mut first_err = entry_error.lock().await; + if first_err.is_none() { + *first_err = Some(err); + callback_rx.cancel(); + } + } + return; + } + let fivs = match load_decommission_entry_versions( &entry, &bucket_name, @@ -4444,6 +4939,13 @@ impl ECStore { remaining += 1; } + if remaining > 0 { + let mut first_path = first_remaining_path.lock().await; + if first_path.is_none() { + *first_path = Some(format!("{bucket_name}/{}", entry.name)); + } + } + versions_found.fetch_add(remaining, Ordering::Relaxed); }) }); @@ -4456,17 +4958,29 @@ impl ECStore { let versions_found = versions_found.load(Ordering::Relaxed); if versions_found > 0 { + let first_remaining_path = first_remaining_path + .lock() + .await + .clone() + .unwrap_or_else(|| format!("{}/", bucket_info.name)); return Err(Error::other(format!( - "at least {versions_found} object(s)/version(s) were found in bucket `{}` after decommissioning", - bucket_info.name + "at least {versions_found} object(s)/version(s) were found in bucket `{}` after decommissioning; first remaining path `{first_remaining_path}`", + bucket_info.name, ))); } } } + self.verify_decommission_durable_ilm_receipts(idx).await?; + Ok(()) } + #[cfg(test)] + pub(crate) async fn check_after_decommission_for_test(self: &Arc, idx: usize) -> Result<()> { + self.check_after_decommission(idx).await + } + #[tracing::instrument(skip(self, rd))] async fn decommission_object( self: Arc, @@ -5777,25 +6291,6 @@ mod pools_tests { ); } - #[test] - fn test_decommission_meta_prefixes_cover_durable_ilm_records() { - let ilm_prefix = format!("{}/", crate::bucket::lifecycle::ILM_META_PREFIX); - let durable_prefixes = [ - crate::bucket::lifecycle::tier_delete_journal::TIER_DELETE_JOURNAL_PREFIX, - crate::bucket::lifecycle::transition_transaction::TRANSITION_TRANSACTION_RECORD_PREFIX, - crate::bucket::lifecycle::manual_transition_job::MANUAL_TRANSITION_JOB_RECORD_PREFIX, - crate::bucket::lifecycle::manual_transition_job::MANUAL_TRANSITION_SCOPE_RECORD_PREFIX, - crate::bucket::lifecycle::manual_transition_job::MANUAL_TRANSITION_TASK_PREFIX, - crate::bucket::lifecycle::manual_transition_job::MANUAL_TRANSITION_WORKER_RESULT_PREFIX, - ]; - - assert!(DECOMMISSION_META_PREFIXES.contains(&crate::bucket::lifecycle::ILM_META_PREFIX)); - assert!( - durable_prefixes.iter().all(|prefix| prefix.starts_with(&ilm_prefix)), - "every durable ILM record must stay under the decommissioned ILM namespace" - ); - } - #[test] fn test_resume_reconciles_missing_decommission_meta_prefixes() { let mut meta = PoolMeta { diff --git a/crates/ecstore/src/store/init.rs b/crates/ecstore/src/store/init.rs index e0452c5bd..fb6dd9379 100644 --- a/crates/ecstore/src/store/init.rs +++ b/crates/ecstore/src/store/init.rs @@ -555,10 +555,18 @@ mod tests { #[cfg(feature = "test-util")] use crate::{ bucket::lifecycle::{ + ILM_META_PREFIX, + bucket_lifecycle_ops::{ManualTransitionRunOptions, recover_manual_transition_jobs_once}, lifecycle::{TRANSITION_PENDING, TransitionOptions}, + manual_transition_job::{ + ManualTransitionJobRecord, ManualTransitionScopeAdmission, ManualTransitionTaskRecord, + ManualTransitionWorkerResult, ManualTransitionWorkerResultRecord, manual_transition_job_record_object_name, + manual_transition_scope_record_object_name, manual_transition_task_object_name, + manual_transition_worker_result_object_name, manual_transition_worker_result_task_key, + }, tier_delete_journal::{ - TIER_DELETE_JOURNAL_PREFIX, persist_tier_delete_journal_entry, recover_tier_delete_journal_entries, - tier_delete_journal_object_name, + TIER_DELETE_JOURNAL_PREFIX, encode_tier_delete_journal_entry, persist_tier_delete_journal_entry, + recover_tier_delete_journal_entries, tier_delete_journal_object_name, }, tier_sweeper::{ Jentry, TierDeleteJournalState, TierDeleteSourceIdentity, transitioned_delete_journal_entry_for_source, @@ -570,12 +578,14 @@ mod tests { delete_transition_candidate_for_operator, finalize_missing_transition_transaction_for_operator, inspect_transition_transaction_for_operator, load_transition_transaction_record, recover_transition_transaction_records, save_transition_transaction_record, + transition_transaction_record_object_name, }, }, bucket::metadata::{BUCKET_LIFECYCLE_CONFIG, BUCKET_VERSIONING_CONFIG}, client::transition_api::ReaderImpl, config::com, - disk::{RUSTFS_META_BUCKET, STORAGE_FORMAT_FILE}, + core::pools::DecomBucketInfo, + disk::{BUCKET_META_PREFIX, RUSTFS_META_BUCKET, STORAGE_FORMAT_FILE}, runtime::{global::set_object_store_resolver, sources as runtime_sources}, services::tier::{ test_util::{MockWarmBackend, MockWarmOp, TransitionCleanupStoreBarrier, register_mock_tier}, @@ -3036,6 +3046,420 @@ mod tests { )); } + #[cfg(feature = "test-util")] + #[tokio::test] + #[serial_test::serial(storage_class_env)] + async fn decommission_migrates_and_verifies_registered_durable_ilm_records() { + 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-decommission", &[4, 4])).await; + + let tier_name = "DECOMMISSION-ILM"; + let backend = register_mock_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 tier_entry = Jentry { + obj_name: "decommissioned-remote-object".to_string(), + version_id: "decommissioned-remote-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 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 { + deployment_id: ctx.deployment_id().expect("test store should initialize deployment id"), + transaction_id: uuid::Uuid::new_v4(), + owner_epoch: uuid::Uuid::new_v4(), + write_id: uuid::Uuid::new_v4(), + source: TransitionSourceIdentity { + bucket: "source-bucket".to_string(), + object: "source-object".to_string(), + version_id: Some(uuid::Uuid::new_v4()), + data_dir: uuid::Uuid::new_v4(), + mod_time_unix_nanos: 1_770_000_000_000_000_000, + size: 42, + etag: "source-etag".to_string(), + version_mode: TransitionSourceVersionMode::Versioned, + }, + tier_name: tier_name.to_string(), + backend_fingerprint: backend_identity, + not_after_unix_nanos: 1_780_000_000_000_000_000, + }) + .expect("transition transaction should build"); + 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"); + + let manual_job_id = uuid::Uuid::new_v4(); + let manual_bucket = format!("manual-decommission-{}", manual_job_id.simple()); + let manual_options = ManualTransitionRunOptions { + prefix: "logs/".to_string(), + tier: Some(tier_name.to_string()), + ..Default::default() + }; + let mut manual_job = ManualTransitionJobRecord::new(manual_job_id, &manual_bucket, &manual_options, "old-owner"); + manual_job.scan_completed = true; + manual_job.report.enqueued = 1; + manual_job.lease_expires_at_unix_nanos = 0; + let manual_scope = ManualTransitionScopeAdmission::from_job(&manual_job); + let task_key = manual_transition_worker_result_task_key(&manual_bucket, "logs/a", None); + let manual_task = ManualTransitionTaskRecord::new(manual_job_id, &task_key, &manual_bucket, "logs/a", None, tier_name); + let manual_result = + ManualTransitionWorkerResultRecord::new(manual_job_id, &task_key, ManualTransitionWorkerResult::Completed); + + let manual_job_path = manual_transition_job_record_object_name(manual_job_id).expect("manual job path should build"); + let manual_scope_path = + manual_transition_scope_record_object_name(&manual_scope.scope_key).expect("manual scope path should build"); + let manual_task_path = + manual_transition_task_object_name(manual_job_id, &task_key).expect("manual task path should build"); + let manual_result_path = manual_transition_worker_result_object_name(manual_job_id, &task_key) + .expect("manual worker result path should build"); + let manual_job_bytes = manual_job.encode().expect("manual job should encode"); + let manual_scope_bytes = serde_json::to_vec(&manual_scope).expect("manual scope should encode"); + let manual_task_bytes = manual_task.encode().expect("manual task should encode"); + let manual_result_bytes = manual_result.encode().expect("manual result should encode"); + + let records = vec![ + (tier_path.clone(), tier_bytes.clone()), + (transaction_path.clone(), transaction_bytes.clone()), + (manual_job_path.clone(), manual_job_bytes.clone()), + (manual_scope_path.clone(), manual_scope_bytes.clone()), + (manual_task_path.clone(), manual_task_bytes.clone()), + (manual_result_path.clone(), manual_result_bytes.clone()), + ]; + for (path, data) in &records { + com::save_config(store.pools[0].clone(), path, data.clone()) + .await + .expect("durable ILM source record should persist"); + } + + let legacy_queue = [com::CONFIG_PREFIX, BUCKET_META_PREFIX] + .into_iter() + .map(|prefix| { + DecomBucketInfo { + name: RUSTFS_META_BUCKET.to_string(), + prefix: prefix.to_string(), + } + .to_string() + }) + .collect(); + let legacy_pool_meta = { + let mut pool_meta = store.pool_meta.write().await; + pool_meta.pools[0].decommission = Some(PoolDecommissionInfo { + queued: true, + queued_buckets: legacy_queue, + ..Default::default() + }); + pool_meta.clone() + }; + legacy_pool_meta + .save(store.pools.clone()) + .await + .expect("legacy decommission queue should persist before restart"); + let mut restarted_pool_meta = PoolMeta::default(); + restarted_pool_meta + .load(store.pools[0].clone(), store.pools.clone()) + .await + .expect("legacy decommission queue should reload after restart"); + *store.pool_meta.write().await = restarted_pool_meta; + store + .promote_queued_decommission_for_test(0) + .await + .expect("legacy queued decommission should resume"); + let expected_ilm_queue = DecomBucketInfo { + name: RUSTFS_META_BUCKET.to_string(), + prefix: ILM_META_PREFIX.to_string(), + } + .to_string(); + { + let pool_meta = store.pool_meta.read().await; + let decommission = pool_meta.pools[0] + .decommission + .as_ref() + .expect("decommission state should remain present"); + assert!(!decommission.queued); + assert!(decommission.queued_buckets.contains(&expected_ilm_queue)); + } + + let ilm_bucket = DecomBucketInfo { + name: RUSTFS_META_BUCKET.to_string(), + prefix: ILM_META_PREFIX.to_string(), + }; + for _ in 0..2 { + store + .decommission_pool_for_test(CancellationToken::new(), 0, store.pools[0].clone(), ilm_bucket.clone()) + .await + .expect("durable ILM decommission should be idempotent"); + } + for (path, expected) in &records { + assert_eq!( + com::read_config(store.pools[0].clone(), path) + .await + .expect("source should remain until the final sweep"), + *expected + ); + assert_eq!( + com::read_config(store.pools[1].clone(), path) + .await + .expect("target should contain the migrated record"), + *expected + ); + } + + com::delete_config(store.pools[1].clone(), &manual_job_path) + .await + .expect("target manual job should delete"); + let missing = store + .verify_and_cleanup_decommissioned_durable_ilm_record_for_test( + 0, + store.pools[0].get_disks_by_key(&manual_job_path), + &manual_job_path, + ) + .await + .expect_err("missing target must block source cleanup"); + let missing = missing.to_string(); + assert!(missing.contains(&manual_job_path) && missing.contains(&manual_job_id.to_string())); + assert_eq!( + com::read_config(store.pools[0].clone(), &manual_job_path) + .await + .expect("missing target must retain source"), + manual_job_bytes + ); + com::save_config(store.pools[1].clone(), &manual_job_path, manual_job_bytes.clone()) + .await + .expect("target manual job should restore"); + + com::save_config(store.pools[1].clone(), &transaction_path, b"{corrupt".to_vec()) + .await + .expect("target transaction should corrupt deterministically"); + let corrupt = store + .verify_and_cleanup_decommissioned_durable_ilm_record_for_test( + 0, + store.pools[0].get_disks_by_key(&transaction_path), + &transaction_path, + ) + .await + .expect_err("corrupt target must block source cleanup"); + let corrupt = corrupt.to_string(); + assert!(corrupt.contains(&transaction_path) && corrupt.contains(&transaction.transaction_id.to_string())); + assert_eq!( + com::read_config(store.pools[0].clone(), &transaction_path) + .await + .expect("corrupt target must retain source"), + transaction_bytes + ); + com::save_config(store.pools[1].clone(), &transaction_path, transaction_bytes.clone()) + .await + .expect("target transaction should restore"); + + com::save_config(store.pools[1].clone(), &manual_scope_path, manual_scope_bytes.clone()) + .await + .expect("target scope rewrite should invalidate cached metadata before the quorum check"); + let target_scope_set = store.pools[1].get_disks_by_key(&manual_scope_path); + let original_target_scope_disks = { + let mut disks = target_scope_set.disks.write().await; + let original = disks.clone(); + for disk in disks.iter_mut().take(3) { + *disk = None; + } + original + }; + let quorum_error = store + .verify_and_cleanup_decommissioned_durable_ilm_record_for_test( + 0, + store.pools[0].get_disks_by_key(&manual_scope_path), + &manual_scope_path, + ) + .await + .expect_err("target below read quorum must block source cleanup"); + *target_scope_set.disks.write().await = original_target_scope_disks; + let quorum_error = quorum_error.to_string(); + assert!(quorum_error.contains(&manual_scope_path) && quorum_error.contains(&manual_job_id.to_string())); + assert!(com::read_config(store.pools[0].clone(), &manual_scope_path).await.is_ok()); + + com::save_config(store.pools[1].clone(), &manual_task_path, manual_task_bytes.clone()) + .await + .expect("target task rewrite should invalidate cached metadata before the quorum check"); + let target_task_set = store.pools[1].get_disks_by_key(&manual_task_path); + let original_target_task_disks = { + let mut disks = target_task_set.disks.write().await; + let original = disks.clone(); + for disk in disks.iter_mut().take(2) { + *disk = None; + } + original + }; + let receipt_quorum_error = store + .verify_and_cleanup_decommissioned_durable_ilm_record_for_test( + 0, + store.pools[0].get_disks_by_key(&manual_task_path), + &manual_task_path, + ) + .await + .expect_err("target read quorum without receipt write quorum must retain the source"); + *target_task_set.disks.write().await = original_target_task_disks; + let receipt_quorum_error = receipt_quorum_error.to_string(); + assert!(receipt_quorum_error.contains("receipt")); + assert!(receipt_quorum_error.contains(&manual_task_path)); + assert!(receipt_quorum_error.contains(&manual_job_id.to_string())); + assert!(com::read_config(store.pools[0].clone(), &manual_task_path).await.is_ok()); + store + .verify_and_cleanup_decommissioned_durable_ilm_record_for_test( + 0, + store.pools[0].get_disks_by_key(&manual_task_path), + &manual_task_path, + ) + .await + .expect("healthy target should persist the receipt before source cleanup"); + + let unknown_path = "ilm/future-durable/jobs/one.json"; + com::save_config(store.pools[0].clone(), unknown_path, b"{}".to_vec()) + .await + .expect("unknown durable ILM record should persist for the guard test"); + let unknown_migration = store + .decommission_pool_for_test(CancellationToken::new(), 0, store.pools[0].clone(), ilm_bucket) + .await + .expect_err("unregistered durable ILM namespace must block migration"); + assert!(unknown_migration.to_string().contains(unknown_path)); + let unknown_final_sweep = store + .check_after_decommission_for_test(0) + .await + .expect_err("unregistered durable ILM namespace must block completion"); + assert!(unknown_final_sweep.to_string().contains(unknown_path)); + com::delete_config(store.pools[0].clone(), unknown_path) + .await + .expect("unknown guard fixture should be removed before the successful final sweep"); + + store + .check_after_decommission_for_test(0) + .await + .expect("production final sweep should validate every target before cleanup"); + assert_eq!( + store + .decommission_durable_ilm_receipt_count_for_test(0) + .await + .expect("durable ILM receipts should be listable"), + records.len(), + "every cleaned source record must have a durable validation receipt" + ); + for (path, expected) in &records { + assert!( + matches!(com::read_config(store.pools[0].clone(), path).await, Err(Error::ConfigNotFound)), + "final sweep should remove the validated source `{path}`" + ); + assert_eq!( + com::read_config(store.pools[1].clone(), path) + .await + .expect("final sweep must preserve the target"), + *expected + ); + } + + let mut crash_restarted_pool_meta = PoolMeta::default(); + crash_restarted_pool_meta + .load(store.pools[0].clone(), store.pools.clone()) + .await + .expect("pool metadata should reload after the simulated pre-complete crash"); + *store.pool_meta.write().await = crash_restarted_pool_meta; + + com::delete_config(store.pools[1].clone(), &manual_job_path) + .await + .expect("post-crash target manual job should delete"); + let missing_after_crash = store + .complete_decommission(0) + .await + .expect_err("completion must reject a missing target after source cleanup and restart") + .to_string(); + assert!(missing_after_crash.contains(&manual_job_path)); + assert!(missing_after_crash.contains(&manual_job_id.to_string())); + assert!( + !store.pool_meta.read().await.pools[0] + .decommission + .as_ref() + .expect("decommission state should survive restart") + .complete + ); + com::save_config(store.pools[1].clone(), &manual_job_path, manual_job_bytes.clone()) + .await + .expect("post-crash target manual job should restore"); + + com::save_config(store.pools[1].clone(), &transaction_path, b"{corrupt".to_vec()) + .await + .expect("post-crash target transaction should corrupt deterministically"); + let corrupt_after_crash = store + .complete_decommission(0) + .await + .expect_err("completion must reject a corrupt target after source cleanup and restart") + .to_string(); + assert!(corrupt_after_crash.contains(&transaction_path)); + assert!(corrupt_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"); + + store + .complete_decommission(0) + .await + .expect("completion should persist before receipt cleanup"); + assert!( + store.pool_meta.read().await.pools[0] + .decommission + .as_ref() + .expect("completed decommission state should remain present") + .complete + ); + assert_eq!( + store + .decommission_durable_ilm_receipt_count_for_test(0) + .await + .expect("receipt cleanup should be observable"), + 0 + ); + store + .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) + )); + } + #[cfg(feature = "test-util")] async fn tier_delete_journal_count(store: Arc) -> usize { store