From ba0ba0d35e9a064598be18be23a27e751619313e Mon Sep 17 00:00:00 2001 From: cxymds Date: Tue, 16 Jun 2026 10:38:41 +0800 Subject: [PATCH] fix(lifecycle): persist remote tier delete journal tasks (#3493) fix: persist remote tier delete journal tasks --- .../bucket/lifecycle/bucket_lifecycle_ops.rs | 47 ++- crates/ecstore/src/bucket/lifecycle/mod.rs | 1 + .../bucket/lifecycle/tier_delete_journal.rs | 302 ++++++++++++++++++ .../src/bucket/lifecycle/tier_sweeper.rs | 10 +- rustfs/src/app/object_usecase.rs | 41 ++- 5 files changed, 389 insertions(+), 12 deletions(-) create mode 100644 crates/ecstore/src/bucket/lifecycle/tier_delete_journal.rs diff --git a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs index cff15e264..7083e5405 100644 --- a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs +++ b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs @@ -17,6 +17,7 @@ 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_delete_journal::{process_tier_delete_journal_entry, run_tier_delete_journal_recovery_loop}; 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_idempotent}; @@ -101,6 +102,7 @@ pub type ExpiryOpType = Box; static XXHASH_SEED: u64 = 0; static TIER_FREE_VERSION_RECOVERY_STARTED: OnceLock<()> = OnceLock::new(); +static TIER_DELETE_JOURNAL_RECOVERY_STARTED: OnceLock<()> = OnceLock::new(); pub const AMZ_OBJECT_TAGGING: &str = "X-Amz-Tagging"; pub const AMZ_TAG_COUNT: &str = "x-amz-tagging-count"; @@ -478,12 +480,17 @@ impl ExpiryState { if wrkr.is_none() { self.stats.increment_missed_tier_journal_tasks(); self.stats.record_scanner_expiry_state(); - return Ok(()); + return Err(std::io::Error::new( + std::io::ErrorKind::WouldBlock, + "lifecycle expiry worker unavailable for tier journal task", + )); } let wrkr = wrkr.expect("worker channel should exist after None check"); let queued = self.send_expiry_task(wrkr, Box::new(je.clone())).await; if !queued { self.stats.increment_missed_tier_journal_tasks(); + self.stats.record_scanner_expiry_state(); + return Err(std::io::Error::new(std::io::ErrorKind::BrokenPipe, "failed to enqueue tier journal task")); } self.stats.record_scanner_expiry_state(); Ok(()) @@ -657,7 +664,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_idempotent(&v.obj_name, &v.version_id, &v.tier_name).await { + if let Err(err) = process_tier_delete_journal_entry(api.clone(), v).await { debug!( event = EVENT_LIFECYCLE_WORKER_STATE, component = LOG_COMPONENT_ECSTORE, @@ -1351,7 +1358,8 @@ pub async fn init_background_expiry(api: Arc) { //let expiry_state = GLOBAL_ExpiryStSate.write().await; ExpiryState::resize_workers(workers, api.clone()).await; - spawn_tier_free_version_recovery_once(api); + spawn_tier_free_version_recovery_once(api.clone()); + spawn_tier_delete_journal_recovery_once(api); } fn spawn_tier_free_version_recovery_once(api: Arc) { @@ -1424,6 +1432,19 @@ fn spawn_tier_free_version_recovery_once(api: Arc) { }); } +fn spawn_tier_delete_journal_recovery_once(api: Arc) { + if TIER_DELETE_JOURNAL_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); + run_tier_delete_journal_recovery_loop(api, cancel_token).await; + }); +} + #[derive(Debug, Clone)] struct StaleMultipartUploadCandidate { path: String, @@ -2838,6 +2859,7 @@ mod tests { }; use crate::bucket::lifecycle::bucket_lifecycle_audit::LcEventSrc; use crate::bucket::lifecycle::core::ExpirationOptions; + use crate::bucket::lifecycle::tier_sweeper::Jentry; use crate::bucket::metadata::BUCKET_LIFECYCLE_CONFIG; use crate::bucket::metadata_sys; use crate::disk::RUSTFS_META_MULTIPART_BUCKET; @@ -2893,6 +2915,25 @@ mod tests { assert_eq!(expiry.queue_missed, 1); } + #[tokio::test] + async fn enqueue_tier_journal_entry_reports_error_without_worker_channel() { + let state = ExpiryState::new(); + let mut state = state.write().await; + let je = Jentry { + obj_name: "remote/object".to_string(), + version_id: "remote-version".to_string(), + tier_name: "WARM".to_string(), + }; + + let err = state + .enqueue_tier_journal_entry(&je) + .await + .expect_err("missing worker should be reported to caller"); + + assert_eq!(err.kind(), std::io::ErrorKind::WouldBlock); + assert_eq!(state.stats.missed_tier_journal_tasks(), 1); + } + #[tokio::test] async fn enqueue_free_version_reports_false_without_worker_channel() { let state = ExpiryState::new(); diff --git a/crates/ecstore/src/bucket/lifecycle/mod.rs b/crates/ecstore/src/bucket/lifecycle/mod.rs index 7a853c4b7..57430062d 100644 --- a/crates/ecstore/src/bucket/lifecycle/mod.rs +++ b/crates/ecstore/src/bucket/lifecycle/mod.rs @@ -18,6 +18,7 @@ pub mod core; pub mod evaluator; pub use self::core as lifecycle; pub mod rule; +pub mod tier_delete_journal; 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_delete_journal.rs b/crates/ecstore/src/bucket/lifecycle/tier_delete_journal.rs new file mode 100644 index 000000000..d8ef828cb --- /dev/null +++ b/crates/ecstore/src/bucket/lifecycle/tier_delete_journal.rs @@ -0,0 +1,302 @@ +// 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 serde::{Deserialize, Serialize}; +use sha2::{Digest, Sha256}; +use tokio_util::sync::CancellationToken; +use tracing::{debug, warn}; + +use crate::bucket::lifecycle::tier_sweeper::{Jentry, delete_object_from_remote_tier_idempotent}; +use crate::config::com::{delete_config, read_config, save_config}; +use crate::disk::RUSTFS_META_BUCKET; +use crate::error::{Error, Result}; +use crate::store::ECStore; +use crate::store_api::{ListOperations, ObjectIO, ObjectOperations}; + +const LOG_COMPONENT_ECSTORE: &str = "ecstore"; +const LOG_SUBSYSTEM_LIFECYCLE: &str = "lifecycle"; +const EVENT_LIFECYCLE_TIER_DELETE_JOURNAL: &str = "lifecycle_tier_delete_journal"; + +pub const DEFAULT_TIER_DELETE_JOURNAL_RECOVERY_LIMIT: usize = 1_000; +const TIER_DELETE_JOURNAL_VERSION: u8 = 1; +const TIER_DELETE_JOURNAL_PREFIX: &str = "ilm/tier-delete-journal/"; + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(deny_unknown_fields)] +struct PersistedTierDeleteJournalEntry { + version: u8, + obj_name: String, + version_id: String, + tier_name: String, +} + +impl PersistedTierDeleteJournalEntry { + fn from_jentry(je: &Jentry) -> Self { + Self { + version: TIER_DELETE_JOURNAL_VERSION, + obj_name: je.obj_name.clone(), + version_id: je.version_id.clone(), + tier_name: je.tier_name.clone(), + } + } + + fn into_jentry(self) -> Result { + if self.version != TIER_DELETE_JOURNAL_VERSION { + return Err(Error::other(format!("unsupported tier delete journal version {}", self.version))); + } + if self.obj_name.is_empty() || self.version_id.is_empty() || self.tier_name.is_empty() { + return Err(Error::other("tier delete journal entry is incomplete")); + } + Ok(Jentry { + obj_name: self.obj_name, + version_id: self.version_id, + tier_name: self.tier_name, + }) + } +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct TierDeleteJournalRecoveryStats { + pub scanned: usize, + pub deleted: usize, + pub failed: usize, + pub next_marker: Option, + pub truncated: bool, +} + +pub(crate) fn tier_delete_journal_object_name(je: &Jentry) -> String { + let mut hasher = Sha256::new(); + hasher.update(je.tier_name.as_bytes()); + hasher.update([0]); + hasher.update(je.obj_name.as_bytes()); + hasher.update([0]); + hasher.update(je.version_id.as_bytes()); + format!( + "{TIER_DELETE_JOURNAL_PREFIX}{}.json", + rustfs_utils::crypto::hex(hasher.finalize().as_slice()) + ) +} + +fn decode_tier_delete_journal_entry(data: &[u8]) -> Result { + let persisted: PersistedTierDeleteJournalEntry = + serde_json::from_slice(data).map_err(|err| Error::other(format!("decode tier delete journal failed: {err}")))?; + persisted.into_jentry() +} + +fn encode_tier_delete_journal_entry(je: &Jentry) -> Result> { + serde_json::to_vec(&PersistedTierDeleteJournalEntry::from_jentry(je)) + .map_err(|err| Error::other(format!("encode tier delete journal failed: {err}"))) +} + +pub async fn persist_tier_delete_journal_entry(api: Arc, je: &Jentry) -> std::io::Result<()> +where + S: ObjectIO, +{ + let data = encode_tier_delete_journal_entry(je).map_err(std::io::Error::other)?; + save_config(api, &tier_delete_journal_object_name(je), data) + .await + .map_err(std::io::Error::other) +} + +pub async fn remove_tier_delete_journal_entry(api: Arc, je: &Jentry) -> std::io::Result<()> +where + S: ObjectOperations, +{ + match delete_config(api, &tier_delete_journal_object_name(je)).await { + Ok(()) | Err(Error::ConfigNotFound) => Ok(()), + Err(err) => Err(std::io::Error::other(err)), + } +} + +pub async fn process_tier_delete_journal_entry(api: Arc, je: &Jentry) -> std::io::Result<()> { + delete_object_from_remote_tier_idempotent(&je.obj_name, &je.version_id, &je.tier_name).await?; + remove_tier_delete_journal_entry(api, je).await +} + +pub async fn recover_tier_delete_journal_entries( + api: Arc, + limit: usize, + marker: Option, +) -> Result { + if limit == 0 { + return Err(Error::other("tier delete journal recovery limit must be greater than zero")); + } + + let list = api + .clone() + .list_objects_v2( + RUSTFS_META_BUCKET, + TIER_DELETE_JOURNAL_PREFIX, + marker.clone(), + None, + i32::try_from(limit).unwrap_or(i32::MAX), + false, + None, + false, + ) + .await?; + + let mut stats = TierDeleteJournalRecoveryStats { + scanned: 0, + deleted: 0, + failed: 0, + next_marker: list.next_continuation_token, + truncated: list.is_truncated, + }; + + for object in list.objects { + stats.scanned += 1; + let data = match read_config(api.clone(), &object.name).await { + Ok(data) => data, + Err(Error::ConfigNotFound) => continue, + Err(err) => { + stats.failed += 1; + warn!( + event = EVENT_LIFECYCLE_TIER_DELETE_JOURNAL, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_LIFECYCLE, + journal_object = %object.name, + error = ?err, + "Failed to read tier delete journal entry" + ); + continue; + } + }; + + let je = match decode_tier_delete_journal_entry(&data) { + Ok(je) => je, + Err(err) => { + stats.failed += 1; + warn!( + event = EVENT_LIFECYCLE_TIER_DELETE_JOURNAL, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_LIFECYCLE, + journal_object = %object.name, + error = ?err, + "Failed to decode tier delete journal entry" + ); + continue; + } + }; + + match process_tier_delete_journal_entry(api.clone(), &je).await { + Ok(()) => stats.deleted += 1, + Err(err) => { + stats.failed += 1; + debug!( + event = EVENT_LIFECYCLE_TIER_DELETE_JOURNAL, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_LIFECYCLE, + journal_object = %object.name, + remote_object = %je.obj_name, + remote_version_id = %je.version_id, + tier = %je.tier_name, + error = ?err, + "Tier delete journal recovery will retry later" + ); + } + } + } + + Ok(stats) +} + +pub async fn run_tier_delete_journal_recovery_loop(api: Arc, cancel_token: CancellationToken) { + let mut interval = tokio::time::interval(std::time::Duration::from_secs(60)); + let mut marker: Option = None; + + loop { + tokio::select! { + _ = cancel_token.cancelled() => return, + _ = interval.tick() => {} + } + + match recover_tier_delete_journal_entries(api.clone(), DEFAULT_TIER_DELETE_JOURNAL_RECOVERY_LIMIT, marker.clone()).await { + Ok(stats) => { + marker = stats.next_marker; + debug!( + event = EVENT_LIFECYCLE_TIER_DELETE_JOURNAL, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_LIFECYCLE, + scanned = stats.scanned, + deleted = stats.deleted, + failed = stats.failed, + truncated = stats.truncated, + next_marker = ?marker, + "Recovered tier delete journal tasks" + ); + } + Err(err) => { + warn!( + event = EVENT_LIFECYCLE_TIER_DELETE_JOURNAL, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_LIFECYCLE, + next_marker = ?marker, + error = ?err, + "Failed to recover tier delete journal tasks" + ); + } + } + } +} + +#[cfg(test)] +mod tests { + use super::{decode_tier_delete_journal_entry, encode_tier_delete_journal_entry, tier_delete_journal_object_name}; + use crate::bucket::lifecycle::tier_sweeper::Jentry; + + fn journal_entry() -> Jentry { + Jentry { + obj_name: "remote/object".to_string(), + version_id: "remote-version".to_string(), + tier_name: "WARM".to_string(), + } + } + + #[test] + fn tier_delete_journal_roundtrips_entry() { + let je = journal_entry(); + + let encoded = encode_tier_delete_journal_entry(&je).expect("journal entry should encode"); + let decoded = decode_tier_delete_journal_entry(&encoded).expect("journal entry should decode"); + + assert_eq!(decoded.obj_name, je.obj_name); + assert_eq!(decoded.version_id, je.version_id); + assert_eq!(decoded.tier_name, je.tier_name); + } + + #[test] + fn tier_delete_journal_path_is_stable_and_sanitized() { + let je = journal_entry(); + + let first = tier_delete_journal_object_name(&je); + let second = tier_delete_journal_object_name(&je); + + assert_eq!(first, second); + assert!(first.starts_with("ilm/tier-delete-journal/")); + assert!(first.ends_with(".json")); + assert!(!first.contains("remote/object")); + } + + #[test] + fn tier_delete_journal_rejects_incomplete_entry() { + let payload = br#"{"version":1,"obj_name":"","version_id":"v1","tier_name":"WARM"}"#; + + let err = decode_tier_delete_journal_entry(payload).expect_err("incomplete journal entry should be rejected"); + + assert!(err.to_string().contains("incomplete")); + } +} diff --git a/crates/ecstore/src/bucket/lifecycle/tier_sweeper.rs b/crates/ecstore/src/bucket/lifecycle/tier_sweeper.rs index ed3d318a7..5aea99eae 100644 --- a/crates/ecstore/src/bucket/lifecycle/tier_sweeper.rs +++ b/crates/ecstore/src/bucket/lifecycle/tier_sweeper.rs @@ -20,15 +20,17 @@ use crate::bucket::lifecycle::bucket_lifecycle_ops::{ExpiryOp, GLOBAL_ExpiryState, TransitionedObject}; use crate::bucket::lifecycle::lifecycle::{self, ObjectOpts}; +use crate::bucket::lifecycle::tier_delete_journal::persist_tier_delete_journal_entry; use crate::client::signer_error::error_chain_contains_signer_header_marker; use crate::global::GLOBAL_TierConfigMgr; +use crate::store::ECStore; use rustfs_utils::get_env_usize; use sha2::{Digest, Sha256}; use std::any::Any; use std::collections::VecDeque; use std::io::Write; -use std::sync::LazyLock; use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::{Arc, LazyLock}; use std::time::{Duration, Instant}; use tokio::sync::{Mutex, Semaphore}; use tracing::warn; @@ -249,10 +251,14 @@ impl ObjSweeper { None } - pub async fn sweep(&self) { + pub async fn sweep(&self, api: Arc) { let Some(je) = self.should_remove_remote_object() else { return; }; + if persist_tier_delete_journal_entry(api, &je).await.is_err() { + GLOBAL_ExpiryState.write().await.increment_missed_tier_journal_tasks(); + return; + } let hash = je.op_hash(); // Grab the sender under a short read lock, then release the lock so we // don't hold it across the async send. diff --git a/rustfs/src/app/object_usecase.rs b/rustfs/src/app/object_usecase.rs index 262c934dc..807cf70ea 100644 --- a/rustfs/src/app/object_usecase.rs +++ b/rustfs/src/app/object_usecase.rs @@ -248,9 +248,15 @@ enum GetObjectTimeoutStage { BeforeRead, } -async fn enqueue_transitioned_delete_cleanup(bucket: &str, object: &str, opts: &ObjectOptions, existing: Option<&ObjectInfo>) { +async fn enqueue_transitioned_delete_cleanup( + store: Arc, + bucket: &str, + object: &str, + opts: &ObjectOptions, + existing: Option<&ObjectInfo>, +) -> std::io::Result<()> { let Some(existing) = existing else { - return; + return Ok(()); }; let _activity_guard = DeleteTailActivityGuard::new(DeleteTailStage::Cleanup); @@ -266,9 +272,11 @@ async fn enqueue_transitioned_delete_cleanup(bucket: &str, object: &str, opts: & ) }; let Some(je) = je else { - return; + return Ok(()); }; + rustfs_ecstore::bucket::lifecycle::tier_delete_journal::persist_tier_delete_journal_entry(store, &je).await?; + let mut expiry_state = rustfs_ecstore::bucket::lifecycle::bucket_lifecycle_ops::GLOBAL_ExpiryState .write() .await; @@ -280,9 +288,10 @@ async fn enqueue_transitioned_delete_cleanup(bucket: &str, object: &str, opts: & remote_version_id = %existing.transitioned_object.version_id, tier = %existing.transitioned_object.tier, error = ?err, - "failed to enqueue transitioned object cleanup" + "transitioned object cleanup journal persisted but was not queued" ); } + Ok(()) } pin_project! { @@ -3274,7 +3283,8 @@ impl DefaultObjectUsecase { dobjs[i].replication_state = Some(object_to_delete[i].replication_state()); } delete_results[didx].delete_object = Some(dobjs[i].clone()); - enqueue_transitioned_delete_cleanup( + if let Err(err) = enqueue_transitioned_delete_cleanup( + store.clone(), &bucket, &object_to_delete[i].object_name, &ObjectOptions { @@ -3285,7 +3295,15 @@ impl DefaultObjectUsecase { }, existing_object_infos[i].as_ref(), ) - .await; + .await + { + warn!( + bucket = %bucket, + object = %object_to_delete[i].object_name, + error = ?err, + "failed to persist transitioned object cleanup journal" + ); + } let size = object_sizes[i].max(0) as u64; rustfs_ecstore::data_usage::record_bucket_object_delete_memory( &bucket, @@ -3508,7 +3526,16 @@ impl DefaultObjectUsecase { } }; - enqueue_transitioned_delete_cleanup(&bucket, &key, &opts, existing_object_info.as_ref()).await; + if let Err(err) = + enqueue_transitioned_delete_cleanup(store.clone(), &bucket, &key, &opts, existing_object_info.as_ref()).await + { + warn!( + bucket = %bucket, + object = %key, + error = ?err, + "failed to persist transitioned object cleanup journal" + ); + } // Fast in-memory update for immediate quota and admin usage consistency rustfs_ecstore::data_usage::record_bucket_object_delete_memory(