fix(ilm): report manual transition tier failures (#5238)

Co-authored-by: heihutu <heihutu@gmail.com>

crates/e2e_test/src/reliant/tiering.rs

@@ -74,16 +74,22 @@ const MANUAL_ASYNC_STATUS_BUCKET: &str = "ilm7-manual-async-status";
 const MANUAL_CONTINUATION_BUCKET: &str = "ilm7-manual-continuation";
 const MANUAL_ASYNC_LIMIT_BUCKET: &str = "ilm7-manual-async-limit";
 const MANUAL_ASYNC_CONFLICT_BUCKET: &str = "ilm7-manual-async-conflict";
+const MANUAL_TIER_FAILURE_BUCKET: &str = "ilm7-manual-tier-failure";
+const MANUAL_WORKER_FAILURE_BUCKET: &str = "ilm7-manual-worker-failure";
 const MANUAL_QUEUE_PRESSURE_PREFIX: &str = "manual-queue-pressure/";
 const MANUAL_CONTINUATION_PREFIX: &str = "manual-continuation/";
 const MANUAL_ASYNC_LIMIT_PREFIX: &str = "manual-async-limit/";
 const MANUAL_ASYNC_CONFLICT_PREFIX: &str = "manual-async-conflict/";
 const MANUAL_ASYNC_CONFLICT_NESTED_PREFIX: &str = "manual-async-conflict/nested/";
+const MANUAL_TIER_FAILURE_PREFIX: &str = "manual-tier-failure/";
+const MANUAL_WORKER_FAILURE_PREFIX: &str = "manual-worker-failure/";
 const OBJECT_KEY: &str = "tier/鲁A12345/report.bin";
 const MANUAL_DUE_KEY: &str = "manual-due/report.bin";
 const MANUAL_DRY_RUN_KEY: &str = "manual-dry-run/report.bin";
 const MANUAL_NOT_DUE_KEY: &str = "manual-not-due/report.bin";
 const MANUAL_ASYNC_STATUS_KEY: &str = "manual-async-status/report.bin";
+const MANUAL_TIER_FAILURE_KEY: &str = "manual-tier-failure/report.bin";
+const MANUAL_WORKER_FAILURE_KEY: &str = "manual-worker-failure/report.bin";
 const CONTENT_TYPE: &str = "application/x-ilm7";
 const USER_META_KEY: &str = "ilm7-origin";
 const USER_META_VAL: &str = "hermetic-transition";
@@ -177,6 +183,15 @@ async fn add_rustfs_tier(hot: &RustFSTestEnvironment, cold: &RustFSTestEnvironme
     Ok(())
 }
 
+async fn remove_rustfs_tier_force(hot: &RustFSTestEnvironment) -> TestResult {
+    let path = format!("/rustfs/admin/v3/tier/{TIER_NAME}?force=true");
+    let (status, resp) = signed_admin_request(&hot.url, Method::DELETE, &path, None, &hot.access_key, &hot.secret_key).await?;
+    if !status.is_success() {
+        return Err(format!("RemoveTier(RustFS) failed: status={status}, body={resp}").into());
+    }
+    Ok(())
+}
+
 /// A current-version `Transition Days=0` rule scoped to the object's prefix.
 fn transition_rule() -> Result<LifecycleRule, Box<dyn std::error::Error + Send + Sync>> {
     transition_rule_for("ilm7-transition", "tier/", 0)
@@ -355,6 +370,14 @@ struct ManualTransitionRunReport {
     skipped_queue_full: u64,
     skipped_queue_closed: u64,
     skipped_queue_timeout: u64,
+    #[serde(default)]
+    transition_completed: u64,
+    #[serde(default)]
+    transition_failed: u64,
+    #[serde(default)]
+    tier_failure: u64,
+    #[serde(default)]
+    cancelled: bool,
     truncated_by_limit: bool,
     truncated_by_duration: bool,
     continuation_token: Option<String>,
@@ -718,6 +741,8 @@ async fn test_manual_transition_run_black_box_semantics() -> TestResult {
     assert_eq!(due.report.skipped_delete_marker, 0);
     assert_eq!(due.report.skipped_directory, 0);
     assert_eq!(due.report.skipped_replication, 0);
+    assert_eq!(due.report.tier_failure, 0);
+    assert!(!due.report.cancelled);
     assert!(!due.report.truncated_by_limit);
     assert!(!due.report.truncated_by_duration);
     wait_for_transition(&hot_client, MANUAL_DUE_BUCKET, MANUAL_DUE_KEY, StdDuration::from_secs(90)).await?;
@@ -745,6 +770,8 @@ async fn test_manual_transition_run_black_box_semantics() -> TestResult {
     assert_eq!(dry.report.dry_run_eligible, 1, "dry-run report: {:#?}", dry.report);
     assert_eq!(dry.report.enqueued, 0, "dry-run report: {:#?}", dry.report);
     assert_eq!(dry.report.skipped_not_transition, 1, "dry-run report: {:#?}", dry.report);
+    assert_eq!(dry.report.tier_failure, 0);
+    assert!(!dry.report.cancelled);
     assert!(!dry.report.truncated_by_duration);
     assert_eq!(
         cold_tier_object_count(&cold_client).await?,
@@ -767,6 +794,8 @@ async fn test_manual_transition_run_black_box_semantics() -> TestResult {
     assert_eq!(not_due.report.eligible, 0, "not-due report: {:#?}", not_due.report);
     assert_eq!(not_due.report.enqueued, 0, "not-due report: {:#?}", not_due.report);
     assert_eq!(not_due.report.skipped_not_transition, 1, "not-due report: {:#?}", not_due.report);
+    assert_eq!(not_due.report.tier_failure, 0);
+    assert!(!not_due.report.cancelled);
     assert_eq!(not_due.report.skipped_queue_full, 0);
     assert_eq!(not_due.report.skipped_queue_closed, 0);
     assert_eq!(not_due.report.skipped_queue_timeout, 0);
@@ -811,6 +840,8 @@ async fn test_manual_transition_async_job_status_polling() -> TestResult {
     assert!(accepted.report.dry_run);
     assert_eq!(accepted.report.scanned, 0);
     assert_eq!(accepted.report.eligible, 0);
+    assert_eq!(accepted.report.transition_completed, 0);
+    assert_eq!(accepted.report.transition_failed, 0);
     let job_id = accepted.job_id.as_deref().ok_or("async response must include job_id")?;
     let status_endpoint = accepted
         .status_endpoint
@@ -840,6 +871,10 @@ async fn test_manual_transition_async_job_status_polling() -> TestResult {
     assert_eq!(terminal.report.skipped_queue_full, 0);
     assert_eq!(terminal.report.skipped_queue_closed, 0);
     assert_eq!(terminal.report.skipped_queue_timeout, 0);
+    assert_eq!(terminal.report.transition_completed, 0);
+    assert_eq!(terminal.report.transition_failed, 0);
+    assert_eq!(terminal.report.tier_failure, 0);
+    assert!(!terminal.report.cancelled);
     assert!(!terminal.report.truncated_by_limit);
     assert!(!terminal.report.truncated_by_duration);
 
@@ -886,6 +921,8 @@ async fn test_manual_transition_async_limit_reports_terminal_partial() -> TestRe
     assert_eq!(accepted.report.bucket, MANUAL_ASYNC_LIMIT_BUCKET);
     assert_eq!(accepted.report.prefix, MANUAL_ASYNC_LIMIT_PREFIX);
     assert_eq!(accepted.report.scanned, 0);
+    assert_eq!(accepted.report.transition_completed, 0);
+    assert_eq!(accepted.report.transition_failed, 0);
     let job_id = accepted.job_id.as_deref().ok_or("async response must include job_id")?;
     let status_endpoint = accepted
         .status_endpoint
@@ -909,6 +946,10 @@ async fn test_manual_transition_async_limit_reports_terminal_partial() -> TestRe
     assert_eq!(terminal.report.skipped_queue_full, 0);
     assert_eq!(terminal.report.skipped_queue_closed, 0);
     assert_eq!(terminal.report.skipped_queue_timeout, 0);
+    assert_eq!(terminal.report.transition_completed, 0);
+    assert_eq!(terminal.report.transition_failed, 0);
+    assert_eq!(terminal.report.tier_failure, 0);
+    assert!(!terminal.report.cancelled);
     assert!(terminal.report.truncated_by_limit);
     assert!(!terminal.report.truncated_by_duration);
 
@@ -920,12 +961,20 @@ async fn test_manual_transition_async_limit_reports_terminal_partial() -> TestRe
     assert_eq!(after_cancel.report.tier, terminal.report.tier);
     assert_eq!(after_cancel.report.scanned, terminal.report.scanned);
     assert_eq!(after_cancel.report.skipped_not_transition, terminal.report.skipped_not_transition);
+    assert_eq!(after_cancel.report.transition_completed, terminal.report.transition_completed);
+    assert_eq!(after_cancel.report.transition_failed, terminal.report.transition_failed);
+    assert_eq!(after_cancel.report.tier_failure, terminal.report.tier_failure);
+    assert_eq!(after_cancel.report.cancelled, terminal.report.cancelled);
     assert_eq!(after_cancel.report.truncated_by_limit, terminal.report.truncated_by_limit);
 
     let second_cancel = manual_transition_job_cancel(&hot, status_endpoint).await?;
     assert_eq!(second_cancel.status, "partial");
     assert!(!second_cancel.cancel_requested);
     assert_eq!(second_cancel.report.scanned, terminal.report.scanned);
+    assert_eq!(second_cancel.report.transition_completed, terminal.report.transition_completed);
+    assert_eq!(second_cancel.report.transition_failed, terminal.report.transition_failed);
+    assert_eq!(second_cancel.report.tier_failure, terminal.report.tier_failure);
+    assert_eq!(second_cancel.report.cancelled, terminal.report.cancelled);
     assert_eq!(second_cancel.report.truncated_by_limit, terminal.report.truncated_by_limit);
     assert_eq!(cold_tier_object_count(&cold_client).await?, before_remote_count);
     Ok(())
@@ -1011,11 +1060,191 @@ async fn test_manual_transition_async_overlapping_scope_conflict_reports_active_
         50,
         "terminal conflict winner response: {terminal:#?}"
     );
+    assert_eq!(
+        terminal.report.transition_completed, terminal.report.enqueued,
+        "terminal conflict winner must wait for all queued transitions: {terminal:#?}"
+    );
+    assert_eq!(terminal.report.transition_failed, 0, "terminal conflict winner response: {terminal:#?}");
+    assert_eq!(terminal.report.tier_failure, 0, "terminal conflict winner response: {terminal:#?}");
     assert!(cold_tier_object_count(&cold_client).await? <= 50);
 
     Ok(())
 }
 
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn test_manual_transition_async_tier_failure_reports_terminal_partial() -> TestResult {
+    let mut cold = RustFSTestEnvironment::new().await?;
+    cold.access_key = "manualtierfailurecoldadmin".to_string();
+    cold.secret_key = "manualtierfailurecoldsecret".to_string();
+    cold.start_rustfs_server_without_cleanup(vec![]).await?;
+    let cold_client = cold.create_s3_client();
+    cold_client.create_bucket().bucket(TIER_BUCKET).send().await?;
+
+    let mut hot = RustFSTestEnvironment::new().await?;
+    hot.start_rustfs_server_with_env(vec![], &[("RUSTFS_SCANNER_ENABLED", "false"), ("RUSTFS_SCANNER_CYCLE", "3600")])
+        .await?;
+    let hot_client = hot.create_s3_client();
+    add_rustfs_tier(&hot, &cold).await?;
+
+    hot_client.create_bucket().bucket(MANUAL_TIER_FAILURE_BUCKET).send().await?;
+    let due_mtime = OffsetDateTime::now_utc() - time::Duration::hours(25);
+    put_backdated_single_part_object(
+        &hot_client,
+        MANUAL_TIER_FAILURE_BUCKET,
+        MANUAL_TIER_FAILURE_KEY,
+        b"manual tier failure object",
+        due_mtime,
+    )
+    .await?;
+    put_lifecycle_transition_rule(
+        &hot_client,
+        MANUAL_TIER_FAILURE_BUCKET,
+        "manual-tier-failure",
+        MANUAL_TIER_FAILURE_PREFIX,
+        0,
+    )
+    .await?;
+    remove_rustfs_tier_force(&hot).await?;
+
+    let before_remote_count = cold_tier_object_count(&cold_client).await?;
+    let accepted = manual_transition_async_run(&hot, MANUAL_TIER_FAILURE_BUCKET, MANUAL_TIER_FAILURE_PREFIX, false, 10).await?;
+    assert_eq!(accepted.state, "accepted");
+    assert_eq!(accepted.mode, "durable_job");
+    assert_eq!(accepted.report.transition_completed, 0);
+    assert_eq!(accepted.report.transition_failed, 0);
+    let job_id = accepted.job_id.as_deref().ok_or("async response must include job_id")?;
+    let status_endpoint = accepted
+        .status_endpoint
+        .as_deref()
+        .ok_or("async response must include status_endpoint")?;
+
+    let terminal = wait_for_manual_transition_job_terminal(&hot, status_endpoint, StdDuration::from_secs(30)).await?;
+    assert_eq!(terminal.job_id, job_id);
+    assert_eq!(terminal.status, "partial", "terminal tier failure job response: {terminal:#?}");
+    assert!(!terminal.cancel_requested);
+    assert_eq!(terminal.failure_reason, None);
+    assert_eq!(terminal.report.bucket, MANUAL_TIER_FAILURE_BUCKET);
+    assert_eq!(terminal.report.prefix, MANUAL_TIER_FAILURE_PREFIX);
+    assert_eq!(terminal.report.tier.as_deref(), Some(TIER_NAME));
+    assert!(!terminal.report.dry_run);
+    assert!(terminal.report.lifecycle_config_found);
+    assert_eq!(terminal.report.scanned, 1, "terminal tier failure job response: {terminal:#?}");
+    assert_eq!(terminal.report.eligible, 0, "terminal tier failure job response: {terminal:#?}");
+    assert_eq!(terminal.report.enqueued, 0, "terminal tier failure job response: {terminal:#?}");
+    assert_eq!(
+        terminal.report.transition_completed, 0,
+        "terminal tier failure job response: {terminal:#?}"
+    );
+    assert_eq!(terminal.report.transition_failed, 0, "terminal tier failure job response: {terminal:#?}");
+    assert_eq!(terminal.report.tier_failure, 1, "terminal tier failure job response: {terminal:#?}");
+    assert_eq!(terminal.report.skipped_queue_full, 0);
+    assert_eq!(terminal.report.skipped_queue_closed, 0);
+    assert_eq!(terminal.report.skipped_queue_timeout, 0);
+    assert!(!terminal.report.cancelled);
+    assert!(!terminal.report.truncated_by_limit);
+    assert!(!terminal.report.truncated_by_duration);
+    assert_eq!(
+        cold_tier_object_count(&cold_client).await?,
+        before_remote_count,
+        "tier failure must not create a remote object"
+    );
+    assert_remains_not_transitioned(
+        &hot_client,
+        MANUAL_TIER_FAILURE_BUCKET,
+        MANUAL_TIER_FAILURE_KEY,
+        StdDuration::from_secs(2),
+    )
+    .await?;
+
+    Ok(())
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn test_manual_transition_async_worker_failure_reports_terminal_partial() -> TestResult {
+    let mut cold = RustFSTestEnvironment::new().await?;
+    cold.access_key = "manualworkerfailurecoldadmin".to_string();
+    cold.secret_key = "manualworkerfailurecoldsecret".to_string();
+    cold.start_rustfs_server_without_cleanup(vec![]).await?;
+    let cold_client = cold.create_s3_client();
+    cold_client.create_bucket().bucket(TIER_BUCKET).send().await?;
+
+    let mut hot = RustFSTestEnvironment::new().await?;
+    hot.start_rustfs_server_with_env(vec![], &[("RUSTFS_SCANNER_ENABLED", "false"), ("RUSTFS_SCANNER_CYCLE", "3600")])
+        .await?;
+    let hot_client = hot.create_s3_client();
+    add_rustfs_tier(&hot, &cold).await?;
+    cold.stop_server();
+
+    hot_client.create_bucket().bucket(MANUAL_WORKER_FAILURE_BUCKET).send().await?;
+    let due_mtime = OffsetDateTime::now_utc() - time::Duration::hours(25);
+    put_backdated_single_part_object(
+        &hot_client,
+        MANUAL_WORKER_FAILURE_BUCKET,
+        MANUAL_WORKER_FAILURE_KEY,
+        b"manual worker failure object",
+        due_mtime,
+    )
+    .await?;
+    put_lifecycle_transition_rule(
+        &hot_client,
+        MANUAL_WORKER_FAILURE_BUCKET,
+        "manual-worker-failure",
+        MANUAL_WORKER_FAILURE_PREFIX,
+        0,
+    )
+    .await?;
+
+    let accepted =
+        manual_transition_async_run(&hot, MANUAL_WORKER_FAILURE_BUCKET, MANUAL_WORKER_FAILURE_PREFIX, false, 10).await?;
+    assert_eq!(accepted.state, "accepted");
+    assert_eq!(accepted.mode, "durable_job");
+    assert_eq!(accepted.report.transition_completed, 0);
+    assert_eq!(accepted.report.transition_failed, 0);
+    let job_id = accepted.job_id.as_deref().ok_or("async response must include job_id")?;
+    let status_endpoint = accepted
+        .status_endpoint
+        .as_deref()
+        .ok_or("async response must include status_endpoint")?;
+
+    let terminal = wait_for_manual_transition_job_terminal(&hot, status_endpoint, StdDuration::from_secs(30)).await?;
+    assert_eq!(terminal.job_id, job_id);
+    assert_eq!(terminal.status, "partial", "terminal worker failure job response: {terminal:#?}");
+    assert!(!terminal.cancel_requested);
+    assert_eq!(terminal.failure_reason, None);
+    assert_eq!(terminal.report.bucket, MANUAL_WORKER_FAILURE_BUCKET);
+    assert_eq!(terminal.report.prefix, MANUAL_WORKER_FAILURE_PREFIX);
+    assert_eq!(terminal.report.tier.as_deref(), Some(TIER_NAME));
+    assert!(!terminal.report.dry_run);
+    assert!(terminal.report.lifecycle_config_found);
+    assert_eq!(terminal.report.scanned, 1, "terminal worker failure job response: {terminal:#?}");
+    assert_eq!(terminal.report.eligible, 1, "terminal worker failure job response: {terminal:#?}");
+    assert_eq!(terminal.report.enqueued, 1, "terminal worker failure job response: {terminal:#?}");
+    assert_eq!(
+        terminal.report.transition_completed, 0,
+        "terminal worker failure job response: {terminal:#?}"
+    );
+    assert_eq!(
+        terminal.report.transition_failed, 1,
+        "terminal worker failure job response: {terminal:#?}"
+    );
+    assert_eq!(terminal.report.tier_failure, 1, "terminal worker failure job response: {terminal:#?}");
+    assert_eq!(terminal.report.skipped_queue_full, 0);
+    assert_eq!(terminal.report.skipped_queue_closed, 0);
+    assert_eq!(terminal.report.skipped_queue_timeout, 0);
+    assert!(!terminal.report.cancelled);
+    assert!(!terminal.report.truncated_by_limit);
+    assert!(!terminal.report.truncated_by_duration);
+    assert_remains_not_transitioned(
+        &hot_client,
+        MANUAL_WORKER_FAILURE_BUCKET,
+        MANUAL_WORKER_FAILURE_KEY,
+        StdDuration::from_secs(2),
+    )
+    .await?;
+
+    Ok(())
+}
+
 #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
 async fn test_manual_transition_run_contract_no_status_cancel_fields() -> TestResult {
     let mut cold = RustFSTestEnvironment::new().await?;
@@ -1092,6 +1321,8 @@ async fn test_manual_transition_run_continuation_token_resumes_without_raw_marke
     assert_eq!(first.report.scanned, 1, "first continuation page: {first:#?}");
     assert_eq!(first.report.eligible, 1, "first continuation page: {first:#?}");
     assert_eq!(first.report.dry_run_eligible, 1, "first continuation page: {first:#?}");
+    assert_eq!(first.report.tier_failure, 0);
+    assert!(!first.report.cancelled);
     assert!(first.report.truncated_by_limit);
     let continuation = first
         .report
@@ -1119,6 +1350,8 @@ async fn test_manual_transition_run_continuation_token_resumes_without_raw_marke
     assert_eq!(second.report.scanned, 1, "second continuation page: {second:#?}");
     assert_eq!(second.report.eligible, 1, "second continuation page: {second:#?}");
     assert_eq!(second.report.dry_run_eligible, 1, "second continuation page: {second:#?}");
+    assert_eq!(second.report.tier_failure, 0);
+    assert!(!second.report.cancelled);
     assert!(!second.report.truncated_by_limit);
     assert!(second.report.continuation_token.is_none());
 
@@ -1173,6 +1406,8 @@ async fn test_manual_transition_run_queue_pressure_partial() -> TestResult {
         "expected queue-pressure path to skip at least one object: {:#?}",
         response.report
     );
+    assert_eq!(response.report.tier_failure, 0);
+    assert!(!response.report.cancelled);
     assert!(!response.report.truncated_by_duration);
     assert!(response.report.enqueued < 20, "partial run should not enqueue all items in this setup");
 

crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs

@@ -23,10 +23,11 @@ use crate::bucket::lifecycle::lifecycle::{
 };
 use crate::bucket::lifecycle::manual_transition_job::{
     MANUAL_TRANSITION_JOB_RECORD_PREFIX, ManualTransitionJobRecord, ManualTransitionScopeAdmission,
-    ManualTransitionScopeAdmissionClaim, claim_manual_transition_scope_admission,
+    ManualTransitionScopeAdmissionClaim, ManualTransitionWorkerResult, claim_manual_transition_scope_admission,
     delete_manual_transition_scope_admission_if_current, load_manual_transition_job_record,
     load_manual_transition_job_record_with_etag, manual_transition_job_id_from_record_object_name,
-    manual_transition_job_lease_expired, persist_manual_transition_job_progress, save_manual_transition_job_record_if_current,
+    manual_transition_job_lease_expired, persist_manual_transition_job_progress, record_manual_transition_worker_result,
+    renew_manual_transition_job_lease, save_manual_transition_job_record_if_current,
 };
 use crate::bucket::lifecycle::replication_sink;
 use crate::bucket::lifecycle::replication_sink::{
@@ -954,6 +955,7 @@ struct TransitionTask {
     obj_info: ObjectInfo,
     src: LcEventSrc,
     event: lifecycle::Event,
+    manual_job_id: Option<Uuid>,
 }
 
 impl ExpiryOp for TransitionTask {
@@ -1259,6 +1261,7 @@ impl TransitionState {
         oi: &ObjectInfo,
         event: &lifecycle::Event,
         src: &LcEventSrc,
+        manual_job_id: Option<Uuid>,
     ) -> TransitionEnqueueOutcome {
         if is_immediate_transition_source(src) && should_force_immediate_transition_enqueue_timeout() {
             self.handle_immediate_enqueue_failure(oi, src, ImmediateEnqueueFailure::ForcedTimeout);
@@ -1280,6 +1283,7 @@ impl TransitionState {
             obj_info: oi.clone(),
             src: src.clone(),
             event: event.clone(),
+            manual_job_id,
         };
         if is_immediate_transition_source(src) {
             let outcome = match self.transition_tx.try_send(Some(task)) {
@@ -1361,7 +1365,7 @@ impl TransitionState {
     }
 
     pub async fn queue_transition_task(self: &Arc<Self>, oi: &ObjectInfo, event: &lifecycle::Event, src: &LcEventSrc) -> bool {
-        self.queue_transition_task_outcome(oi, event, src).await.is_handled()
+        self.queue_transition_task_outcome(oi, event, src, None).await.is_handled()
     }
 
     pub async fn init(api: Arc<ECStore>) {
@@ -1454,24 +1458,43 @@ impl TransitionState {
                             ..Default::default()
                         };
 
-                            if let Err(err) = transition_object(api.clone(), &task.obj_info, LcAuditEvent::new(task.event.clone(), task.src.clone())).await {
-                                global_metrics().record_scanner_transition_failed(1);
-                                if !is_err_version_not_found(&err) && !is_err_object_not_found(&err) && !is_network_or_host_down(&err.to_string(), false) && !err.to_string().contains("use of closed network connection") {
-                                    error!(
-                                        event = EVENT_LIFECYCLE_TIER_OPERATION_FAILED,
-                                        component = LOG_COMPONENT_ECSTORE,
-                                        subsystem = LOG_SUBSYSTEM_LIFECYCLE,
-                                        bucket = %task.obj_info.bucket,
-                                        object = %task.obj_info.name,
-                                        version_id = %task.obj_info.version_id.map(|v| v.to_string()).unwrap_or_default(),
-                                        tier = %task.event.storage_class,
-                                        operation = "transition_object",
-                                        error = %err,
-                                        "Lifecycle tier operation failed"
-                                    );
-                                }
+                        if let Err(err) =
+                            transition_object(api.clone(), &task.obj_info, LcAuditEvent::new(task.event.clone(), task.src.clone()))
+                                .await
+                        {
+                            if let Some(job_id) = task.manual_job_id {
+                                record_manual_transition_worker_result_for_task(
+                                    api.clone(),
+                                    job_id,
+                                    ManualTransitionWorkerResult::TierFailure,
+                                )
+                                .await;
+                            }
+                            global_metrics().record_scanner_transition_failed(1);
+                            if !is_err_version_not_found(&err) && !is_err_object_not_found(&err) && !is_network_or_host_down(&err.to_string(), false) && !err.to_string().contains("use of closed network connection") {
+                                error!(
+                                    event = EVENT_LIFECYCLE_TIER_OPERATION_FAILED,
+                                    component = LOG_COMPONENT_ECSTORE,
+                                    subsystem = LOG_SUBSYSTEM_LIFECYCLE,
+                                    bucket = %task.obj_info.bucket,
+                                    object = %task.obj_info.name,
+                                    version_id = %task.obj_info.version_id.map(|v| v.to_string()).unwrap_or_default(),
+                                    tier = %task.event.storage_class,
+                                    operation = "transition_object",
+                                    error = %err,
+                                    "Lifecycle tier operation failed"
+                                );
+                            }
                             emit_transition_failed_event(obj_info_for_event);
                         } else {
+                            if let Some(job_id) = task.manual_job_id {
+                                record_manual_transition_worker_result_for_task(
+                                    api.clone(),
+                                    job_id,
+                                    ManualTransitionWorkerResult::Completed,
+                                )
+                                .await;
+                            }
                             global_metrics().record_scanner_transition_completed(1);
                             let mut ts = TierStats {
                                 total_size: task.obj_info.size as u64,
@@ -1577,6 +1600,20 @@ impl TransitionState {
     }
 }
 
+async fn record_manual_transition_worker_result_for_task(api: Arc<ECStore>, job_id: Uuid, result: ManualTransitionWorkerResult) {
+    if let Err(err) = record_manual_transition_worker_result(api, job_id, result, manual_transition_queue_snapshot()).await {
+        warn!(
+            event = EVENT_LIFECYCLE_WORKER_STATE,
+            component = LOG_COMPONENT_ECSTORE,
+            subsystem = LOG_SUBSYSTEM_LIFECYCLE,
+            job_id = %job_id,
+            error = %err,
+            state = "manual_transition_worker_result_failed",
+            "Manual transition worker failed to persist job result"
+        );
+    }
+}
+
 pub async fn init_background_expiry(api: Arc<ECStore>) {
     let mut workers = get_env_usize("RUSTFS_MAX_EXPIRY_WORKERS", std::cmp::min(num_cpus::get(), 16));
     //globalILMConfig.getExpirationWorkers()
@@ -1606,7 +1643,7 @@ fn spawn_manual_transition_job_recovery_once(api: Arc<ECStore>) -> Option<JoinHa
         let cancel_token = runtime_sources::background_services_cancel_token().unwrap_or_default();
         select! {
             _ = cancel_token.cancelled() => {}
-            result = recover_manual_transition_jobs_once(api, DEFAULT_MANUAL_TRANSITION_JOB_RECOVERY_LIMIT, None) => {
+            result = recover_manual_transition_jobs(api, DEFAULT_MANUAL_TRANSITION_JOB_RECOVERY_LIMIT) => {
                 match result {
                     Ok(stats) => {
                         debug!(
@@ -1658,6 +1695,32 @@ enum ManualTransitionJobRecoveryOutcome {
     Skipped,
 }
 
+async fn recover_manual_transition_jobs(api: Arc<ECStore>, limit: usize) -> Result<ManualTransitionJobRecoveryStats, Error> {
+    let mut marker = None;
+    let mut total = ManualTransitionJobRecoveryStats::default();
+
+    loop {
+        let stats = recover_manual_transition_jobs_once(api.clone(), limit, marker).await?;
+        total.scanned = total.scanned.saturating_add(stats.scanned);
+        total.resumed = total.resumed.saturating_add(stats.resumed);
+        total.cancelled = total.cancelled.saturating_add(stats.cancelled);
+        total.skipped = total.skipped.saturating_add(stats.skipped);
+        total.failed = total.failed.saturating_add(stats.failed);
+
+        if !stats.truncated {
+            total.truncated = false;
+            total.next_marker = None;
+            return Ok(total);
+        }
+        let Some(next_marker) = stats.next_marker else {
+            return Err(Error::other("manual transition job recovery page is truncated without a next marker"));
+        };
+        total.truncated = true;
+        total.next_marker = Some(next_marker.clone());
+        marker = Some(next_marker);
+    }
+}
+
 pub async fn recover_manual_transition_jobs_once(
     api: Arc<ECStore>,
     limit: usize,
@@ -1771,11 +1834,16 @@ async fn recover_manual_transition_job(api: Arc<ECStore>, job_id: Uuid) -> Resul
     }
 
     let mut options = record.resume_options();
+    options.job_id = Some(job_id);
     options.cancel_check = Some(manual_transition_recovery_cancel_check(api.clone(), job_id));
     options.progress_sink = Some(manual_transition_recovery_progress_sink(api.clone(), job_id));
     let result = enqueue_transition_for_existing_objects_scoped(api.clone(), &record.bucket, options).await;
     let final_record = finalize_recovered_manual_transition_job(api.clone(), job_id, result).await?;
-    release_manual_transition_recovery_admission(api, &final_record).await;
+    if final_record.is_terminal() {
+        release_manual_transition_recovery_admission(api, &final_record).await;
+    } else {
+        spawn_manual_transition_recovery_heartbeat(api, job_id);
+    }
     Ok(ManualTransitionJobRecoveryOutcome::Resumed)
 }
 
@@ -1845,6 +1913,34 @@ async fn release_manual_transition_recovery_admission(api: Arc<ECStore>, record:
     }
 }
 
+fn spawn_manual_transition_recovery_heartbeat(api: Arc<ECStore>, job_id: Uuid) {
+    tokio::spawn(async move {
+        let mut interval = tokio::time::interval(std::time::Duration::from_secs(5));
+        loop {
+            interval.tick().await;
+            match renew_manual_transition_job_lease(api.clone(), job_id, manual_transition_queue_snapshot()).await {
+                Ok(record) if record.is_terminal() => {
+                    release_manual_transition_recovery_admission(api, &record).await;
+                    return;
+                }
+                Ok(_) => {}
+                Err(Error::ConfigNotFound) => return,
+                Err(err) => {
+                    warn!(
+                        event = EVENT_LIFECYCLE_WORKER_STATE,
+                        component = LOG_COMPONENT_ECSTORE,
+                        subsystem = LOG_SUBSYSTEM_LIFECYCLE,
+                        job_id = %job_id,
+                        error = %err,
+                        state = "manual_transition_recovery_heartbeat_failed",
+                        "Manual transition recovery failed to renew job lease"
+                    );
+                }
+            }
+        }
+    });
+}
+
 async fn abandon_manual_transition_recovery_lease(api: Arc<ECStore>, job_id: Uuid, lease_id: Uuid) -> Result<(), Error> {
     for _ in 0..4 {
         let (mut record, etag) = match load_manual_transition_job_record_with_etag(api.clone(), job_id).await {
@@ -2770,6 +2866,8 @@ pub struct ManualTransitionRunOptions {
     pub max_objects: Option<u64>,
     pub max_duration: Option<std::time::Duration>,
     #[serde(skip)]
+    pub job_id: Option<Uuid>,
+    #[serde(skip)]
     pub cancel_token: Option<CancellationToken>,
     #[serde(skip)]
     pub cancel_check: Option<ManualTransitionCancelCheck>,
@@ -2788,6 +2886,7 @@ impl std::fmt::Debug for ManualTransitionRunOptions {
             .field("dry_run", &self.dry_run)
             .field("max_objects", &self.max_objects)
             .field("max_duration", &self.max_duration)
+            .field("job_id", &self.job_id)
             .field("cancel_token", &self.cancel_token.is_some())
             .field("cancel_check", &self.cancel_check.is_some())
             .field("progress_sink", &self.progress_sink.is_some())
@@ -2810,6 +2909,10 @@ impl PartialEq for ManualTransitionRunOptions {
 
 impl Eq for ManualTransitionRunOptions {}
 
+fn is_zero_u64(value: &u64) -> bool {
+    *value == 0
+}
+
 #[derive(Debug, Clone, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
 #[serde(deny_unknown_fields)]
 pub struct ManualTransitionRunReport {
@@ -2832,6 +2935,10 @@ pub struct ManualTransitionRunReport {
     pub skipped_queue_full: u64,
     pub skipped_queue_closed: u64,
     pub skipped_queue_timeout: u64,
+    #[serde(default, skip_serializing_if = "is_zero_u64")]
+    pub transition_completed: u64,
+    #[serde(default, skip_serializing_if = "is_zero_u64")]
+    pub transition_failed: u64,
     pub tier_failure: u64,
     pub truncated_by_limit: bool,
     pub truncated_by_duration: bool,
@@ -2898,6 +3005,19 @@ impl ManualTransitionRunReport {
             }
         }
     }
+
+    pub fn merge_scan_report_preserving_worker(&mut self, scan_report: &ManualTransitionRunReport) {
+        let transition_completed = self.transition_completed;
+        let transition_failed = self.transition_failed;
+        *self = scan_report.clone();
+        self.transition_completed = transition_completed;
+        self.transition_failed = transition_failed;
+        self.tier_failure = scan_report.tier_failure.saturating_add(transition_failed);
+    }
+
+    pub fn worker_transition_pending(&self) -> bool {
+        self.transition_completed.saturating_add(self.transition_failed) < self.enqueued
+    }
 }
 
 #[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
@@ -3334,13 +3454,22 @@ async fn enqueue_transition_with_lifecycle_report(
                 report.skipped_tier = report.skipped_tier.saturating_add(1);
                 return false;
             }
+            if !options.dry_run
+                && !runtime_sources::tier_config_mgr_handle()
+                    .read()
+                    .await
+                    .is_tier_valid(&event.storage_class)
+            {
+                report.tier_failure = report.tier_failure.saturating_add(1);
+                return false;
+            }
             report.eligible = report.eligible.saturating_add(1);
             if options.dry_run {
                 report.dry_run_eligible = report.dry_run_eligible.saturating_add(1);
                 return true;
             }
             let outcome = runtime_sources::transition_state_handle()
-                .queue_transition_task_outcome(oi, &event, src)
+                .queue_transition_task_outcome(oi, &event, src, options.job_id)
                 .await;
             report.record_enqueue_outcome(outcome);
             return outcome.is_handled();
@@ -4223,12 +4352,12 @@ mod tests {
         lifecycle_rule_has_date_expiration, lifecycle_version_purge_state_from_completed_targets,
         manual_transition_duration_elapsed, manual_transition_has_more_after_limit, manual_transition_version_marker,
         mark_delete_opts_skip_decommissioned_on_remote_success, merge_stale_multipart_candidate,
-        persist_manual_transition_page_checkpoint, recover_manual_transition_jobs_once, replication_state_for_delete,
-        resolve_transition_queue_capacity, resolve_transition_queue_send_timeout, resolve_transition_worker_count,
-        resolve_transition_workers_absolute_max, run_tier_free_version_recovery_loop, select_restore_s3_location,
-        set_recovered_free_version_enqueue_observer, should_defer_date_expiry_for_recent_config_update,
-        should_reuse_lifecycle_delete_replication_state, transitioned_cleanup_tuple, transitioned_object_delete_opts,
-        wait_for_tier_free_version_recovery,
+        persist_manual_transition_page_checkpoint, recover_manual_transition_jobs, recover_manual_transition_jobs_once,
+        replication_state_for_delete, resolve_transition_queue_capacity, resolve_transition_queue_send_timeout,
+        resolve_transition_worker_count, resolve_transition_workers_absolute_max, run_tier_free_version_recovery_loop,
+        select_restore_s3_location, set_recovered_free_version_enqueue_observer,
+        should_defer_date_expiry_for_recent_config_update, should_reuse_lifecycle_delete_replication_state,
+        transitioned_cleanup_tuple, transitioned_object_delete_opts, wait_for_tier_free_version_recovery,
     };
     #[cfg(feature = "test-util")]
     use super::{delete_free_version_remote_object_then, encode_dir_object, get_transitioned_object_reader_with_tier_manager};
@@ -6375,10 +6504,10 @@ mod tests {
         };
 
         let first = state
-            .queue_transition_task_outcome(&object, &event, &LcEventSrc::Scanner)
+            .queue_transition_task_outcome(&object, &event, &LcEventSrc::Scanner, None)
             .await;
         let second = state
-            .queue_transition_task_outcome(&object, &event, &LcEventSrc::Scanner)
+            .queue_transition_task_outcome(&object, &event, &LcEventSrc::Scanner, None)
             .await;
 
         assert_eq!(first, TransitionEnqueueOutcome::Queued);
@@ -6406,10 +6535,10 @@ mod tests {
         };
 
         let first = state
-            .queue_transition_task_outcome(&first_object, &event, &LcEventSrc::Scanner)
+            .queue_transition_task_outcome(&first_object, &event, &LcEventSrc::Scanner, None)
             .await;
         let second = state
-            .queue_transition_task_outcome(&second_object, &event, &LcEventSrc::Scanner)
+            .queue_transition_task_outcome(&second_object, &event, &LcEventSrc::Scanner, None)
             .await;
 
         assert_eq!(first, TransitionEnqueueOutcome::Queued);
@@ -6910,6 +7039,22 @@ mod tests {
         assert_eq!(report.skipped_tier, 1);
     }
 
+    #[tokio::test]
+    async fn manual_transition_reports_runtime_tier_failure_before_enqueue() {
+        let lc = latest_transition_lifecycle();
+        let object = current_object(ReplicationStatusType::Completed);
+        let options = ManualTransitionRunOptions::default();
+        let mut report = ManualTransitionRunReport::new(&object.bucket, &options);
+
+        let handled = enqueue_transition_with_lifecycle_report(&object, &lc, &LcEventSrc::Scanner, &options, &mut report).await;
+
+        assert!(!handled);
+        assert_eq!(report.eligible, 0);
+        assert_eq!(report.enqueued, 0);
+        assert_eq!(report.tier_failure, 1);
+        assert!(!report.has_partial_enqueue());
+    }
+
     #[tokio::test]
     async fn manual_transition_counts_already_transitioned_object() {
         let lc = latest_transition_lifecycle();
@@ -7108,6 +7253,93 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    #[serial]
+    async fn manual_transition_recovery_drains_multiple_record_pages() {
+        let (_paths, ecstore) = setup_test_env().await;
+        let mut job_ids = Vec::new();
+
+        for bucket in ["manual-recovery-page-a", "manual-recovery-page-b"] {
+            let job_id = Uuid::new_v4();
+            let options = ManualTransitionRunOptions {
+                prefix: "logs/".to_string(),
+                ..Default::default()
+            };
+            let mut record = ManualTransitionJobRecord::new(job_id, bucket, &options, "old-owner");
+            record.lease_expires_at_unix_nanos = 0;
+            save_manual_transition_job_record(ecstore.clone(), &record)
+                .await
+                .expect("expired job record should save");
+            save_manual_transition_scope_admission_if_absent(ecstore.clone(), &ManualTransitionScopeAdmission::from_job(&record))
+                .await
+                .expect("expired scope admission should save");
+            job_ids.push((job_id, record.scope_key.clone()));
+        }
+
+        let stats = recover_manual_transition_jobs(ecstore.clone(), 1)
+            .await
+            .expect("manual transition recovery should drain all pages");
+
+        assert_eq!(stats.resumed, 2);
+        assert_eq!(stats.failed, 0);
+        assert!(!stats.truncated);
+        assert!(stats.next_marker.is_none());
+        for (job_id, scope_key) in job_ids {
+            let recovered = load_manual_transition_job_record(ecstore.clone(), job_id)
+                .await
+                .expect("recovered job should load");
+            assert_eq!(recovered.state, ManualTransitionJobState::Completed);
+            assert!(
+                matches!(
+                    load_manual_transition_scope_admission(ecstore.clone(), &scope_key).await,
+                    Err(Error::ConfigNotFound)
+                ),
+                "completed recovery must release every scope admission"
+            );
+        }
+    }
+
+    #[tokio::test]
+    #[serial]
+    async fn manual_transition_recovery_completes_expired_cancelled_record() {
+        let (_paths, ecstore) = setup_test_env().await;
+        let job_id = Uuid::new_v4();
+        let options = ManualTransitionRunOptions {
+            prefix: "logs/".to_string(),
+            ..Default::default()
+        };
+        let mut record = ManualTransitionJobRecord::new(job_id, "manual-recovery-cancel-bucket", &options, "old-owner");
+        record.lease_expires_at_unix_nanos = 0;
+        record.mark_cancel_requested();
+        save_manual_transition_job_record(ecstore.clone(), &record)
+            .await
+            .expect("expired cancelled job record should save");
+        save_manual_transition_scope_admission_if_absent(ecstore.clone(), &ManualTransitionScopeAdmission::from_job(&record))
+            .await
+            .expect("expired cancelled scope admission should save");
+
+        let stats = recover_manual_transition_jobs(ecstore.clone(), 10)
+            .await
+            .expect("manual transition recovery should process cancelled jobs");
+
+        assert_eq!(stats.cancelled, 1);
+        assert_eq!(stats.resumed, 0);
+        assert_eq!(stats.failed, 0);
+        let recovered = load_manual_transition_job_record(ecstore.clone(), job_id)
+            .await
+            .expect("cancelled job should load");
+        assert_eq!(recovered.state, ManualTransitionJobState::Cancelled);
+        assert!(recovered.cancel_requested);
+        assert!(recovered.report.cancelled);
+        assert!(
+            matches!(
+                load_manual_transition_scope_admission(ecstore, &record.scope_key).await,
+                Err(Error::ConfigNotFound)
+            ),
+            "cancelled recovery must release the scope admission"
+        );
+    }
+
     #[test]
     fn manual_transition_cancelled_report_is_partial_and_resumable() {
         let report = ManualTransitionRunReport {

crates/ecstore/src/bucket/lifecycle/manual_transition_job.rs

@@ -37,6 +37,10 @@ pub const MAX_MANUAL_TRANSITION_JOB_RECORD_SIZE: usize = 64 * 1024;
 const MANUAL_TRANSITION_JOB_LEASE_SECONDS: i128 = 60;
 const MANUAL_TRANSITION_LEGACY_SCOPE_SCAN_LIMIT: i32 = 1000;
 
+fn is_false(value: &bool) -> bool {
+    !*value
+}
+
 #[derive(Debug, thiserror::Error)]
 pub enum ManualTransitionJobError {
     #[error("manual transition job is corrupt: {0}")]
@@ -75,6 +79,8 @@ pub struct ManualTransitionJobRecord {
     pub lease_id: Uuid,
     pub lease_expires_at_unix_nanos: i128,
     pub state: ManualTransitionJobState,
+    #[serde(default, skip_serializing_if = "is_false")]
+    pub scan_completed: bool,
     pub cancel_requested: bool,
     pub created_at_unix_nanos: i128,
     pub updated_at_unix_nanos: i128,
@@ -103,6 +109,7 @@ impl ManualTransitionJobRecord {
             lease_id,
             lease_expires_at_unix_nanos: manual_transition_job_lease_expires_at(now),
             state: ManualTransitionJobState::Running,
+            scan_completed: false,
             cancel_requested: false,
             created_at_unix_nanos: now,
             updated_at_unix_nanos: now,
@@ -120,17 +127,11 @@ impl ManualTransitionJobRecord {
     }
 
     pub fn complete(&mut self, report: ManualTransitionRunReport, queue_snapshot: ManualTransitionQueueSnapshot) {
-        self.state = if report.cancelled {
-            ManualTransitionJobState::Cancelled
-        } else if report.was_truncated() || report.has_partial_enqueue() || report.tier_failure > 0 {
-            ManualTransitionJobState::Partial
-        } else {
-            ManualTransitionJobState::Completed
-        };
-        self.report = report;
+        self.scan_completed = true;
+        self.report.merge_scan_report_preserving_worker(&report);
         self.queue_snapshot = queue_snapshot;
         self.error = None;
-        self.mark_updated_terminal();
+        self.mark_terminal_if_worker_drained();
     }
 
     pub fn fail(&mut self, error: impl Into<String>) {
@@ -140,6 +141,24 @@ impl ManualTransitionJobRecord {
         self.mark_updated_terminal();
     }
 
+    pub fn record_worker_result(&mut self, result: ManualTransitionWorkerResult, queue_snapshot: ManualTransitionQueueSnapshot) {
+        if self.is_terminal() {
+            return;
+        }
+        match result {
+            ManualTransitionWorkerResult::Completed => {
+                self.report.transition_completed = self.report.transition_completed.saturating_add(1);
+            }
+            ManualTransitionWorkerResult::TierFailure => {
+                self.report.transition_failed = self.report.transition_failed.saturating_add(1);
+                self.report.tier_failure = self.report.tier_failure.saturating_add(1);
+            }
+        }
+        self.queue_snapshot = queue_snapshot;
+        self.updated_at_unix_nanos = OffsetDateTime::now_utc().unix_timestamp_nanos();
+        self.mark_terminal_if_worker_drained();
+    }
+
     pub fn mark_cancel_requested(&mut self) {
         self.cancel_requested = true;
         self.updated_at_unix_nanos = OffsetDateTime::now_utc().unix_timestamp_nanos();
@@ -178,9 +197,25 @@ impl ManualTransitionJobRecord {
         self.queue_snapshot = queue_snapshot;
     }
 
+    pub fn mark_unknown_if_worker_results_lost(&mut self, queue_snapshot: ManualTransitionQueueSnapshot) -> bool {
+        if self.state != ManualTransitionJobState::Running
+            || !self.scan_completed
+            || !self.report.worker_transition_pending()
+            || queue_snapshot.queued > 0
+            || queue_snapshot.active > 0
+        {
+            return false;
+        }
+        self.queue_snapshot = queue_snapshot;
+        self.state = ManualTransitionJobState::Unknown;
+        self.error = Some("manual transition worker result was not persisted before the transition queue drained".to_string());
+        self.mark_updated_terminal();
+        true
+    }
+
     pub fn update_running_progress(&mut self, report: ManualTransitionRunReport, queue_snapshot: ManualTransitionQueueSnapshot) {
         if self.state == ManualTransitionJobState::Running {
-            self.report = report;
+            self.report.merge_scan_report_preserving_worker(&report);
             self.renew_lease(queue_snapshot);
         }
     }
@@ -210,6 +245,24 @@ impl ManualTransitionJobRecord {
         self.completed_at_unix_nanos = Some(now);
     }
 
+    fn mark_terminal_if_worker_drained(&mut self) {
+        if !self.scan_completed || self.report.worker_transition_pending() {
+            return;
+        }
+        self.state = if self.cancel_requested || self.report.cancelled {
+            ManualTransitionJobState::Cancelled
+        } else if self.report.was_truncated()
+            || self.report.has_partial_enqueue()
+            || self.report.tier_failure > 0
+            || self.report.transition_failed > 0
+        {
+            ManualTransitionJobState::Partial
+        } else {
+            ManualTransitionJobState::Completed
+        };
+        self.mark_updated_terminal();
+    }
+
     pub fn encode(&self) -> Result<Vec<u8>, ManualTransitionJobError> {
         self.validate()?;
         let job_bytes = serde_json::to_vec(self)?;
@@ -275,6 +328,12 @@ impl ManualTransitionJobRecord {
     }
 }
 
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum ManualTransitionWorkerResult {
+    Completed,
+    TierFailure,
+}
+
 fn manual_transition_job_lease_expires_at(now_unix_nanos: i128) -> i128 {
     now_unix_nanos.saturating_add(MANUAL_TRANSITION_JOB_LEASE_SECONDS.saturating_mul(1_000_000_000))
 }
@@ -696,6 +755,40 @@ pub async fn persist_manual_transition_job_progress(
     Ok(record)
 }
 
+pub async fn record_manual_transition_worker_result(
+    api: Arc<ECStore>,
+    job_id: Uuid,
+    result: ManualTransitionWorkerResult,
+    queue_snapshot: ManualTransitionQueueSnapshot,
+) -> EcstoreResult<ManualTransitionJobRecord> {
+    for _ in 0..4 {
+        let (mut record, etag) = load_manual_transition_job_record_with_etag(api.clone(), job_id).await?;
+        if record.is_terminal() {
+            return Ok(record);
+        }
+        record.record_worker_result(result, queue_snapshot);
+        match save_manual_transition_job_record_if_current(api.clone(), &record, &etag).await {
+            Ok(()) => {
+                if record.is_terminal() {
+                    delete_manual_transition_scope_admission_if_current(
+                        api.clone(),
+                        &record.scope_key,
+                        record.job_id,
+                        record.lease_id,
+                    )
+                    .await?;
+                } else {
+                    renew_manual_transition_scope_admission_from_job(api, &record).await?;
+                }
+                return Ok(record);
+            }
+            Err(Error::PreconditionFailed) => continue,
+            Err(err) => return Err(err),
+        }
+    }
+    Err(Error::PreconditionFailed)
+}
+
 pub async fn renew_manual_transition_job_lease(
     api: Arc<ECStore>,
     job_id: Uuid,
@@ -703,9 +796,16 @@ pub async fn renew_manual_transition_job_lease(
 ) -> EcstoreResult<ManualTransitionJobRecord> {
     let (mut record, etag) = load_manual_transition_job_record_with_etag(api.clone(), job_id).await?;
     if record.state == ManualTransitionJobState::Running {
-        record.renew_lease(queue_snapshot);
+        let became_terminal = record.mark_unknown_if_worker_results_lost(queue_snapshot);
+        if !became_terminal {
+            record.renew_lease(queue_snapshot);
+        }
         save_manual_transition_job_record_if_current(api.clone(), &record, &etag).await?;
-        renew_manual_transition_scope_admission_from_job(api, &record).await?;
+        if became_terminal {
+            delete_manual_transition_scope_admission_if_current(api, &record.scope_key, record.job_id, record.lease_id).await?;
+        } else {
+            renew_manual_transition_scope_admission_from_job(api, &record).await?;
+        }
     }
     Ok(record)
 }
@@ -801,6 +901,88 @@ mod tests {
         assert_eq!(decoded.max_objects, Some(17));
     }
 
+    #[test]
+    fn manual_transition_job_record_waits_for_worker_results() {
+        let options = ManualTransitionRunOptions::default();
+        let mut record = ManualTransitionJobRecord::new(Uuid::new_v4(), "bucket", &options, TEST_OWNER);
+
+        record.complete(
+            ManualTransitionRunReport {
+                bucket: "bucket".to_string(),
+                enqueued: 2,
+                ..Default::default()
+            },
+            ManualTransitionQueueSnapshot::default(),
+        );
+
+        assert!(record.scan_completed);
+        assert_eq!(record.state, ManualTransitionJobState::Running);
+        assert!(record.completed_at_unix_nanos.is_none());
+
+        record.record_worker_result(ManualTransitionWorkerResult::Completed, ManualTransitionQueueSnapshot::default());
+        assert_eq!(record.state, ManualTransitionJobState::Running);
+        assert_eq!(record.report.transition_completed, 1);
+
+        record.record_worker_result(ManualTransitionWorkerResult::TierFailure, ManualTransitionQueueSnapshot::default());
+        assert_eq!(record.state, ManualTransitionJobState::Partial);
+        assert_eq!(record.report.transition_completed, 1);
+        assert_eq!(record.report.transition_failed, 1);
+        assert_eq!(record.report.tier_failure, 1);
+        assert!(record.completed_at_unix_nanos.is_some());
+    }
+
+    #[test]
+    fn manual_transition_job_scan_progress_preserves_worker_counters() {
+        let options = ManualTransitionRunOptions::default();
+        let mut record = ManualTransitionJobRecord::new(Uuid::new_v4(), "bucket", &options, TEST_OWNER);
+        record.record_worker_result(ManualTransitionWorkerResult::TierFailure, ManualTransitionQueueSnapshot::default());
+
+        record.update_running_progress(
+            ManualTransitionRunReport {
+                bucket: "bucket".to_string(),
+                scanned: 3,
+                enqueued: 1,
+                tier_failure: 2,
+                ..Default::default()
+            },
+            ManualTransitionQueueSnapshot::default(),
+        );
+
+        assert_eq!(record.report.scanned, 3);
+        assert_eq!(record.report.enqueued, 1);
+        assert_eq!(record.report.transition_failed, 1);
+        assert_eq!(record.report.tier_failure, 3);
+    }
+
+    #[test]
+    fn manual_transition_job_marks_unknown_when_worker_results_are_lost() {
+        let options = ManualTransitionRunOptions::default();
+        let mut record = ManualTransitionJobRecord::new(Uuid::new_v4(), "bucket", &options, TEST_OWNER);
+
+        record.complete(
+            ManualTransitionRunReport {
+                bucket: "bucket".to_string(),
+                enqueued: 1,
+                ..Default::default()
+            },
+            ManualTransitionQueueSnapshot {
+                queued: 1,
+                ..Default::default()
+            },
+        );
+
+        assert_eq!(record.state, ManualTransitionJobState::Running);
+        assert!(!record.mark_unknown_if_worker_results_lost(ManualTransitionQueueSnapshot {
+            queued: 1,
+            ..Default::default()
+        }));
+
+        assert!(record.mark_unknown_if_worker_results_lost(ManualTransitionQueueSnapshot::default()));
+        assert_eq!(record.state, ManualTransitionJobState::Unknown);
+        assert!(record.error.as_deref().is_some_and(|err| err.contains("worker result")));
+        assert!(record.completed_at_unix_nanos.is_some());
+    }
+
     #[test]
     fn manual_transition_job_record_builds_resume_options() {
         let options = ManualTransitionRunOptions {