From 156e21c90e7900890581d9fa06eeaeade2216fbd Mon Sep 17 00:00:00 2001 From: cxymds Date: Tue, 16 Jun 2026 08:45:49 +0800 Subject: [PATCH] fix(lifecycle): make tier cleanup recoverable (#3491) --- .../bucket/lifecycle/bucket_lifecycle_ops.rs | 360 ++++++++++++-- crates/ecstore/src/bucket/lifecycle/mod.rs | 1 + .../lifecycle/tier_free_version_recovery.rs | 469 ++++++++++++++++++ .../src/bucket/lifecycle/tier_sweeper.rs | 156 +++++- crates/ecstore/src/pools.rs | 112 ++++- crates/ecstore/src/rebalance.rs | 47 +- crates/ecstore/src/store_api/types.rs | 1 + crates/ecstore/src/store_list_objects.rs | 36 +- crates/filemeta/src/metacache.rs | 213 +++++++- scripts/test/tier_lifecycle_cleanup_stress.md | 45 ++ 10 files changed, 1357 insertions(+), 83 deletions(-) create mode 100644 crates/ecstore/src/bucket/lifecycle/tier_free_version_recovery.rs create mode 100644 scripts/test/tier_lifecycle_cleanup_stress.md diff --git a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs index 478ac34d5..cff15e264 100644 --- a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs +++ b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs @@ -17,8 +17,9 @@ use crate::bucket::lifecycle::evaluator::Evaluator; use crate::bucket::lifecycle::lifecycle::{ self, ExpirationOptions, Lifecycle, ObjectOpts, TransitionOptions, abort_incomplete_multipart_upload_due, }; +use crate::bucket::lifecycle::tier_free_version_recovery::{DEFAULT_FREE_VERSION_RECOVERY_LIMIT, recover_tier_free_versions}; use crate::bucket::lifecycle::tier_last_day_stats::{DailyAllTierStats, LastDayTierStats}; -use crate::bucket::lifecycle::tier_sweeper::{Jentry, delete_object_from_remote_tier}; +use crate::bucket::lifecycle::tier_sweeper::{Jentry, delete_object_from_remote_tier_idempotent}; use crate::bucket::object_lock::objectlock_sys::check_object_lock_for_deletion; use crate::bucket::replication::{ DeletedObjectReplicationInfo, ReplicationConfig, check_replicate_delete, schedule_replication_delete, @@ -71,7 +72,7 @@ use std::collections::{HashMap, HashSet}; use std::env; use std::pin::Pin; use std::sync::atomic::{AtomicI64, Ordering}; -use std::sync::{Arc, Mutex, Weak}; +use std::sync::{Arc, Mutex, OnceLock, Weak}; use std::time::Duration as StdDuration; use time::OffsetDateTime; use tokio::select; @@ -99,6 +100,7 @@ pub type TraceFn = pub type ExpiryOpType = Box; static XXHASH_SEED: u64 = 0; +static TIER_FREE_VERSION_RECOVERY_STARTED: OnceLock<()> = OnceLock::new(); pub const AMZ_OBJECT_TAGGING: &str = "X-Amz-Tagging"; pub const AMZ_TAG_COUNT: &str = "x-amz-tagging-count"; @@ -441,6 +443,23 @@ impl ExpiryState { })) } + #[cfg(test)] + fn new_with_unconsumed_worker_channel(capacity: usize) -> Arc> { + let (tx, rx) = mpsc::channel(capacity); + Arc::new(RwLock::new(Self { + tasks_tx: vec![tx], + tasks_rx: vec![Arc::new(tokio::sync::Mutex::new(rx))], + stats: Arc::new(ExpiryStats { + missed_expiry_tasks: AtomicI64::new(0), + missed_freevers_tasks: AtomicI64::new(0), + missed_tier_journal_tasks: AtomicI64::new(0), + pending_tasks: AtomicI64::new(0), + active_tasks: AtomicI64::new(0), + workers: AtomicI64::new(1), + }), + })) + } + pub fn pending_tasks(&self) -> usize { usize::try_from(self.stats.pending_tasks().max(0)).unwrap_or(usize::MAX) } @@ -470,13 +489,13 @@ impl ExpiryState { Ok(()) } - pub async fn enqueue_free_version(&mut self, oi: ObjectInfo) { + pub async fn enqueue_free_version(&mut self, oi: ObjectInfo) -> bool { let task = FreeVersionTask(oi); let wrkr = self.get_worker_ch(task.op_hash()); if wrkr.is_none() { self.stats.increment_missed_freevers_tasks(); self.stats.record_scanner_expiry_state(); - return; + return false; } let wrkr = wrkr.expect("worker channel should exist after None check"); let queued = self.send_expiry_task(wrkr, Box::new(task)).await; @@ -484,6 +503,7 @@ impl ExpiryState { self.stats.increment_missed_freevers_tasks(); } self.stats.record_scanner_expiry_state(); + queued } pub async fn enqueue_by_days(&mut self, oi: &ObjectInfo, event: &lifecycle::Event, src: &LcEventSrc) -> bool { @@ -637,7 +657,7 @@ impl ExpiryState { } else if v.as_any().is::() { let v = v.as_any().downcast_ref::().expect("Jentry downcast failed"); - if let Err(err) = delete_object_from_remote_tier(&v.obj_name, &v.version_id, &v.tier_name).await { + if let Err(err) = delete_object_from_remote_tier_idempotent(&v.obj_name, &v.version_id, &v.tier_name).await { debug!( event = EVENT_LIFECYCLE_WORKER_STATE, component = LOG_COMPONENT_ECSTORE, @@ -654,7 +674,7 @@ impl ExpiryState { else if v.as_any().is::() { let v = v.as_any().downcast_ref::().expect("FreeVersionTask downcast failed"); let oi = v.0.clone(); - if let Err(err) = delete_object_from_remote_tier( + if let Err(err) = delete_object_from_remote_tier_idempotent( &oi.transitioned_object.name, &oi.transitioned_object.version_id, &oi.transitioned_object.tier, @@ -744,6 +764,33 @@ impl ExpiryState { } } +async fn enqueue_recovered_free_version_with_state(state: &Arc>, oi: ObjectInfo) -> bool { + let task = FreeVersionTask(oi); + let hash = task.op_hash(); + let (wrkr, stats) = { + let state = state.read().await; + (state.get_worker_ch(hash), Arc::clone(&state.stats)) + }; + let Some(wrkr) = wrkr else { + stats.increment_missed_freevers_tasks(); + stats.record_scanner_expiry_state(); + return false; + }; + + let queued = wrkr.try_send(Some(Box::new(task))).is_ok(); + if !queued { + stats.increment_missed_freevers_tasks(); + } else { + stats.increment_pending_tasks(); + } + stats.record_scanner_expiry_state(); + queued +} + +pub async fn enqueue_recovered_free_version(oi: ObjectInfo) -> bool { + enqueue_recovered_free_version_with_state(&GLOBAL_ExpiryState, oi).await +} + struct TransitionTask { obj_info: ObjectInfo, src: LcEventSrc, @@ -1303,7 +1350,78 @@ pub async fn init_background_expiry(api: Arc) { } //let expiry_state = GLOBAL_ExpiryStSate.write().await; - ExpiryState::resize_workers(workers, api).await; + ExpiryState::resize_workers(workers, api.clone()).await; + spawn_tier_free_version_recovery_once(api); +} + +fn spawn_tier_free_version_recovery_once(api: Arc) { + if TIER_FREE_VERSION_RECOVERY_STARTED.set(()).is_err() { + return; + } + + tokio::spawn(async move { + let cancel_token = crate::global::get_background_services_cancel_token() + .cloned() + .unwrap_or_else(CancellationToken::new); + let mut interval = tokio::time::interval(StdDuration::from_secs(60)); + let mut bucket_marker: Option = None; + let mut object_marker: Option = None; + + loop { + select! { + _ = cancel_token.cancelled() => return, + _ = interval.tick() => {} + } + + let started_at = std::time::Instant::now(); + match recover_tier_free_versions( + api.clone(), + DEFAULT_FREE_VERSION_RECOVERY_LIMIT, + bucket_marker.clone(), + object_marker.clone(), + ) + .await + { + Ok(stats) => { + bucket_marker = stats.next_bucket_marker; + object_marker = stats.next_object_marker; + let (pending_tasks, active_tasks) = { + let state = GLOBAL_ExpiryState.read().await; + (state.pending_tasks(), state.stats.active_tasks()) + }; + debug!( + event = EVENT_LIFECYCLE_WORKER_STATE, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_LIFECYCLE, + duration_ms = started_at.elapsed().as_millis(), + scanned = stats.scanned, + scanned_entries = stats.scanned_entries, + buckets_scanned = stats.buckets_scanned, + enqueued = stats.enqueued, + failed = stats.failed, + truncated = stats.truncated, + next_bucket_marker = ?bucket_marker, + next_object_marker = ?object_marker, + pending_tasks, + active_tasks, + "Recovered tier free-version cleanup tasks" + ); + } + Err(err) => { + warn!( + event = EVENT_LIFECYCLE_WORKER_STATE, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_LIFECYCLE, + duration_ms = started_at.elapsed().as_millis(), + next_bucket_marker = ?bucket_marker, + next_object_marker = ?object_marker, + error = ?err, + "Failed to recover tier free-version cleanup tasks" + ); + } + } + } + }); } #[derive(Debug, Clone)] @@ -1751,6 +1869,17 @@ fn mark_delete_opts_skip_decommissioned_on_remote_success(opts: &mut ObjectOptio } } +fn transitioned_cleanup_tuple(oi: &ObjectInfo) -> Result<(&str, &str, &str), std::io::Error> { + let transitioned = &oi.transitioned_object; + if transitioned.status != lifecycle::TRANSITION_COMPLETE { + return Err(std::io::Error::other("transitioned object cleanup tuple is not complete")); + } + if transitioned.name.is_empty() || transitioned.version_id.is_empty() || transitioned.tier.is_empty() { + return Err(std::io::Error::other("transitioned object cleanup tuple is incomplete")); + } + Ok((&transitioned.name, &transitioned.version_id, &transitioned.tier)) +} + pub async fn enqueue_transition_immediate(oi: &ObjectInfo, src: LcEventSrc) { if let Some(lc) = GLOBAL_LifecycleSys.get(&oi.bucket).await { enqueue_transition_with_lifecycle(oi, &lc, &src).await; @@ -1984,6 +2113,7 @@ pub async fn expire_transitioned_object( //let traceFn = GLOBAL_LifecycleSys.trace(oi); let mut opts = ObjectOptions { versioned: BucketVersioningSys::prefix_enabled(&oi.bucket, &oi.name).await, + version_suspended: BucketVersioningSys::prefix_suspended(&oi.bucket, &oi.name).await, expiration: ExpirationOptions { expire: true }, ..Default::default() }; @@ -2002,29 +2132,12 @@ pub async fn expire_transitioned_object( }; } - let ret = delete_object_from_remote_tier( - &oi.transitioned_object.name, - &oi.transitioned_object.version_id, - &oi.transitioned_object.tier, - ) - .await; - if let Err(e) = &ret { - error!( - event = EVENT_LIFECYCLE_TIER_OPERATION_FAILED, - component = LOG_COMPONENT_ECSTORE, - subsystem = LOG_SUBSYSTEM_LIFECYCLE, - bucket = %oi.bucket, - object = %oi.name, - tier = %oi.transitioned_object.tier, - tier_object = %oi.transitioned_object.name, - tier_version_id = %oi.transitioned_object.version_id, - operation = "delete_remote_transitioned_object", - error = ?e, - "Lifecycle tier operation failed" - ); - } - mark_delete_opts_skip_decommissioned_on_remote_success(&mut opts, ret.is_ok()); + let (_remote_object, _remote_version, _tier) = transitioned_cleanup_tuple(oi)?; + // Delete local metadata first so concurrent GET cannot observe metadata + // pointing to a remote tier version that has already been removed. If this + // only creates a delete marker, remote cleanup must be driven by persisted + // free-version recovery rather than the visible delete result. let dobj = match api.delete_object(&oi.bucket, &oi.name, opts).await { Ok(obj) => obj, Err(e) => { @@ -2038,8 +2151,7 @@ pub async fn expire_transitioned_object( error = ?e, "Lifecycle delete failed" ); - // Return the original object info if deletion fails - oi.clone() + return Err(std::io::Error::other(e)); } }; @@ -2716,14 +2828,16 @@ mod tests { use super::{ DATE_EXPIRY_EXISTING_OBJECTS_GRACE_SECS, DEFAULT_TRANSITION_QUEUE_CAPACITY, DEFAULT_TRANSITION_WORKERS_ABSOLUTE_MAX, DEFAULT_TRANSITION_WORKERS_CAP, ExpiryState, GLOBAL_TransitionState, StaleMultipartUploadCandidate, TransitionState, - cleanup_empty_multipart_sha_dirs_on_local_disks, cleanup_stale_multipart_uploads_once_at, lifecycle_deleted_object, - lifecycle_rule_has_date_expiration, lifecycle_version_purge_state_from_completed_targets, - mark_delete_opts_skip_decommissioned_on_remote_success, merge_stale_multipart_candidate, replication_state_for_delete, - resolve_transition_queue_capacity, resolve_transition_queue_send_timeout, resolve_transition_worker_count, - resolve_transition_workers_absolute_max, should_defer_date_expiry_for_recent_config_update, - should_reuse_lifecycle_delete_replication_state, + TransitionedObject, cleanup_empty_multipart_sha_dirs_on_local_disks, cleanup_stale_multipart_uploads_once_at, + enqueue_recovered_free_version_with_state, lifecycle_deleted_object, lifecycle_rule_has_date_expiration, + lifecycle_version_purge_state_from_completed_targets, mark_delete_opts_skip_decommissioned_on_remote_success, + merge_stale_multipart_candidate, replication_state_for_delete, resolve_transition_queue_capacity, + resolve_transition_queue_send_timeout, resolve_transition_worker_count, resolve_transition_workers_absolute_max, + should_defer_date_expiry_for_recent_config_update, should_reuse_lifecycle_delete_replication_state, + transitioned_cleanup_tuple, }; use crate::bucket::lifecycle::bucket_lifecycle_audit::LcEventSrc; + use crate::bucket::lifecycle::core::ExpirationOptions; use crate::bucket::metadata::BUCKET_LIFECYCLE_CONFIG; use crate::bucket::metadata_sys; use crate::disk::RUSTFS_META_MULTIPART_BUCKET; @@ -2779,6 +2893,104 @@ mod tests { assert_eq!(expiry.queue_missed, 1); } + #[tokio::test] + async fn enqueue_free_version_reports_false_without_worker_channel() { + let state = ExpiryState::new(); + let mut state = state.write().await; + let oi = ObjectInfo { + bucket: "bucket".to_string(), + name: "object".to_string(), + transitioned_object: TransitionedObject { + name: "remote/object".to_string(), + version_id: "remote-version".to_string(), + tier: "WARM".to_string(), + free_version: true, + ..Default::default() + }, + ..Default::default() + }; + + let queued = state.enqueue_free_version(oi).await; + + assert!(!queued); + assert_eq!(state.stats.missed_free_vers_tasks(), 1); + } + + #[tokio::test] + async fn enqueue_recovered_free_version_reports_false_without_worker_channel() { + let state = ExpiryState::new(); + let oi = ObjectInfo { + bucket: "bucket".to_string(), + name: "object".to_string(), + transitioned_object: TransitionedObject { + name: "remote/object".to_string(), + version_id: "remote-version".to_string(), + tier: "WARM".to_string(), + free_version: true, + ..Default::default() + }, + ..Default::default() + }; + + let queued = enqueue_recovered_free_version_with_state(&state, oi).await; + let state = state.read().await; + + assert!(!queued); + assert_eq!(state.stats.missed_free_vers_tasks(), 1); + } + + #[tokio::test] + async fn enqueue_recovered_free_version_reports_false_when_worker_queue_full() { + let state = ExpiryState::new_with_unconsumed_worker_channel(1); + let oi = ObjectInfo { + bucket: "bucket".to_string(), + name: "object".to_string(), + transitioned_object: TransitionedObject { + name: "remote/object".to_string(), + version_id: "remote-version".to_string(), + tier: "WARM".to_string(), + free_version: true, + ..Default::default() + }, + ..Default::default() + }; + + let first = enqueue_recovered_free_version_with_state(&state, oi.clone()).await; + let second = enqueue_recovered_free_version_with_state(&state, oi).await; + let state = state.read().await; + + assert!(first); + assert!(!second); + assert_eq!(state.stats.pending_tasks(), 1); + assert_eq!(state.stats.missed_free_vers_tasks(), 1); + } + + #[tokio::test] + async fn lifecycle_free_version_recovery_enqueue_reports_retryable_queue_failure() { + let state = ExpiryState::new_with_unconsumed_worker_channel(1); + let oi = ObjectInfo { + bucket: "bucket".to_string(), + name: "object".to_string(), + transitioned_object: TransitionedObject { + name: "remote/object".to_string(), + version_id: "remote-version".to_string(), + tier: "WARM".to_string(), + free_version: true, + ..Default::default() + }, + ..Default::default() + }; + + let first = enqueue_recovered_free_version_with_state(&state, oi.clone()).await; + let second = enqueue_recovered_free_version_with_state(&state, oi).await; + let state = state.read().await; + + assert!(first); + assert!(!second); + assert_eq!(state.stats.pending_tasks(), 1); + assert_eq!(state.stats.missed_free_vers_tasks(), 1); + } + #[tokio::test] #[serial] async fn scanner_transition_enqueue_reports_full_queue() { @@ -2839,6 +3051,80 @@ mod tests { assert!(opts.skip_decommissioned); } + #[test] + fn transitioned_expiry_must_not_skip_free_version_before_remote_cleanup() { + let mut opts = ObjectOptions::default(); + + mark_delete_opts_skip_decommissioned_on_remote_success(&mut opts, false); + + assert!(!opts.skip_decommissioned); + assert!(!opts.skip_free_version); + } + + #[test] + fn transitioned_cleanup_tuple_requires_remote_name_version_and_tier() { + let mut oi = ObjectInfo::default(); + oi.transitioned_object.status = crate::bucket::lifecycle::lifecycle::TRANSITION_COMPLETE.to_string(); + oi.transitioned_object.name = "remote/object".to_string(); + oi.transitioned_object.version_id = "remote-version".to_string(); + oi.transitioned_object.tier = "WARM".to_string(); + + let tuple = transitioned_cleanup_tuple(&oi).expect("complete tuple should be accepted"); + + assert_eq!(tuple, ("remote/object", "remote-version", "WARM")); + } + + #[test] + fn transitioned_cleanup_tuple_rejects_missing_remote_version() { + let mut oi = ObjectInfo::default(); + oi.transitioned_object.status = crate::bucket::lifecycle::lifecycle::TRANSITION_COMPLETE.to_string(); + oi.transitioned_object.name = "remote/object".to_string(); + oi.transitioned_object.tier = "WARM".to_string(); + + let err = transitioned_cleanup_tuple(&oi).expect_err("missing version must be rejected"); + + assert!(err.to_string().contains("cleanup tuple is incomplete")); + } + + #[test] + fn transitioned_cleanup_tuple_rejects_non_complete_status() { + let mut oi = ObjectInfo::default(); + oi.transitioned_object.name = "remote/object".to_string(); + oi.transitioned_object.version_id = "remote-version".to_string(); + oi.transitioned_object.tier = "WARM".to_string(); + oi.transitioned_object.status = "pending".to_string(); + + let err = transitioned_cleanup_tuple(&oi).expect_err("non-complete transition must be rejected"); + + assert!(err.to_string().contains("not complete")); + } + + #[test] + fn transitioned_expiry_sets_version_suspended_in_delete_options() { + let opts = ObjectOptions { + versioned: true, + version_suspended: true, + expiration: ExpirationOptions { expire: true }, + ..Default::default() + }; + + assert!(opts.versioned); + assert!(opts.version_suspended); + assert!(!opts.skip_free_version); + } + + #[test] + fn delete_marker_result_must_not_drive_remote_cleanup() { + let dobj = ObjectInfo { + delete_marker: true, + transitioned_object: TransitionedObject::default(), + ..Default::default() + }; + + assert!(dobj.delete_marker); + assert!(dobj.transitioned_object.name.is_empty()); + } + // SAFETY: this helper is only used from `#[serial]` tests and those tests run under a // single-thread runtime (`worker_threads = 1`), so no concurrent reader/writer can access // process environment while `env::set_var`/`env::remove_var` is active. diff --git a/crates/ecstore/src/bucket/lifecycle/mod.rs b/crates/ecstore/src/bucket/lifecycle/mod.rs index 6a5233811..7a853c4b7 100644 --- a/crates/ecstore/src/bucket/lifecycle/mod.rs +++ b/crates/ecstore/src/bucket/lifecycle/mod.rs @@ -18,5 +18,6 @@ pub mod core; pub mod evaluator; pub use self::core as lifecycle; pub mod rule; +pub mod tier_free_version_recovery; pub mod tier_last_day_stats; pub mod tier_sweeper; diff --git a/crates/ecstore/src/bucket/lifecycle/tier_free_version_recovery.rs b/crates/ecstore/src/bucket/lifecycle/tier_free_version_recovery.rs new file mode 100644 index 000000000..fa496b2db --- /dev/null +++ b/crates/ecstore/src/bucket/lifecycle/tier_free_version_recovery.rs @@ -0,0 +1,469 @@ +// 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 std::sync::Arc; + +use tokio::sync::mpsc; +use tokio_util::sync::CancellationToken; + +use crate::disk::RUSTFS_META_BUCKET; +use crate::error::Result; +use crate::store::ECStore; +use crate::store_api::{BucketOperations, BucketOptions, ListOperations, ObjectInfo, ObjectInfoOrErr, WalkOptions}; + +pub const DEFAULT_FREE_VERSION_RECOVERY_LIMIT: usize = 1_000; +const DEFAULT_FREE_VERSION_RECOVERY_SCAN_LIMIT: usize = 10_000; + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct FreeVersionRecoveryStats { + pub scanned: usize, + pub enqueued: usize, + pub failed: usize, + pub next_bucket_marker: Option, + pub next_object_marker: Option, + pub scanned_entries: usize, + pub buckets_scanned: usize, + pub truncated: bool, +} + +#[derive(Debug, Clone)] +pub struct FreeVersionRecoveryPage { + pub items: Vec, + pub next_bucket_marker: Option, + pub next_object_marker: Option, + pub scanned_entries: usize, + pub buckets_scanned: usize, + pub truncated: bool, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +struct RecoveryCursor { + bucket: String, + object: String, +} + +pub async fn recover_tier_free_versions( + api: Arc, + limit: usize, + bucket_marker: Option, + object_marker: Option, +) -> Result { + if limit == 0 { + return Err(std::io::Error::other("free-version recovery limit must be greater than zero").into()); + } + + let page = list_tier_free_versions(api, limit, bucket_marker.clone(), object_marker.clone()).await?; + let mut stats = FreeVersionRecoveryStats { + scanned: 0, + enqueued: 0, + failed: 0, + next_bucket_marker: page.next_bucket_marker, + next_object_marker: page.next_object_marker, + scanned_entries: page.scanned_entries, + buckets_scanned: page.buckets_scanned, + truncated: page.truncated, + }; + + let mut retry_cursor = RetryCursor::new(bucket_marker, object_marker); + for oi in page.items { + retry_cursor.visit(&oi); + if !record_recovered_free_version_enqueue(&mut stats, queue_recovered_free_version(oi).await) { + let (bucket_marker, object_marker) = retry_cursor.retry_markers(); + stats.truncated = true; + stats.next_bucket_marker = bucket_marker; + stats.next_object_marker = object_marker; + break; + } + } + + Ok(stats) +} + +fn record_recovered_free_version_enqueue(stats: &mut FreeVersionRecoveryStats, queued: bool) -> bool { + stats.scanned += 1; + if queued { + stats.enqueued += 1; + true + } else { + stats.failed += 1; + false + } +} + +async fn queue_recovered_free_version(oi: ObjectInfo) -> bool { + crate::bucket::lifecycle::bucket_lifecycle_ops::enqueue_recovered_free_version(oi).await +} + +#[derive(Debug, Clone, PartialEq, Eq)] +struct RetryCursor { + input_bucket_marker: Option, + input_object_marker: Option, + current: Option, + completed: Option, +} + +impl RetryCursor { + fn new(input_bucket_marker: Option, input_object_marker: Option) -> Self { + Self { + input_bucket_marker, + input_object_marker, + current: None, + completed: None, + } + } + + fn visit(&mut self, oi: &ObjectInfo) { + let cursor = RecoveryCursor { + bucket: oi.bucket.clone(), + object: oi.name.clone(), + }; + if self.current.as_ref() == Some(&cursor) { + return; + } + if let Some(previous) = self.current.replace(cursor) { + self.completed = Some(previous); + } + } + + fn retry_markers(&self) -> (Option, Option) { + if let Some(completed) = &self.completed { + return (Some(completed.bucket.clone()), Some(completed.object.clone())); + } + if let Some(current) = &self.current { + if self.input_bucket_marker.as_deref() == Some(current.bucket.as_str()) { + return (Some(current.bucket.clone()), self.input_object_marker.clone()); + } + return (Some(current.bucket.clone()), None); + } + (self.input_bucket_marker.clone(), self.input_object_marker.clone()) + } +} + +async fn list_tier_free_versions( + api: Arc, + limit: usize, + bucket_marker: Option, + object_marker: Option, +) -> Result { + let mut page = FreeVersionRecoveryPage { + items: Vec::new(), + next_bucket_marker: None, + next_object_marker: None, + scanned_entries: 0, + buckets_scanned: 0, + truncated: false, + }; + + if limit == 0 { + return Ok(page); + } + + let buckets = api.list_bucket(&BucketOptions::default()).await?; + let mut bucket_seen = bucket_marker.is_none(); + let mut truncated_after: Option = None; + let walk_scan_limit = recovery_walk_scan_limit(limit); + + for bucket in buckets { + if bucket.name == RUSTFS_META_BUCKET { + continue; + } + if !bucket_seen { + if bucket_marker.as_deref() == Some(bucket.name.as_str()) { + bucket_seen = true; + } else { + continue; + } + } + + page.buckets_scanned += 1; + let bucket_object_marker = if bucket_marker.as_deref() == Some(bucket.name.as_str()) { + object_marker.clone() + } else { + None + }; + + let (tx, mut rx) = mpsc::channel::(100); + let cancel = CancellationToken::new(); + let mut draining_after_truncation = false; + let mut last_seen_object: Option = None; + let mut scanned_objects = 0usize; + let walk = tokio::spawn({ + let api = api.clone(); + let bucket_name = bucket.name.clone(); + let object_marker = bucket_object_marker.clone(); + let cancel = cancel.clone(); + async move { + api.walk( + cancel, + &bucket_name, + "", + tx, + WalkOptions { + include_free_versions: true, + limit: walk_scan_limit, + marker: object_marker, + ..Default::default() + }, + ) + .await + } + }); + + while let Some(item) = rx.recv().await { + page.scanned_entries += 1; + if draining_after_truncation { + continue; + } + if let Some(err) = item.err { + cancel.cancel(); + walk.await.map_err(|err| std::io::Error::other(err.to_string()))??; + return Err(err); + } + let Some(oi) = item.item else { + continue; + }; + record_scanned_object(&mut last_seen_object, &mut scanned_objects, &oi.name); + if let Some(cursor) = &truncated_after + && cursor.object != oi.name + { + page.truncated = true; + cancel.cancel(); + draining_after_truncation = true; + continue; + } + if is_recoverable_tier_free_version(&oi) { + let cursor = RecoveryCursor { + bucket: bucket.name.clone(), + object: oi.name.clone(), + }; + page.items.push(oi); + page.next_bucket_marker = Some(cursor.bucket.clone()); + page.next_object_marker = Some(cursor.object.clone()); + if page.items.len() >= limit && truncated_after.is_none() { + truncated_after = Some(cursor); + } + } + } + + walk.await.map_err(|err| std::io::Error::other(err.to_string()))??; + mark_scan_truncated_if_needed(&mut page, scanned_objects, walk_scan_limit, &bucket.name, last_seen_object.as_deref()); + + if page.truncated { + page.next_bucket_marker = Some(bucket.name.clone()); + break; + } + } + + if !page.truncated { + page.next_bucket_marker = None; + page.next_object_marker = None; + } + + Ok(page) +} + +fn recovery_walk_scan_limit(limit: usize) -> usize { + DEFAULT_FREE_VERSION_RECOVERY_SCAN_LIMIT.max(limit.saturating_add(1)) +} + +fn record_scanned_object(last_seen_object: &mut Option, scanned_objects: &mut usize, object: &str) { + if last_seen_object.as_deref() == Some(object) { + return; + } + *last_seen_object = Some(object.to_string()); + *scanned_objects = scanned_objects.saturating_add(1); +} + +fn mark_scan_truncated_if_needed( + page: &mut FreeVersionRecoveryPage, + scanned_objects: usize, + walk_scan_limit: usize, + bucket: &str, + last_seen_object: Option<&str>, +) { + if page.truncated || scanned_objects < walk_scan_limit { + return; + } + if let Some(last_seen_object) = last_seen_object { + page.truncated = true; + page.next_bucket_marker = Some(bucket.to_string()); + page.next_object_marker = Some(last_seen_object.to_string()); + } +} + +fn is_recoverable_tier_free_version(oi: &ObjectInfo) -> bool { + oi.transitioned_object.free_version + && !oi.transitioned_object.name.is_empty() + && !oi.transitioned_object.version_id.is_empty() + && !oi.transitioned_object.tier.is_empty() +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn recovery_stats_count_failed_enqueue_without_treating_scan_as_lost() { + let mut stats = FreeVersionRecoveryStats { + scanned: 0, + enqueued: 0, + failed: 0, + next_bucket_marker: None, + next_object_marker: None, + scanned_entries: 1, + buckets_scanned: 1, + truncated: false, + }; + + record_recovered_free_version_enqueue(&mut stats, false); + + assert_eq!(stats.scanned, 1); + assert_eq!(stats.enqueued, 0); + assert_eq!(stats.failed, 1); + } + + #[test] + fn recovery_enqueue_failure_returns_false_so_cursor_can_retry_same_object() { + let mut stats = FreeVersionRecoveryStats { + scanned: 0, + enqueued: 0, + failed: 0, + next_bucket_marker: Some("bucket".to_string()), + next_object_marker: Some("object".to_string()), + scanned_entries: 1, + buckets_scanned: 1, + truncated: false, + }; + + let queued = record_recovered_free_version_enqueue(&mut stats, false); + + assert!(!queued); + assert_eq!(stats.scanned, 1); + assert_eq!(stats.failed, 1); + } + + #[test] + fn recover_tier_free_versions_failed_enqueue_retries_same_persisted_cursor() { + let mut stats = FreeVersionRecoveryStats { + scanned: 0, + enqueued: 0, + failed: 0, + next_bucket_marker: Some("bucket".to_string()), + next_object_marker: Some("object-a".to_string()), + scanned_entries: 1, + buckets_scanned: 1, + truncated: false, + }; + let mut cursor = RetryCursor::new(stats.next_bucket_marker.clone(), stats.next_object_marker.clone()); + let oi = object_info("bucket", "object-b"); + + cursor.visit(&oi); + let queued = record_recovered_free_version_enqueue(&mut stats, false); + let (bucket_marker, object_marker) = cursor.retry_markers(); + + assert!(!queued); + assert_eq!(bucket_marker, Some("bucket".to_string())); + assert_eq!(object_marker, Some("object-a".to_string())); + assert_eq!(stats.failed, 1); + } + + #[test] + fn retry_cursor_rewinds_to_previous_object_on_enqueue_failure() { + let mut cursor = RetryCursor::new(Some("bucket".to_string()), Some("object-a".to_string())); + cursor.visit(&object_info("bucket", "object-b")); + cursor.visit(&object_info("bucket", "object-b")); + + assert_eq!(cursor.retry_markers(), (Some("bucket".to_string()), Some("object-a".to_string()))); + + cursor.visit(&object_info("bucket", "object-c")); + + assert_eq!(cursor.retry_markers(), (Some("bucket".to_string()), Some("object-b".to_string()))); + } + + #[test] + fn retry_cursor_rewinds_to_bucket_start_when_first_bucket_object_fails() { + let mut cursor = RetryCursor::new(None, None); + cursor.visit(&object_info("bucket", "object-a")); + + assert_eq!(cursor.retry_markers(), (Some("bucket".to_string()), None)); + } + + #[test] + fn recoverable_tier_free_version_requires_complete_remote_tuple() { + let mut oi = ObjectInfo::default(); + oi.transitioned_object.free_version = true; + oi.transitioned_object.name = "remote/object".to_string(); + oi.transitioned_object.version_id = "remote-version".to_string(); + oi.transitioned_object.tier = "WARM".to_string(); + + assert!(is_recoverable_tier_free_version(&oi)); + + oi.transitioned_object.version_id.clear(); + assert!(!is_recoverable_tier_free_version(&oi)); + } + + #[test] + fn recovery_scan_limit_is_independent_from_enqueue_limit() { + assert_eq!(recovery_walk_scan_limit(1), DEFAULT_FREE_VERSION_RECOVERY_SCAN_LIMIT); + assert_eq!( + recovery_walk_scan_limit(DEFAULT_FREE_VERSION_RECOVERY_SCAN_LIMIT), + DEFAULT_FREE_VERSION_RECOVERY_SCAN_LIMIT + 1 + ); + } + + #[test] + fn scan_truncation_keeps_marker_after_nonrecoverable_window() { + let mut page = FreeVersionRecoveryPage { + items: Vec::new(), + next_bucket_marker: None, + next_object_marker: None, + scanned_entries: DEFAULT_FREE_VERSION_RECOVERY_SCAN_LIMIT, + buckets_scanned: 1, + truncated: false, + }; + + mark_scan_truncated_if_needed( + &mut page, + DEFAULT_FREE_VERSION_RECOVERY_SCAN_LIMIT, + DEFAULT_FREE_VERSION_RECOVERY_SCAN_LIMIT, + "bucket", + Some("object-z"), + ); + + assert!(page.truncated); + assert_eq!(page.next_bucket_marker, Some("bucket".to_string())); + assert_eq!(page.next_object_marker, Some("object-z".to_string())); + } + + #[test] + fn scanned_object_count_advances_only_on_new_objects() { + let mut last_seen = None; + let mut scanned_objects = 0; + + record_scanned_object(&mut last_seen, &mut scanned_objects, "object-a"); + record_scanned_object(&mut last_seen, &mut scanned_objects, "object-a"); + record_scanned_object(&mut last_seen, &mut scanned_objects, "object-b"); + + assert_eq!(scanned_objects, 2); + assert_eq!(last_seen, Some("object-b".to_string())); + } + + fn object_info(bucket: &str, object: &str) -> ObjectInfo { + ObjectInfo { + bucket: bucket.to_string(), + name: object.to_string(), + ..Default::default() + } + } +} diff --git a/crates/ecstore/src/bucket/lifecycle/tier_sweeper.rs b/crates/ecstore/src/bucket/lifecycle/tier_sweeper.rs index 26f07c9ab..ed3d318a7 100644 --- a/crates/ecstore/src/bucket/lifecycle/tier_sweeper.rs +++ b/crates/ecstore/src/bucket/lifecycle/tier_sweeper.rs @@ -45,6 +45,8 @@ const DEFAULT_REMOTE_DELETE_BREAKER_WINDOW_SECS: usize = 30; const METRIC_DELETE_REMOTE_FAILED_TOTAL: &str = "rustfs_delete_remote_failed_total"; const METRIC_DELETE_REMOTE_BREAKER_TOTAL: &str = "rustfs_delete_remote_breaker_total"; const METRIC_DELETE_REMOTE_INFLIGHT: &str = "rustfs_delete_remote_inflight"; +const ERR_REMOTE_DELETE_BREAKER_OPEN: &str = "remote tier delete breaker is open due to signer/header failures"; +const ERR_REMOTE_DELETE_LIMITER_CLOSED: &str = "remote tier delete limiter is closed"; static REMOTE_DELETE_INFLIGHT: AtomicUsize = AtomicUsize::new(0); @@ -63,6 +65,11 @@ static REMOTE_DELETE_BREAKER: LazyLock> = LazyLock::n )) }); +#[cfg(test)] +static REMOTE_TIER_DELETE_TEST_HOOK: std::sync::LazyLock< + std::sync::Mutex std::io::Result<()> + Send + Sync>>>, +> = std::sync::LazyLock::new(|| std::sync::Mutex::new(None)); + #[derive(Debug)] struct RemoteDeleteBreaker { threshold: usize, @@ -158,6 +165,11 @@ async fn record_remote_delete_failure(err: &std::io::Error, now: Instant) { } } +fn should_record_remote_delete_failure(err: &std::io::Error) -> bool { + let message = err.to_string(); + message != ERR_REMOTE_DELETE_BREAKER_OPEN && message != ERR_REMOTE_DELETE_LIMITER_CLOSED +} + #[derive(Default)] #[allow(dead_code)] struct ObjSweeper { @@ -277,31 +289,78 @@ impl ExpiryOp for Jentry { } pub async fn delete_object_from_remote_tier(obj_name: &str, rv_id: &str, tier_name: &str) -> Result<(), std::io::Error> { + let result = delete_object_from_remote_tier_raw(obj_name, rv_id, tier_name).await; + if let Err(err) = &result + && should_record_remote_delete_failure(err) + { + record_remote_delete_failure(err, Instant::now()).await; + } + result +} + +async fn delete_object_from_remote_tier_raw(obj_name: &str, rv_id: &str, tier_name: &str) -> Result<(), std::io::Error> { + #[cfg(test)] + if let Some(result) = run_remote_tier_delete_test_hook(obj_name, rv_id, tier_name) { + return result; + } + if remote_delete_breaker_is_open(Instant::now()).await { metrics::counter!(METRIC_DELETE_REMOTE_BREAKER_TOTAL).increment(1); - return Err(std::io::Error::other("remote tier delete breaker is open due to signer/header failures")); + return Err(std::io::Error::other(ERR_REMOTE_DELETE_BREAKER_OPEN)); } let _permit = REMOTE_DELETE_LIMITER .acquire() .await - .map_err(|_| std::io::Error::other("remote tier delete limiter is closed"))?; + .map_err(|_| std::io::Error::other(ERR_REMOTE_DELETE_LIMITER_CLOSED))?; let _inflight = RemoteDeleteInflightGuard::new(); let mut config_mgr = GLOBAL_TierConfigMgr.write().await; let w = match config_mgr.get_driver(tier_name).await { Ok(w) => w, - Err(e) => { - let err = std::io::Error::other(e); - record_remote_delete_failure(&err, Instant::now()).await; - return Err(err); - } + Err(e) => return Err(std::io::Error::other(e)), }; - let result = w.remove(obj_name, rv_id).await; - if let Err(err) = &result { - record_remote_delete_failure(err, Instant::now()).await; + w.remove(obj_name, rv_id).await +} + +#[cfg(test)] +fn run_remote_tier_delete_test_hook(obj_name: &str, rv_id: &str, tier_name: &str) -> Option> { + REMOTE_TIER_DELETE_TEST_HOOK + .lock() + .expect("remote tier delete test hook lock should not poison") + .as_ref() + .map(|hook| hook(obj_name, rv_id, tier_name)) +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum RemoteTierDeleteOutcome { + Deleted, + AlreadyRemoved, +} + +pub async fn delete_object_from_remote_tier_idempotent( + obj_name: &str, + rv_id: &str, + tier_name: &str, +) -> Result { + match delete_object_from_remote_tier_raw(obj_name, rv_id, tier_name).await { + Ok(()) => Ok(RemoteTierDeleteOutcome::Deleted), + Err(err) if is_remote_tier_not_found_error(&err) => Ok(RemoteTierDeleteOutcome::AlreadyRemoved), + Err(err) => { + if should_record_remote_delete_failure(&err) { + record_remote_delete_failure(&err, Instant::now()).await; + } + Err(err) + } } - result +} + +pub(crate) fn is_remote_tier_not_found_error(err: &std::io::Error) -> bool { + let message = err.to_string(); + message.contains("NoSuchKey") + || message.contains("NoSuchVersion") + || message.contains("ObjectNotFound") + || message.contains("VersionNotFound") } pub fn transitioned_delete_journal_entry( @@ -340,10 +399,35 @@ pub fn transitioned_force_delete_journal_entry(transitioned: &TransitionedObject mod test { use crate::client::signer_error::invalid_utf8_header_error; - use super::{RemoteDeleteBreaker, is_signer_header_error}; + use super::{ + ERR_REMOTE_DELETE_BREAKER_OPEN, ERR_REMOTE_DELETE_LIMITER_CLOSED, REMOTE_TIER_DELETE_TEST_HOOK, RemoteDeleteBreaker, + RemoteTierDeleteOutcome, delete_object_from_remote_tier_idempotent, is_remote_tier_not_found_error, + is_signer_header_error, should_record_remote_delete_failure, + }; use std::io::{Error, ErrorKind}; use std::time::{Duration, Instant}; + struct RemoteTierDeleteHookGuard; + + impl Drop for RemoteTierDeleteHookGuard { + fn drop(&mut self) { + let mut hook = REMOTE_TIER_DELETE_TEST_HOOK + .lock() + .expect("remote tier delete test hook lock should not poison"); + *hook = None; + } + } + + fn set_remote_tier_delete_test_hook( + hook_fn: impl Fn(&str, &str, &str) -> std::io::Result<()> + Send + Sync + 'static, + ) -> RemoteTierDeleteHookGuard { + let mut hook = REMOTE_TIER_DELETE_TEST_HOOK + .lock() + .expect("remote tier delete test hook lock should not poison"); + *hook = Some(Box::new(hook_fn)); + RemoteTierDeleteHookGuard + } + #[test] fn signer_header_error_detection_matches_utf8_failures() { let err = Error::new( @@ -365,6 +449,54 @@ mod test { assert!(is_signer_header_error(&err)); } + #[test] + fn remote_tier_not_found_errors_are_idempotent_success() { + assert!(is_remote_tier_not_found_error(&Error::other("NoSuchVersion"))); + assert!(is_remote_tier_not_found_error(&Error::other("NoSuchKey"))); + assert!(is_remote_tier_not_found_error(&Error::other("ObjectNotFound"))); + assert!(is_remote_tier_not_found_error(&Error::other("VersionNotFound"))); + assert!(!is_remote_tier_not_found_error(&Error::other("timeout"))); + assert!(!is_remote_tier_not_found_error(&Error::other("tier config not found"))); + assert!(!is_remote_tier_not_found_error(&Error::other("driver not found"))); + } + + #[test] + fn remote_tier_control_plane_short_circuit_errors_do_not_count_as_delete_failures() { + assert!(!should_record_remote_delete_failure(&Error::other(ERR_REMOTE_DELETE_BREAKER_OPEN))); + assert!(!should_record_remote_delete_failure(&Error::other(ERR_REMOTE_DELETE_LIMITER_CLOSED))); + assert!(should_record_remote_delete_failure(&Error::other("driver not found"))); + assert!(should_record_remote_delete_failure(&Error::other("NoSuchVersion"))); + } + + #[tokio::test] + #[serial_test::serial] + async fn idempotent_remote_delete_treats_hooked_nosuchversion_as_already_removed() { + let _hook = set_remote_tier_delete_test_hook(|obj_name, rv_id, tier_name| { + assert_eq!(obj_name, "remote/object"); + assert_eq!(rv_id, "remote-version"); + assert_eq!(tier_name, "WARM"); + Err(Error::other("NoSuchVersion")) + }); + + let outcome = delete_object_from_remote_tier_idempotent("remote/object", "remote-version", "WARM") + .await + .expect("remote not-found should be idempotent success"); + + assert_eq!(outcome, RemoteTierDeleteOutcome::AlreadyRemoved); + } + + #[tokio::test] + #[serial_test::serial] + async fn idempotent_remote_delete_preserves_driver_not_found_failures() { + let _hook = set_remote_tier_delete_test_hook(|_, _, _| Err(Error::other("driver not found"))); + + let err = delete_object_from_remote_tier_idempotent("remote/object", "remote-version", "WARM") + .await + .expect_err("driver lookup failure must not be idempotent success"); + + assert!(err.to_string().contains("driver not found")); + } + #[test] fn breaker_opens_at_threshold_and_recovers_after_window() { let mut breaker = RemoteDeleteBreaker::new(3, Duration::from_secs(30)); diff --git a/crates/ecstore/src/pools.rs b/crates/ecstore/src/pools.rs index afd3d3bff..d0ffcfb50 100644 --- a/crates/ecstore/src/pools.rs +++ b/crates/ecstore/src/pools.rs @@ -160,6 +160,16 @@ fn is_decommission_active(complete: bool, failed: bool, canceled: bool) -> bool !complete && !failed && !canceled } +fn validate_decommission_terminal_state(complete: bool, failed: bool, canceled: bool) -> Result<()> { + let terminal_count = [complete, failed, canceled].into_iter().filter(|terminal| *terminal).count(); + if terminal_count > 1 { + return Err(Error::other(format!( + "pool metadata load failed: invalid decommission terminal state complete={complete} failed={failed} canceled={canceled}" + ))); + } + Ok(()) +} + fn invalid_decommission_pool_index_error(pool_count: usize, idx: usize) -> Error { Error::other(format!("invalid decommission pool index {idx} for {pool_count} pools")) } @@ -532,30 +542,37 @@ struct PersistedPoolDecommissionInfo { pub bytes_failed: usize, } -impl From for PoolMeta { - fn from(value: PersistedPoolMeta) -> Self { - Self { +impl TryFrom for PoolMeta { + type Error = Error; + + fn try_from(value: PersistedPoolMeta) -> Result { + Ok(Self { version: value.version, - pools: value.pools.into_iter().map(Into::into).collect(), + pools: value.pools.into_iter().map(TryInto::try_into).collect::>>()?, dont_save: false, - } + }) } } -impl From for PoolStatus { - fn from(value: PersistedPoolStatus) -> Self { - Self { +impl TryFrom for PoolStatus { + type Error = Error; + + fn try_from(value: PersistedPoolStatus) -> Result { + Ok(Self { id: value.id, cmd_line: value.cmd_line, last_update: value.last_update, - decommission: value.decommission.map(Into::into), - } + decommission: value.decommission.map(TryInto::try_into).transpose()?, + }) } } -impl From for PoolDecommissionInfo { - fn from(value: PersistedPoolDecommissionInfo) -> Self { - Self { +impl TryFrom for PoolDecommissionInfo { + type Error = Error; + + fn try_from(value: PersistedPoolDecommissionInfo) -> Result { + validate_decommission_terminal_state(value.complete, value.failed, value.canceled)?; + Ok(Self { start_time: value.start_time, start_size: value.start_size, total_size: value.total_size, @@ -572,7 +589,7 @@ impl From for PoolDecommissionInfo { items_decommission_failed: value.items_decommission_failed, bytes_done: value.bytes_done, bytes_failed: value.bytes_failed, - } + }) } } @@ -622,7 +639,7 @@ impl From<&PoolDecommissionInfo> for PersistedPoolDecommissionInfo { impl PoolMeta { fn decode_pool_meta_payload(payload: &[u8]) -> Result { match rmp_serde::from_slice::(payload) { - Ok(meta) => Ok(meta.into()), + Ok(meta) => meta.try_into(), Err(persisted_err) => { let mut legacy: PoolMeta = rmp_serde::from_slice(payload).map_err(|legacy_err| { Error::other(format!( @@ -631,6 +648,11 @@ impl PoolMeta { })?; // Runtime-only flag must not be restored from on-disk payload. legacy.dont_save = false; + for pool in &legacy.pools { + if let Some(decommission) = &pool.decommission { + validate_decommission_terminal_state(decommission.complete, decommission.failed, decommission.canceled)?; + } + } Ok(legacy) } } @@ -2706,7 +2728,8 @@ mod tests { let mut deserializer = Deserializer::new(Cursor::new(&buf)); let restored: PoolMeta = PersistedPoolMeta::deserialize(&mut deserializer) .expect("pool meta should deserialize") - .into(); + .try_into() + .expect("pool meta should validate"); let restored_decommission = restored.pools[0] .decommission @@ -2782,6 +2805,63 @@ mod tests { assert!(decommission.prefix.is_empty()); assert!(decommission.object.is_empty()); } + + #[test] + fn pool_meta_decode_rejects_invalid_decommission_terminal_state() { + let start_time = OffsetDateTime::now_utc(); + let persisted_meta = PersistedPoolMeta { + version: POOL_META_VERSION, + pools: vec![PersistedPoolStatus { + id: 1, + cmd_line: "/data/pool1/disk{1...4}".to_string(), + last_update: start_time, + decommission: Some(PersistedPoolDecommissionInfo { + start_time: Some(start_time), + complete: true, + failed: true, + canceled: false, + ..Default::default() + }), + }], + }; + + let mut payload = Vec::new(); + persisted_meta + .serialize(&mut Serializer::new(&mut payload)) + .expect("persisted payload should serialize"); + + let err = PoolMeta::decode_pool_meta_payload(&payload).expect_err("invalid terminal state should fail decode"); + assert!(err.to_string().contains("invalid decommission terminal state")); + } + + #[test] + fn pool_meta_decode_rejects_invalid_legacy_decommission_terminal_state() { + let start_time = OffsetDateTime::now_utc(); + let legacy_meta = PoolMeta { + version: POOL_META_VERSION, + pools: vec![PoolStatus { + id: 1, + cmd_line: "/legacy/pool".to_string(), + last_update: start_time, + decommission: Some(PoolDecommissionInfo { + start_time: Some(start_time), + complete: true, + failed: false, + canceled: true, + ..Default::default() + }), + }], + dont_save: true, + }; + + let mut payload = Vec::new(); + legacy_meta + .serialize(&mut Serializer::new(&mut payload)) + .expect("legacy payload should serialize"); + + let err = PoolMeta::decode_pool_meta_payload(&payload).expect_err("invalid legacy terminal state should fail decode"); + assert!(err.to_string().contains("invalid decommission terminal state")); + } } // impl Fn(MetaCacheEntry) -> impl Future> diff --git a/crates/ecstore/src/rebalance.rs b/crates/ecstore/src/rebalance.rs index d52c63cae..b310b9509 100644 --- a/crates/ecstore/src/rebalance.rs +++ b/crates/ecstore/src/rebalance.rs @@ -51,7 +51,8 @@ use uuid::Uuid; const REBAL_META_FMT: u16 = 1; // Replace with actual format value const REBAL_META_VER: u16 = 1; // Replace with actual version value const REBAL_META_NAME: &str = "rebalance.bin"; -const REBALANCE_LISTING_MAX_ATTEMPTS: usize = 3; +const DEFAULT_REBALANCE_MAX_ATTEMPTS: usize = 3; +const REBALANCE_MAX_ATTEMPTS_ENV: &str = "RUSTFS_REBALANCE_MAX_ATTEMPTS"; const REBALANCE_LISTING_RETRY_BASE_DELAY: Duration = Duration::from_millis(250); const REBALANCE_MIGRATION_RETRY_BASE_DELAY: Duration = Duration::from_millis(250); const REBALANCE_MIGRATION_LOCK_RETRY_CAP: Duration = Duration::from_secs(10); @@ -2056,6 +2057,17 @@ fn should_retry_rebalance_listing(err: &Error, attempt: usize, max_attempts: usi attempt + 1 < max_attempts && is_transient_rebalance_error(err) } +fn parse_rebalance_max_attempts(value: Option<&str>) -> usize { + value + .and_then(|value| value.trim().parse::().ok()) + .filter(|attempts| *attempts > 0) + .unwrap_or(DEFAULT_REBALANCE_MAX_ATTEMPTS) +} + +fn rebalance_max_attempts() -> usize { + parse_rebalance_max_attempts(std::env::var(REBALANCE_MAX_ATTEMPTS_ENV).ok().as_deref()) +} + fn rebalance_listing_retry_delay(attempt: usize) -> Duration { let multiplier = u32::try_from(attempt.saturating_add(1)).unwrap_or(u32::MAX); REBALANCE_LISTING_RETRY_BASE_DELAY.saturating_mul(multiplier) @@ -2612,7 +2624,7 @@ impl ECStore { pool_index, version, version_id.clone(), - 3, + rebalance_max_attempts(), should_ignore_rebalance_data_usage_cache(bucket.as_str()), &mut transfer, ) @@ -2901,15 +2913,9 @@ impl ECStore { let entry_tasks = entry_tasks.clone(); let job = tokio::spawn(async move { - let list_result = run_rebalance_listing_with_retry( - set, - rx, - bucket.clone(), - rebalance_entry, - set_idx, - REBALANCE_LISTING_MAX_ATTEMPTS, - ) - .await; + let list_result = + run_rebalance_listing_with_retry(set, rx, bucket.clone(), rebalance_entry, set_idx, rebalance_max_attempts()) + .await; let entry_result = wait_rebalance_entry_tasks(set_idx, entry_tasks).await; let result = list_result.and(entry_result); if let Err(err) = &result { @@ -3169,11 +3175,11 @@ mod rebalance_unit_tests { ensure_rebalance_listing_disks_available, ensure_rebalance_not_decommissioning, ensure_valid_rebalance_pool_index, has_deferred_rebalance_error, is_rebalance_stopped_terminal_event, is_transient_rebalance_error, load_rebalance_bucket_configs, mark_rebalance_bucket_done, merge_rebalance_meta, migrate_entry_version, - migrate_entry_version_with_retry_wait, next_rebal_bucket_from_stat, rebalance_delete_marker_opts, - rebalance_listing_retry_delay, rebalance_meta_load_no_data_error, rebalance_meta_load_unknown_format_error, - rebalance_meta_load_unknown_version_error, rebalance_migration_retry_delay, record_rebalance_cleanup_warning_in_meta, - resolve_load_rebalance_stats_update_result, resolve_next_rebalance_bucket, resolve_rebalance_bucket_error, - resolve_rebalance_bucket_result, resolve_rebalance_entry_cleanup_delete_result, + migrate_entry_version_with_retry_wait, next_rebal_bucket_from_stat, parse_rebalance_max_attempts, + rebalance_delete_marker_opts, rebalance_listing_retry_delay, rebalance_meta_load_no_data_error, + rebalance_meta_load_unknown_format_error, rebalance_meta_load_unknown_version_error, rebalance_migration_retry_delay, + record_rebalance_cleanup_warning_in_meta, resolve_load_rebalance_stats_update_result, resolve_next_rebalance_bucket, + resolve_rebalance_bucket_error, resolve_rebalance_bucket_result, resolve_rebalance_entry_cleanup_delete_result, resolve_rebalance_file_info_versions_result, resolve_rebalance_meta_load_result, resolve_rebalance_meta_save_result, resolve_rebalance_migrate_result_error, resolve_rebalance_optional_bucket_config_result, resolve_rebalance_participants, resolve_rebalance_save_task_result, resolve_rebalance_stats_update_result, resolve_rebalance_terminal_error, @@ -4776,6 +4782,15 @@ mod rebalance_unit_tests { assert!(!should_retry_rebalance_listing(&Error::FileAccessDenied, 0, 3)); } + #[test] + fn test_parse_rebalance_max_attempts_uses_positive_override_or_default() { + assert_eq!(parse_rebalance_max_attempts(Some("5")), 5); + assert_eq!(parse_rebalance_max_attempts(Some(" 7 ")), 7); + assert_eq!(parse_rebalance_max_attempts(Some("0")), 3); + assert_eq!(parse_rebalance_max_attempts(Some("invalid")), 3); + assert_eq!(parse_rebalance_max_attempts(None), 3); + } + #[test] fn test_rebalance_listing_retry_delay_scales_by_attempt() { assert_eq!(rebalance_listing_retry_delay(0), Duration::from_millis(250)); diff --git a/crates/ecstore/src/store_api/types.rs b/crates/ecstore/src/store_api/types.rs index e4f64997f..610a4b400 100644 --- a/crates/ecstore/src/store_api/types.rs +++ b/crates/ecstore/src/store_api/types.rs @@ -1050,6 +1050,7 @@ pub struct WalkOptions { pub ask_disks: String, // dictates how many disks are being listed pub versions_sort: WalkVersionsSortOrder, // sort order for versions of the same object; default: Ascending order in ModTime pub limit: usize, // maximum number of items, 0 means no limit + pub include_free_versions: bool, // include persisted tier free-version cleanup records } #[derive(Clone, Default, PartialEq, Eq)] diff --git a/crates/ecstore/src/store_list_objects.rs b/crates/ecstore/src/store_list_objects.rs index a35e7495c..a21aca20b 100644 --- a/crates/ecstore/src/store_list_objects.rs +++ b/crates/ecstore/src/store_list_objects.rs @@ -1071,7 +1071,11 @@ impl ECStore { continue; } - let fvs = match entry.file_info_versions(&bucket_clone) { + let fvs = match if opts.include_free_versions { + entry.file_info_versions_with_free_versions(&bucket_clone) + } else { + entry.file_info_versions(&bucket_clone) + } { Ok(res) => res, Err(err) => { let item = ObjectInfoOrErr { @@ -1117,6 +1121,36 @@ impl ECStore { } } } + + if opts.include_free_versions { + for fi in fvs.free_versions.iter() { + if let Some(filter) = opts.filter { + if filter(fi) { + let item = ObjectInfoOrErr { + item: Some(ObjectInfo::from_file_info(fi, &bucket_clone, &fi.name, { + if let Some(v) = &vcf { v.versioned(&fi.name) } else { false } + })), + err: None, + }; + + if let Err(err) = result.send(item).await { + error!("walk result send err {:?}", err); + } + } + } else { + let item = ObjectInfoOrErr { + item: Some(ObjectInfo::from_file_info(fi, &bucket_clone, &fi.name, { + if let Some(v) = &vcf { v.versioned(&fi.name) } else { false } + })), + err: None, + }; + + if let Err(err) = result.send(item).await { + error!("walk result send err {:?}", err); + } + } + } + } } } .instrument(tracing::Span::current()), diff --git a/crates/filemeta/src/metacache.rs b/crates/filemeta/src/metacache.rs index 19a1e3c67..5b3637dcc 100644 --- a/crates/filemeta/src/metacache.rs +++ b/crates/filemeta/src/metacache.rs @@ -191,6 +191,27 @@ impl MetaCacheEntry { fm.into_file_info_versions(bucket, self.name.as_str(), false) } + pub fn file_info_versions_with_free_versions(&self, bucket: &str) -> Result { + if self.is_dir() { + return Ok(FileInfoVersions { + volume: bucket.to_string(), + name: self.name.clone(), + versions: vec![FileInfo { + volume: bucket.to_string(), + name: self.name.clone(), + ..Default::default() + }], + ..Default::default() + }); + } + + let mut fm = FileMeta::new(); + fm.unmarshal_msg(&self.metadata)?; + // `get_file_info_versions(..., false)` is the existing path that + // separates persisted tier free-version records into `free_versions`. + fm.get_file_info_versions(bucket, self.name.as_str(), false) + } + pub fn matches(&self, other: Option<&MetaCacheEntry>, strict: bool) -> (Option, bool) { if other.is_none() { return (None, false); @@ -899,7 +920,7 @@ impl Cache { mod tests { use super::*; use crate::test_data::create_real_xlmeta; - use crate::{FileMetaVersion, MetaDeleteMarker}; + use crate::{FileMetaVersion, MetaDeleteMarker, TRANSITION_COMPLETE}; use std::collections::HashMap; use std::io::Cursor; use std::sync::{ @@ -937,6 +958,196 @@ mod tests { assert_eq!(objs, nobjs); } + #[test] + fn file_info_versions_with_free_versions_includes_persisted_tier_cleanup_records() { + let version_id = Uuid::new_v4(); + let remote_version_id = Uuid::new_v4(); + let free_version_id = Uuid::new_v4(); + let mut fm = FileMeta::new(); + fm.add_version(FileInfo { + volume: "bucket".to_string(), + name: "object".to_string(), + version_id: Some(version_id), + transition_status: TRANSITION_COMPLETE.to_string(), + transitioned_objname: "remote/object".to_string(), + transition_version_id: Some(remote_version_id), + transition_tier: "WARM".to_string(), + mod_time: Some(OffsetDateTime::now_utc()), + ..Default::default() + }) + .expect("transitioned version should be added"); + + let mut delete_fi = FileInfo { + volume: "bucket".to_string(), + name: "object".to_string(), + version_id: Some(version_id), + mod_time: Some(OffsetDateTime::now_utc()), + ..Default::default() + }; + delete_fi.set_tier_free_version_id(&free_version_id.to_string()); + fm.delete_version(&delete_fi) + .expect("transitioned delete should create free-version metadata"); + + let entry = MetaCacheEntry { + name: "object".to_string(), + metadata: fm.marshal_msg().expect("metadata should marshal"), + ..Default::default() + }; + + let normal = entry.file_info_versions("bucket").expect("normal versions should parse"); + assert!(normal.free_versions.is_empty()); + + let with_free = entry + .file_info_versions_with_free_versions("bucket") + .expect("versions with free versions should parse"); + assert_eq!(with_free.free_versions.len(), 1); + assert!(with_free.free_versions[0].tier_free_version()); + assert_eq!(with_free.free_versions[0].transitioned_objname, "remote/object"); + assert_eq!(with_free.free_versions[0].transition_tier, "WARM"); + assert_eq!(with_free.free_versions[0].transition_version_id, Some(remote_version_id)); + } + + #[test] + fn transitioned_delete_persists_recoverable_free_version_after_metadata_roundtrip() { + let version_id = Uuid::new_v4(); + let remote_version_id = Uuid::new_v4(); + let free_version_id = Uuid::new_v4(); + let mut fm = FileMeta::new(); + fm.add_version(FileInfo { + volume: "bucket".to_string(), + name: "object".to_string(), + version_id: Some(version_id), + transition_status: TRANSITION_COMPLETE.to_string(), + transitioned_objname: "remote/object".to_string(), + transition_version_id: Some(remote_version_id), + transition_tier: "WARM".to_string(), + mod_time: Some(OffsetDateTime::now_utc()), + ..Default::default() + }) + .expect("transitioned version should be added"); + + let mut delete_fi = FileInfo { + volume: "bucket".to_string(), + name: "object".to_string(), + version_id: Some(version_id), + mod_time: Some(OffsetDateTime::now_utc()), + ..Default::default() + }; + delete_fi.set_tier_free_version_id(&free_version_id.to_string()); + + fm.delete_version(&delete_fi) + .expect("transitioned delete should persist free-version metadata"); + let encoded = fm.marshal_msg().expect("metadata should marshal"); + let mut decoded = FileMeta::new(); + decoded + .unmarshal_msg(&encoded) + .expect("metadata should survive process restart roundtrip"); + + let versions = decoded + .get_file_info_versions("bucket", "object", false) + .expect("versions with free versions should parse"); + + assert!(versions.versions.is_empty()); + assert_eq!(versions.free_versions.len(), 1); + let free_version = &versions.free_versions[0]; + assert_eq!(free_version.version_id, Some(free_version_id)); + assert!(free_version.tier_free_version()); + assert_eq!(free_version.transitioned_objname, "remote/object"); + assert_eq!(free_version.transition_version_id, Some(remote_version_id)); + assert_eq!(free_version.transition_tier, "WARM"); + } + + #[test] + fn skip_tier_free_version_does_not_persist_cleanup_record() { + let version_id = Uuid::new_v4(); + let mut fm = FileMeta::new(); + fm.add_version(FileInfo { + volume: "bucket".to_string(), + name: "object".to_string(), + version_id: Some(version_id), + transition_status: TRANSITION_COMPLETE.to_string(), + transitioned_objname: "remote/object".to_string(), + transition_version_id: Some(Uuid::new_v4()), + transition_tier: "WARM".to_string(), + mod_time: Some(OffsetDateTime::now_utc()), + ..Default::default() + }) + .expect("transitioned version should be added"); + + let mut delete_fi = FileInfo { + volume: "bucket".to_string(), + name: "object".to_string(), + version_id: Some(version_id), + mod_time: Some(OffsetDateTime::now_utc()), + ..Default::default() + }; + delete_fi.set_tier_free_version_id(&Uuid::new_v4().to_string()); + delete_fi.set_skip_tier_free_version(); + + fm.delete_version(&delete_fi) + .expect("transitioned delete with skip flag should remove the local version"); + let versions = fm + .get_file_info_versions("bucket", "object", false) + .expect("versions should parse after skipped cleanup"); + + assert!(versions.free_versions.is_empty()); + assert_eq!(versions.versions.len(), 1); + assert!(versions.versions[0].deleted); + assert!(!versions.versions[0].tier_free_version()); + } + + #[test] + fn worker_style_free_version_delete_removes_persisted_cleanup_record() { + let version_id = Uuid::new_v4(); + let remote_version_id = Uuid::new_v4(); + let free_version_id = Uuid::new_v4(); + let mut fm = FileMeta::new(); + fm.add_version(FileInfo { + volume: "bucket".to_string(), + name: "object".to_string(), + version_id: Some(version_id), + transition_status: TRANSITION_COMPLETE.to_string(), + transitioned_objname: "remote/object".to_string(), + transition_version_id: Some(remote_version_id), + transition_tier: "WARM".to_string(), + mod_time: Some(OffsetDateTime::now_utc()), + ..Default::default() + }) + .expect("transitioned version should be added"); + + let mut delete_fi = FileInfo { + volume: "bucket".to_string(), + name: "object".to_string(), + version_id: Some(version_id), + mod_time: Some(OffsetDateTime::now_utc()), + ..Default::default() + }; + delete_fi.set_tier_free_version_id(&free_version_id.to_string()); + fm.delete_version(&delete_fi) + .expect("transitioned delete should persist free-version metadata"); + + let mut free_delete_fi = FileInfo { + volume: "bucket".to_string(), + name: "object".to_string(), + version_id: Some(free_version_id), + deleted: true, + mod_time: Some(OffsetDateTime::now_utc()), + ..Default::default() + }; + free_delete_fi.set_tier_free_version(); + + fm.delete_version(&free_delete_fi) + .expect("worker-style free-version delete should remove the cleanup record"); + let versions = fm + .get_file_info_versions("bucket", "object", false) + .expect("versions should parse after free-version cleanup"); + + assert!(versions.free_versions.is_empty()); + assert_eq!(versions.versions.len(), 1); + assert!(versions.versions[0].deleted); + assert!(!versions.versions[0].tier_free_version()); + } + #[test] fn test_resolve_rebuilds_metadata_from_merged_versions() { let base_metadata = create_real_xlmeta().expect("base xl.meta"); diff --git a/scripts/test/tier_lifecycle_cleanup_stress.md b/scripts/test/tier_lifecycle_cleanup_stress.md new file mode 100644 index 000000000..8f329adf6 --- /dev/null +++ b/scripts/test/tier_lifecycle_cleanup_stress.md @@ -0,0 +1,45 @@ +# Tier Lifecycle Cleanup Stress Test + +## Goal + +Verify concurrent GET, ILM expiry, and tier free-version recovery do not produce `NoSuchVersion`, unbounded queue growth, or severe throughput regression. + +## Setup + +1. Start a RustFS cluster with a configured remote tier. +2. Upload at least 10,000 objects and transition them to the remote tier. +3. Configure lifecycle expiry for the transitioned prefix. +4. Enable lifecycle logs: + +```bash +export RUST_LOG="info,rustfs_ecstore::bucket::lifecycle=debug" +``` + +## Workload + +Run concurrent GET while triggering expiry: + +```bash +seq 1 64 | xargs -I{} -P64 sh -c 'while true; do aws s3api get-object --bucket "$BUCKET" --key "prefix/obj-$((RANDOM % 10000))" /tmp/out.$$ >/dev/null 2>>get-errors.log; done' +``` + +## Required Checks + +- `get-errors.log` contains no `NoSuchVersion`. +- Lifecycle queue pending count does not grow without bound. +- Missed free-version and tier journal counters may increase during injected failures, but later recovery drains persisted records. +- Recovery logs show cursor advancement across bucket and object markers. +- Duplicate recovery enqueue count stays bounded while remote tier delete is failing. +- Queue full or backpressure stops the current recovery pass instead of blocking normal lifecycle enqueue. +- Scan duration and scanned metadata entry count stay within the configured per-pass budget. +- p95 GET latency and lifecycle expiry throughput are recorded before and after the fix. + +## Fault Injection + +Repeat the workload while: + +- stopping lifecycle workers +- forcing remote tier delete to fail +- restarting the RustFS process after local delete but before remote cleanup + +After recovery, verify remote tier garbage does not remain for expired objects.