fix(ilm): drain tier-delete recovery pages (#7133)

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<Duration> {
+    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<ECStore>,
+    cancel_token: &CancellationToken,
+    marker: &mut Option<String>,
+    deadline: tokio::time::Instant,
+    stats: &mut TierDeleteJournalRecoveryPassStats,
+) -> Option<TierDeleteJournalRecoveryPageStep> {
+    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<ECStore>,
+    cancel_token: &CancellationToken,
+    marker: &mut Option<String>,
+    deadline: tokio::time::Instant,
+    stats: &mut TierDeleteJournalRecoveryPassStats,
+) -> Option<TierDeleteJournalRecoveryPageStep> {
+    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<ECStore>,
+    cancel_token: &CancellationToken,
+    manifest_marker: &mut Option<String>,
+    journal_marker: &mut Option<String>,
+    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<ECStore>,
+    cancel_token: &CancellationToken,
+    manifest_marker: &mut Option<String>,
+    journal_marker: &mut Option<String>,
+    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<ECStore>, 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<ECStore>, 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"
-                );
-            }
-        }
     }
 }
 

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<crate::store::ECStore>,
+        tier_name: &str,
+        backend_identity: [u8; 32],
+        count: usize,
+    ) -> Vec<Jentry> {
+        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::<Vec<_>>();
+        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)]