From ff3c5a4989efde7734a6239e0a11311920d32a39 Mon Sep 17 00:00:00 2001 From: cxymds Date: Fri, 4 Sep 2026 18:46:19 +0800 Subject: [PATCH] fix(ilm): drain tier-delete recovery pages (#7133) --- .../bucket/lifecycle/tier_delete_journal.rs | 320 ++++++++++++++---- crates/ecstore/src/store/init.rs | 225 +++++++++++- 2 files changed, 467 insertions(+), 78 deletions(-) diff --git a/crates/ecstore/src/bucket/lifecycle/tier_delete_journal.rs b/crates/ecstore/src/bucket/lifecycle/tier_delete_journal.rs index ed502e570..3a6a7e451 100644 --- a/crates/ecstore/src/bucket/lifecycle/tier_delete_journal.rs +++ b/crates/ecstore/src/bucket/lifecycle/tier_delete_journal.rs @@ -61,12 +61,13 @@ pub(crate) const LOG_COMPONENT_ECSTORE: &str = "ecstore"; pub(crate) const LOG_SUBSYSTEM_LIFECYCLE: &str = "lifecycle"; pub(crate) const EVENT_LIFECYCLE_TIER_DELETE_JOURNAL: &str = "lifecycle_tier_delete_journal"; -// Keep one background pass small enough that a slow remote tier cannot hold -// the shared recovery worker for minutes. Subsequent passes resume from the -// returned marker, so this bounds latency without reducing eventual coverage. +// Keep one page small enough that slow remote tiers remain bounded by per-entry +// deadlines and worker concurrency. A production pass may consume multiple +// successful pages while its wall-clock budget remains. pub const DEFAULT_TIER_DELETE_JOURNAL_RECOVERY_LIMIT: usize = 8; const TIER_DELETE_JOURNAL_RECOVERY_INTERVAL: Duration = Duration::from_secs(60); const TIER_DELETE_JOURNAL_RECOVERY_TIMEOUT: Duration = Duration::from_secs(300); +const TIER_DELETE_JOURNAL_RECOVERY_PASS_BUDGET: Duration = Duration::from_secs(240); const TIER_DELETE_REMOTE_DEADLINE: Duration = Duration::from_secs(30); const TIER_DELETE_JOURNAL_ENTRY_RECOVERY_TIMEOUT: Duration = Duration::from_secs(90); const TIER_DELETE_JOURNAL_RECOVERY_CONCURRENCY: usize = 4; @@ -1085,7 +1086,7 @@ fn validate_version_state( Ok(()) } -#[derive(Debug, Clone, PartialEq, Eq)] +#[derive(Debug, Clone, Default, PartialEq, Eq)] pub struct TierDeleteJournalRecoveryStats { pub scanned: usize, pub deleted: usize, @@ -1094,7 +1095,7 @@ pub struct TierDeleteJournalRecoveryStats { pub truncated: bool, } -#[derive(Debug, Clone, PartialEq, Eq)] +#[derive(Debug, Clone, Default, PartialEq, Eq)] pub struct TierDeleteDispatchManifestRecoveryStats { pub scanned: usize, pub advanced: usize, @@ -1105,6 +1106,18 @@ pub struct TierDeleteDispatchManifestRecoveryStats { pub truncated: bool, } +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub(crate) struct TierDeleteJournalRecoveryPassStats { + pub manifest_pages: usize, + pub journal_pages: usize, + pub manifest_errors: usize, + pub journal_errors: usize, + pub deadline_exhausted: bool, + pub canceled: bool, + pub manifests: TierDeleteDispatchManifestRecoveryStats, + pub journals: TierDeleteJournalRecoveryStats, +} + pub(crate) fn tier_delete_journal_object_name(je: &Jentry) -> String { let digest = tier_delete_journal_identity_digest(je); if let Some(dispatch) = &je.dispatch { @@ -5660,6 +5673,201 @@ pub async fn recover_tier_delete_journal_entries( Ok(stats) } +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +struct TierDeleteJournalRecoveryPageStep { + stop_queue: bool, + made_progress: bool, +} + +fn remaining_recovery_pass_budget( + deadline: tokio::time::Instant, + stats: &mut TierDeleteJournalRecoveryPassStats, +) -> Option { + match deadline.checked_duration_since(tokio::time::Instant::now()) { + Some(remaining) if !remaining.is_zero() => Some(remaining), + _ => { + stats.deadline_exhausted = true; + None + } + } +} + +fn accumulate_dispatch_manifest_recovery_stats( + total: &mut TierDeleteDispatchManifestRecoveryStats, + page: TierDeleteDispatchManifestRecoveryStats, +) { + total.scanned += page.scanned; + total.advanced += page.advanced; + total.deleted += page.deleted; + total.retained += page.retained; + total.failed += page.failed; + total.next_marker = page.next_marker; + total.truncated = page.truncated; +} + +fn accumulate_tier_delete_journal_recovery_stats( + total: &mut TierDeleteJournalRecoveryStats, + page: TierDeleteJournalRecoveryStats, +) { + total.scanned += page.scanned; + total.deleted += page.deleted; + total.failed += page.failed; + total.next_marker = page.next_marker; + total.truncated = page.truncated; +} + +async fn recover_tier_delete_dispatch_manifest_pass_page( + api: Arc, + cancel_token: &CancellationToken, + marker: &mut Option, + deadline: tokio::time::Instant, + stats: &mut TierDeleteJournalRecoveryPassStats, +) -> Option { + if cancel_token.is_cancelled() { + stats.canceled = true; + return None; + } + let remaining = remaining_recovery_pass_budget(deadline, stats)?; + let page_marker = marker.clone(); + let recovery = recover_tier_delete_dispatch_manifests(api, DEFAULT_TIER_DELETE_JOURNAL_RECOVERY_LIMIT, page_marker.clone()); + let page = + await_tier_delete_journal_recovery(cancel_token, remaining.min(TIER_DELETE_JOURNAL_RECOVERY_TIMEOUT), recovery).await; + let Some(page) = page else { + stats.canceled = true; + return None; + }; + match page { + Ok(page) => { + stats.manifest_pages += 1; + let made_progress = page.advanced > 0 || page.deleted > 0; + let stop_queue = !page.truncated || !made_progress; + *marker = page.next_marker.clone(); + accumulate_dispatch_manifest_recovery_stats(&mut stats.manifests, page); + Some(TierDeleteJournalRecoveryPageStep { + stop_queue, + made_progress, + }) + } + Err(err) => { + stats.manifest_errors += 1; + warn!( + event = EVENT_LIFECYCLE_TIER_DELETE_JOURNAL, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_LIFECYCLE, + next_marker = ?page_marker, + error = ?err, + "Failed to recover tier delete dispatch manifest page" + ); + Some(TierDeleteJournalRecoveryPageStep { + stop_queue: true, + made_progress: false, + }) + } + } +} + +async fn recover_tier_delete_journal_pass_page( + api: Arc, + cancel_token: &CancellationToken, + marker: &mut Option, + deadline: tokio::time::Instant, + stats: &mut TierDeleteJournalRecoveryPassStats, +) -> Option { + if cancel_token.is_cancelled() { + stats.canceled = true; + return None; + } + let remaining = remaining_recovery_pass_budget(deadline, stats)?; + let page_marker = marker.clone(); + let recovery = recover_tier_delete_journal_entries(api, DEFAULT_TIER_DELETE_JOURNAL_RECOVERY_LIMIT, page_marker.clone()); + let page = + await_tier_delete_journal_recovery(cancel_token, remaining.min(TIER_DELETE_JOURNAL_RECOVERY_TIMEOUT), recovery).await; + let Some(page) = page else { + stats.canceled = true; + return None; + }; + match page { + Ok(page) => { + stats.journal_pages += 1; + let made_progress = page.deleted > 0; + let stop_queue = !page.truncated || !made_progress; + *marker = page.next_marker.clone(); + accumulate_tier_delete_journal_recovery_stats(&mut stats.journals, page); + Some(TierDeleteJournalRecoveryPageStep { + stop_queue, + made_progress, + }) + } + Err(err) => { + stats.journal_errors += 1; + warn!( + event = EVENT_LIFECYCLE_TIER_DELETE_JOURNAL, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_LIFECYCLE, + next_marker = ?page_marker, + error = ?err, + "Failed to recover tier delete journal page" + ); + Some(TierDeleteJournalRecoveryPageStep { + stop_queue: true, + made_progress: false, + }) + } + } +} + +async fn recover_tier_delete_journal_pass( + api: Arc, + cancel_token: &CancellationToken, + manifest_marker: &mut Option, + journal_marker: &mut Option, + budget: Duration, +) -> TierDeleteJournalRecoveryPassStats { + let deadline = tokio::time::Instant::now() + budget; + let mut stats = TierDeleteJournalRecoveryPassStats::default(); + let mut manifest_stopped = false; + let mut journal_stopped = false; + + while !manifest_stopped || !journal_stopped { + let mut made_progress = false; + if !manifest_stopped { + let Some(step) = + recover_tier_delete_dispatch_manifest_pass_page(api.clone(), cancel_token, manifest_marker, deadline, &mut stats) + .await + else { + return stats; + }; + manifest_stopped = step.stop_queue; + made_progress |= step.made_progress; + } + if !journal_stopped { + let Some(step) = + recover_tier_delete_journal_pass_page(api.clone(), cancel_token, journal_marker, deadline, &mut stats).await + else { + return stats; + }; + journal_stopped = step.stop_queue; + made_progress |= step.made_progress; + } + if !made_progress { + break; + } + } + + stats +} + +#[cfg(all(test, feature = "test-util"))] +pub(crate) async fn recover_test_tier_delete_journal_pass_with_budget( + api: Arc, + cancel_token: &CancellationToken, + manifest_marker: &mut Option, + journal_marker: &mut Option, + budget: Duration, +) -> TierDeleteJournalRecoveryPassStats { + recover_tier_delete_journal_pass(api, cancel_token, manifest_marker, journal_marker, budget).await +} + pub async fn run_tier_delete_journal_recovery_loop(api: Arc, cancel_token: CancellationToken) { let mut interval = tokio::time::interval(TIER_DELETE_JOURNAL_RECOVERY_INTERVAL); interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); @@ -5674,78 +5882,40 @@ pub async fn run_tier_delete_journal_recovery_loop(api: Arc, cancel_tok _ = api.ctx.wait_for_tier_delete_journal_recovery() => {}, } - let manifest_recovery = recover_tier_delete_dispatch_manifests( + let stats = recover_tier_delete_journal_pass( api.clone(), - DEFAULT_TIER_DELETE_JOURNAL_RECOVERY_LIMIT, - manifest_marker.clone(), + &cancel_token, + &mut manifest_marker, + &mut marker, + TIER_DELETE_JOURNAL_RECOVERY_PASS_BUDGET, + ) + .await; + if stats.canceled { + return; + } + debug!( + event = EVENT_LIFECYCLE_TIER_DELETE_JOURNAL, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_LIFECYCLE, + manifest_pages = stats.manifest_pages, + manifest_scanned = stats.manifests.scanned, + manifest_advanced = stats.manifests.advanced, + manifest_deleted = stats.manifests.deleted, + manifest_retained = stats.manifests.retained, + manifest_failed = stats.manifests.failed, + manifest_errors = stats.manifest_errors, + manifest_truncated = stats.manifests.truncated, + manifest_next_marker = ?manifest_marker, + journal_pages = stats.journal_pages, + journal_scanned = stats.journals.scanned, + journal_deleted = stats.journals.deleted, + journal_failed = stats.journals.failed, + journal_errors = stats.journal_errors, + journal_truncated = stats.journals.truncated, + journal_next_marker = ?marker, + deadline_exhausted = stats.deadline_exhausted, + "Recovered tier delete journal pass" ); - let Some(manifest_result) = - await_tier_delete_journal_recovery(&cancel_token, TIER_DELETE_JOURNAL_RECOVERY_TIMEOUT, manifest_recovery).await - else { - return; - }; - match manifest_result { - Ok(stats) => { - manifest_marker = stats.next_marker; - debug!( - event = EVENT_LIFECYCLE_TIER_DELETE_JOURNAL, - component = LOG_COMPONENT_ECSTORE, - subsystem = LOG_SUBSYSTEM_LIFECYCLE, - scanned = stats.scanned, - advanced = stats.advanced, - deleted = stats.deleted, - retained = stats.retained, - failed = stats.failed, - truncated = stats.truncated, - next_marker = ?manifest_marker, - "Reconciled tier delete dispatch manifests" - ); - } - Err(err) => { - warn!( - event = EVENT_LIFECYCLE_TIER_DELETE_JOURNAL, - component = LOG_COMPONENT_ECSTORE, - subsystem = LOG_SUBSYSTEM_LIFECYCLE, - next_marker = ?manifest_marker, - error = ?err, - "Failed to recover tier delete dispatch manifests" - ); - } - } - - let recovery = - recover_tier_delete_journal_entries(api.clone(), DEFAULT_TIER_DELETE_JOURNAL_RECOVERY_LIMIT, marker.clone()); - let Some(result) = - await_tier_delete_journal_recovery(&cancel_token, TIER_DELETE_JOURNAL_RECOVERY_TIMEOUT, recovery).await - else { - return; - }; - match result { - 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" - ); - } - } } } diff --git a/crates/ecstore/src/store/init.rs b/crates/ecstore/src/store/init.rs index 00a74906e..a89ef05f1 100644 --- a/crates/ecstore/src/store/init.rs +++ b/crates/ecstore/src/store/init.rs @@ -821,9 +821,9 @@ mod tests { TierDeleteDispatchRollbackTestHook, complete_tier_delete_dispatch, encode_tier_delete_journal_entry, install_test_tier_delete_dispatch_fixture, persist_tier_delete_journal_entry, prepare_tier_delete_dispatch, recover_test_tier_delete_dispatch_manifest, recover_test_tier_delete_dispatch_manifest_with_page_budget, - recover_tier_delete_dispatch_manifests, recover_tier_delete_journal_entries, - test_tier_delete_dispatch_manifest_checkpoint, test_tier_delete_dispatch_manifest_state, - tier_delete_dispatch_manifest_operation_lock_held_for_test, + recover_test_tier_delete_journal_pass_with_budget, recover_tier_delete_dispatch_manifests, + recover_tier_delete_journal_entries, test_tier_delete_dispatch_manifest_checkpoint, + test_tier_delete_dispatch_manifest_state, tier_delete_dispatch_manifest_operation_lock_held_for_test, tier_delete_dispatch_manifest_recovery_count_for_test, tier_delete_dispatch_manifest_recovery_inflight_for_test, tier_delete_journal_object_name, }, @@ -8697,6 +8697,35 @@ mod tests { .len() } + #[cfg(feature = "test-util")] + async fn install_committed_tier_delete_journals( + store: Arc, + tier_name: &str, + backend_identity: [u8; 32], + count: usize, + ) -> Vec { + let mut entries = Vec::with_capacity(count); + for index in 0..count { + let entry = Jentry { + persisted_version: 0, + obj_name: format!("remote/pass-drain-{index:06}.bin"), + version_id: uuid::Uuid::new_v4().to_string(), + tier_name: tier_name.to_string(), + backend_identity: Some(backend_identity), + version_id_exact: true, + version_state: rustfs_filemeta::TransitionVersionState::Exact, + state: TierDeleteJournalState::Committed, + source: None, + dispatch: None, + }; + persist_tier_delete_journal_entry(store.clone(), &entry) + .await + .expect("committed tier-delete journal fixture should persist"); + entries.push(entry); + } + entries + } + #[cfg(feature = "test-util")] fn synthetic_v6_dispatch_entry( bucket: &str, @@ -8793,6 +8822,196 @@ mod tests { .expect("same-disk restart recovery should converge without retained dispatch state"); } + #[cfg(feature = "test-util")] + #[tokio::test] + #[serial_test::serial(storage_class_env)] + async fn tier_delete_recovery_pass_drains_multiple_fast_manifest_pages() { + const MANIFEST_COUNT: usize = 10; + + let temp_dir = tempfile::tempdir().expect("create fast manifest pass recovery store dir"); + let (ctx, store, _shutdown) = + without_storage_class_env(build_isolated_test_store(temp_dir.path(), "tier-delete-fast-manifest-pass", &[4])).await; + crate::bucket::metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await; + let bucket = "tier-delete-fast-manifest-pass-bucket"; + store + .make_bucket(bucket, &MakeBucketOptions::default()) + .await + .expect("fast manifest pass bucket should be created"); + let incarnation = store + .bucket_incarnation_id(bucket) + .await + .expect("fast manifest pass bucket incarnation should resolve"); + let tier_name = "FAST-MANIFEST-PASS"; + let backend = register_mock_tier(&ctx.tier_config_mgr(), tier_name).await; + let backend_identity = TierConfigMgr::acquire_operation_lease(&ctx.tier_config_mgr(), tier_name) + .await + .expect("fast manifest pass tier lease should resolve") + .backend_identity(); + for index in 0..MANIFEST_COUNT { + install_aborting_dispatch_fixture( + store.clone(), + bucket, + incarnation, + &format!("manifest-page-{index:06}/"), + tier_name, + backend_identity, + 1, + ) + .await; + } + assert_eq!(tier_delete_dispatch_manifest_count(store.clone()).await, MANIFEST_COUNT); + assert_eq!(tier_delete_journal_count(store.clone()).await, MANIFEST_COUNT); + + let cancel = CancellationToken::new(); + let mut manifest_marker = None; + let mut journal_marker = None; + let stats = recover_test_tier_delete_journal_pass_with_budget( + store.clone(), + &cancel, + &mut manifest_marker, + &mut journal_marker, + Duration::from_secs(30), + ) + .await; + + assert!(!stats.canceled); + assert!(!stats.deadline_exhausted); + assert!( + stats.manifest_pages > 1, + "one production pass must cross the default eight-manifest page limit" + ); + assert_eq!(stats.manifests.scanned, MANIFEST_COUNT); + assert_eq!(stats.manifests.deleted, MANIFEST_COUNT); + assert_eq!(stats.manifests.failed, 0); + assert_eq!(manifest_marker, None); + assert_eq!(tier_delete_dispatch_manifest_count(store.clone()).await, 0); + assert_eq!(tier_delete_journal_count(store).await, 0); + assert_eq!(backend.remove_count().await, 0, "rollback recovery must not call the remote tier"); + } + + #[cfg(feature = "test-util")] + #[tokio::test] + #[serial_test::serial(storage_class_env)] + async fn tier_delete_recovery_pass_drains_multiple_fast_journal_pages() { + const JOURNAL_COUNT: usize = 25; + + let temp_dir = tempfile::tempdir().expect("create fast pass recovery store dir"); + let (ctx, store, _shutdown) = + without_storage_class_env(build_isolated_test_store(temp_dir.path(), "tier-delete-fast-pass", &[4])).await; + crate::bucket::metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await; + let tier_name = "FAST-PASS-RECOVERY"; + let backend = register_mock_tier(&ctx.tier_config_mgr(), tier_name).await; + let backend_identity = TierConfigMgr::acquire_operation_lease(&ctx.tier_config_mgr(), tier_name) + .await + .expect("fast pass tier lease should resolve") + .backend_identity(); + let entries = install_committed_tier_delete_journals(store.clone(), tier_name, backend_identity, JOURNAL_COUNT).await; + assert_eq!(tier_delete_journal_count(store.clone()).await, JOURNAL_COUNT); + + let cancel = CancellationToken::new(); + let mut manifest_marker = None; + let mut journal_marker = None; + let stats = recover_test_tier_delete_journal_pass_with_budget( + store.clone(), + &cancel, + &mut manifest_marker, + &mut journal_marker, + Duration::from_secs(30), + ) + .await; + + assert!(!stats.canceled); + assert!(!stats.deadline_exhausted); + assert!( + stats.journal_pages > 1, + "one production pass must cross the default eight-record page limit" + ); + assert_eq!(stats.journals.scanned, JOURNAL_COUNT); + assert_eq!(stats.journals.deleted, JOURNAL_COUNT); + assert_eq!(stats.journals.failed, 0); + assert_eq!(journal_marker, None); + assert_eq!(tier_delete_journal_count(store).await, 0); + assert_eq!(backend.exact_remove_count(), JOURNAL_COUNT); + let mut removed = backend.remove_versions().await; + removed.sort(); + let mut expected = entries + .into_iter() + .map(|entry| (entry.obj_name, entry.version_id)) + .collect::>(); + expected.sort(); + assert_eq!(removed, expected); + } + + #[cfg(feature = "test-util")] + #[tokio::test] + #[serial_test::serial(storage_class_env)] + async fn tier_delete_recovery_pass_rotates_failed_journal_pages_without_busy_loop() { + const JOURNAL_COUNT: usize = 9; + + let temp_dir = tempfile::tempdir().expect("create failed pass recovery store dir"); + let (ctx, store, _shutdown) = + without_storage_class_env(build_isolated_test_store(temp_dir.path(), "tier-delete-failed-pass", &[4])).await; + crate::bucket::metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await; + let tier_name = "FAILED-PASS-RECOVERY"; + let backend = register_mock_tier(&ctx.tier_config_mgr(), tier_name).await; + let backend_identity = TierConfigMgr::acquire_operation_lease(&ctx.tier_config_mgr(), tier_name) + .await + .expect("failed pass tier lease should resolve") + .backend_identity(); + install_committed_tier_delete_journals(store.clone(), tier_name, backend_identity, JOURNAL_COUNT).await; + backend.set_remove_failure(true); + + let cancel = CancellationToken::new(); + let mut manifest_marker = None; + let mut journal_marker = None; + let first = recover_test_tier_delete_journal_pass_with_budget( + store.clone(), + &cancel, + &mut manifest_marker, + &mut journal_marker, + Duration::from_secs(30), + ) + .await; + assert_eq!(first.journal_pages, 1, "a no-progress failed page must stop the pass"); + assert_eq!(first.journals.scanned, 8); + assert_eq!(first.journals.deleted, 0); + assert_eq!(first.journals.failed, 8); + assert!(journal_marker.is_some(), "a truncated failed page must retain its continuation marker"); + assert_eq!(tier_delete_journal_count(store.clone()).await, JOURNAL_COUNT); + + let second = recover_test_tier_delete_journal_pass_with_budget( + store.clone(), + &cancel, + &mut manifest_marker, + &mut journal_marker, + Duration::from_secs(30), + ) + .await; + assert_eq!(second.journal_pages, 1); + assert_eq!(second.journals.scanned, 1); + assert_eq!(second.journals.deleted, 0); + assert_eq!(second.journals.failed, 1); + assert_eq!( + journal_marker, None, + "end-of-list rotation must revisit the failed prefix on a later pass" + ); + assert_eq!(tier_delete_journal_count(store.clone()).await, JOURNAL_COUNT); + + backend.set_remove_failure(false); + let third = recover_test_tier_delete_journal_pass_with_budget( + store.clone(), + &cancel, + &mut manifest_marker, + &mut journal_marker, + Duration::from_secs(30), + ) + .await; + assert_eq!(third.journals.scanned, JOURNAL_COUNT); + assert_eq!(third.journals.deleted, JOURNAL_COUNT); + assert_eq!(third.journals.failed, 0); + assert_eq!(tier_delete_journal_count(store).await, 0); + } + #[cfg(feature = "test-util")] #[tokio::test] #[serial_test::serial(storage_class_env)]