fix(lifecycle): persist remote tier delete journal tasks (#3493)

fix: persist remote tier delete journal tasks

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<dyn ExpiryOp + Send + Sync + 'static>;
 
 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::<Jentry>() {
                         let v = v.as_any().downcast_ref::<Jentry>().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<ECStore>) {
 
     //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<ECStore>) {
@@ -1424,6 +1432,19 @@ fn spawn_tier_free_version_recovery_once(api: Arc<ECStore>) {
     });
 }
 
+fn spawn_tier_delete_journal_recovery_once(api: Arc<ECStore>) {
+    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();

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;

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<Jentry> {
+        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<String>,
+    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<Jentry> {
+    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<Vec<u8>> {
+    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<S>(api: Arc<S>, 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<S>(api: Arc<S>, 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<ECStore>, 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<ECStore>,
+    limit: usize,
+    marker: Option<String>,
+) -> Result<TierDeleteJournalRecoveryStats> {
+    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<ECStore>, cancel_token: CancellationToken) {
+    let mut interval = tokio::time::interval(std::time::Duration::from_secs(60));
+    let mut marker: Option<String> = 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"));
+    }
+}

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<ECStore>) {
         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.

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<ECStore>,
+    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(