feat(ilm): persist manual transition task journals (#5296)

* feat(ilm): persist manual transition task journal

Co-Authored-By: heihutu <heihutu@gmail.com>

* fix(ilm): reconcile manual task journals on recovery

Co-Authored-By: heihutu <heihutu@gmail.com>

---------

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

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

@@ -23,12 +23,13 @@ use crate::bucket::lifecycle::lifecycle::{
 };
 use crate::bucket::lifecycle::manual_transition_job::{
     MANUAL_TRANSITION_JOB_RECORD_PREFIX, ManualTransitionJobRecord, ManualTransitionJobState, ManualTransitionScopeAdmission,
-    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, manual_transition_worker_result_task_key, persist_manual_transition_job_progress,
-    reconcile_manual_transition_worker_results, record_manual_transition_worker_result, renew_manual_transition_job_lease,
-    save_manual_transition_job_record_if_current,
+    ManualTransitionScopeAdmissionClaim, ManualTransitionTaskRecord, 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,
+    manual_transition_worker_result_task_key, persist_manual_transition_job_progress, reconcile_manual_transition_worker_results,
+    record_manual_transition_worker_result, renew_manual_transition_job_lease, save_manual_transition_job_record_if_current,
+    save_manual_transition_task_if_absent,
 };
 use crate::bucket::lifecycle::replication_sink;
 use crate::bucket::lifecycle::replication_sink::{
@@ -1155,6 +1156,7 @@ enum TransitionEnqueueOutcome {
     QueueFull,
     QueueClosed,
     QueueSendTimedOut,
+    TaskJournalFailed,
 }
 
 impl TransitionEnqueueOutcome {
@@ -1401,6 +1403,7 @@ impl TransitionState {
 
     async fn queue_transition_task_outcome(
         self: &Arc<Self>,
+        api: Option<Arc<ECStore>>,
         oi: &ObjectInfo,
         event: &lifecycle::Event,
         src: &LcEventSrc,
@@ -1422,13 +1425,58 @@ impl TransitionState {
             return TransitionEnqueueOutcome::AlreadyInFlight;
         }
 
+        let manual_result_key =
+            manual_job_id.map(|_| manual_transition_worker_result_task_key(&oi.bucket, &oi.name, oi.version_id));
+        if let (Some(job_id), Some(result_key)) = (manual_job_id, manual_result_key.as_deref()) {
+            let Some(api) = api else {
+                self.release_transition(oi);
+                warn!(
+                    event = EVENT_LIFECYCLE_WORKER_STATE,
+                    component = LOG_COMPONENT_ECSTORE,
+                    subsystem = LOG_SUBSYSTEM_LIFECYCLE,
+                    bucket = %oi.bucket,
+                    object = %oi.name,
+                    version_id = %oi.version_id.map(|v| v.to_string()).unwrap_or_default(),
+                    job_id = %job_id,
+                    state = "manual_transition_task_journal_missing_store",
+                    "Manual transition task was not enqueued because no task journal store was available"
+                );
+                self.record_scanner_transition_state();
+                return TransitionEnqueueOutcome::TaskJournalFailed;
+            };
+            let task_record = ManualTransitionTaskRecord::new(
+                job_id,
+                result_key,
+                oi.bucket.clone(),
+                oi.name.clone(),
+                oi.version_id,
+                event.storage_class.clone(),
+            );
+            if let Err(err) = save_manual_transition_task_if_absent(api, &task_record).await {
+                self.release_transition(oi);
+                warn!(
+                    event = EVENT_LIFECYCLE_WORKER_STATE,
+                    component = LOG_COMPONENT_ECSTORE,
+                    subsystem = LOG_SUBSYSTEM_LIFECYCLE,
+                    bucket = %oi.bucket,
+                    object = %oi.name,
+                    version_id = %oi.version_id.map(|v| v.to_string()).unwrap_or_default(),
+                    job_id = %job_id,
+                    error = %err,
+                    state = "manual_transition_task_journal_failed",
+                    "Manual transition task was not enqueued because its task journal marker could not be persisted"
+                );
+                self.record_scanner_transition_state();
+                return TransitionEnqueueOutcome::TaskJournalFailed;
+            }
+        }
+
         let task = TransitionTask {
             obj_info: oi.clone(),
             src: src.clone(),
             event: event.clone(),
             manual_job_id,
-            manual_result_key: manual_job_id
-                .map(|_| manual_transition_worker_result_task_key(&oi.bucket, &oi.name, oi.version_id)),
+            manual_result_key,
         };
         if is_immediate_transition_source(src) {
             let outcome = match self.transition_tx.try_send(Some(task)) {
@@ -1510,7 +1558,9 @@ 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, None).await.is_handled()
+        self.queue_transition_task_outcome(None, oi, event, src, None)
+            .await
+            .is_handled()
     }
 
     pub async fn init(api: Arc<ECStore>) {
@@ -1963,7 +2013,7 @@ async fn recover_manual_transition_job(
     }
 
     let recovery_unknown_snapshot = ManualTransitionQueueSnapshot::default();
-    if record.scan_completed && record.report.worker_transition_pending() {
+    if record.scan_completed {
         let reconciled = reconcile_manual_transition_worker_results(api.clone(), job_id, recovery_unknown_snapshot).await?;
         if reconciled.is_terminal() {
             release_manual_transition_recovery_admission(api, &reconciled).await;
@@ -3246,6 +3296,9 @@ impl ManualTransitionRunReport {
             TransitionEnqueueOutcome::QueueSendTimedOut => {
                 self.skipped_queue_timeout = self.skipped_queue_timeout.saturating_add(1);
             }
+            TransitionEnqueueOutcome::TaskJournalFailed => {
+                self.skipped_queue_closed = self.skipped_queue_closed.saturating_add(1);
+            }
         }
     }
 
@@ -3391,7 +3444,7 @@ pub async fn enqueue_transition_for_existing_objects_scoped(
                 return Ok(report);
             }
             report.scanned = report.scanned.saturating_add(1);
-            enqueue_transition_with_lifecycle_report(object, &lc, &src, &options, &mut report).await;
+            enqueue_transition_with_lifecycle_report(Some(api.clone()), object, &lc, &src, &options, &mut report).await;
             if report.has_partial_enqueue() {
                 report.next_marker.clone_from(&previous_marker);
                 report.next_version_idmarker.clone_from(&previous_version_marker);
@@ -3656,6 +3709,7 @@ fn manual_transition_version_marker(oi: &ObjectInfo) -> String {
 }
 
 async fn enqueue_transition_with_lifecycle_report(
+    api: Option<Arc<ECStore>>,
     oi: &ObjectInfo,
     lc: &BucketLifecycleConfiguration,
     src: &LcEventSrc,
@@ -3713,7 +3767,7 @@ async fn enqueue_transition_with_lifecycle_report(
                 return true;
             }
             let outcome = runtime_sources::transition_state_handle()
-                .queue_transition_task_outcome(oi, &event, src, options.job_id)
+                .queue_transition_task_outcome(api.clone(), oi, &event, src, options.job_id)
                 .await;
             report.record_enqueue_outcome(outcome);
             return outcome.is_handled();
@@ -4606,13 +4660,12 @@ mod tests {
         manual_transition_duration_elapsed, manual_transition_has_more_after_limit, manual_transition_recovery_progress_sink,
         manual_transition_version_marker, mark_delete_opts_skip_decommissioned_on_remote_success,
         merge_stale_multipart_candidate, persist_manual_transition_job_progress, persist_manual_transition_page_checkpoint,
-        recover_manual_transition_job, recover_manual_transition_jobs, recover_manual_transition_jobs_once,
-        replication_state_for_delete, resolve_tier_free_version_recovery_enabled, 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_lifecycle_observability_observer,
-        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,
+        recover_manual_transition_job, recover_manual_transition_jobs, replication_state_for_delete,
+        resolve_tier_free_version_recovery_enabled, 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_lifecycle_observability_observer, 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};
@@ -4623,14 +4676,15 @@ mod tests {
     use crate::bucket::lifecycle::config_boundary;
     use crate::bucket::lifecycle::manual_transition_job::{
         ManualTransitionJobRecord, ManualTransitionJobState, ManualTransitionScopeAdmission, ManualTransitionScopeAdmissionClaim,
-        ManualTransitionWorkerResult, ManualTransitionWorkerResultRecord, claim_manual_transition_scope_admission,
-        delete_manual_transition_scope_admission_if_current, legacy_manual_transition_scope_key,
-        load_manual_transition_job_record, load_manual_transition_scope_admission,
-        load_manual_transition_scope_admission_with_etag, manual_transition_scope_record_object_name,
-        manual_transition_worker_result_object_name, manual_transition_worker_result_task_key,
-        reconcile_manual_transition_worker_results, record_manual_transition_worker_result, renew_manual_transition_job_lease,
-        request_manual_transition_job_cancel, save_manual_transition_job_record,
-        save_manual_transition_scope_admission_if_absent, save_manual_transition_scope_admission_if_current,
+        ManualTransitionTaskRecord, ManualTransitionWorkerResult, ManualTransitionWorkerResultRecord,
+        claim_manual_transition_scope_admission, delete_manual_transition_scope_admission_if_current,
+        legacy_manual_transition_scope_key, load_manual_transition_job_record, load_manual_transition_scope_admission,
+        load_manual_transition_scope_admission_with_etag, load_manual_transition_task_record,
+        manual_transition_scope_record_object_name, manual_transition_worker_result_object_name,
+        manual_transition_worker_result_task_key, reconcile_manual_transition_worker_results,
+        record_manual_transition_worker_result, renew_manual_transition_job_lease, request_manual_transition_job_cancel,
+        save_manual_transition_job_record, save_manual_transition_scope_admission_if_absent,
+        save_manual_transition_scope_admission_if_current, save_manual_transition_task_if_absent,
         save_manual_transition_worker_result_if_absent,
     };
     use crate::bucket::lifecycle::replication_sink::{
@@ -6823,10 +6877,10 @@ mod tests {
         };
 
         let first = state
-            .queue_transition_task_outcome(&object, &event, &LcEventSrc::Scanner, None)
+            .queue_transition_task_outcome(None, &object, &event, &LcEventSrc::Scanner, None)
             .await;
         let second = state
-            .queue_transition_task_outcome(&object, &event, &LcEventSrc::Scanner, None)
+            .queue_transition_task_outcome(None, &object, &event, &LcEventSrc::Scanner, None)
             .await;
 
         assert_eq!(first, TransitionEnqueueOutcome::Queued);
@@ -6854,10 +6908,10 @@ mod tests {
         };
 
         let first = state
-            .queue_transition_task_outcome(&first_object, &event, &LcEventSrc::Scanner, None)
+            .queue_transition_task_outcome(None, &first_object, &event, &LcEventSrc::Scanner, None)
             .await;
         let second = state
-            .queue_transition_task_outcome(&second_object, &event, &LcEventSrc::Scanner, None)
+            .queue_transition_task_outcome(None, &second_object, &event, &LcEventSrc::Scanner, None)
             .await;
 
         assert_eq!(first, TransitionEnqueueOutcome::Queued);
@@ -6865,6 +6919,73 @@ mod tests {
         assert_eq!(state.transition_rx.len(), 1);
     }
 
+    #[tokio::test]
+    #[serial]
+    async fn queue_transition_task_outcome_persists_manual_task_journal() {
+        let (_paths, ecstore) = setup_test_env().await;
+        let state = TransitionState::new_with_capacity(4);
+        let job_id = Uuid::new_v4();
+        let version_id = Uuid::new_v4();
+        let object = ObjectInfo {
+            bucket: "manual-task-journal-bucket".to_string(),
+            name: "logs/object".to_string(),
+            version_id: Some(version_id),
+            ..Default::default()
+        };
+        let event = crate::bucket::lifecycle::lifecycle::Event {
+            action: IlmAction::TransitionAction,
+            storage_class: "WARM".to_string(),
+            ..Default::default()
+        };
+        let task_key = manual_transition_worker_result_task_key(&object.bucket, &object.name, object.version_id);
+
+        let outcome = state
+            .queue_transition_task_outcome(Some(ecstore.clone()), &object, &event, &LcEventSrc::Scanner, Some(job_id))
+            .await;
+
+        assert_eq!(outcome, TransitionEnqueueOutcome::Queued);
+        assert_eq!(state.transition_rx.len(), 1);
+        let task_record = load_manual_transition_task_record(ecstore, job_id, &task_key)
+            .await
+            .expect("manual task journal marker should load");
+        assert_eq!(task_record.job_id, job_id);
+        assert_eq!(task_record.task_key, task_key);
+        assert_eq!(task_record.bucket, object.bucket);
+        assert_eq!(task_record.object, object.name);
+        assert_eq!(task_record.version_id, Some(version_id));
+        assert_eq!(task_record.storage_class, "WARM");
+    }
+
+    #[tokio::test]
+    #[serial]
+    async fn queue_transition_task_outcome_fails_closed_when_manual_task_journal_fails() {
+        let (_paths, ecstore) = setup_test_env().await;
+        let state = TransitionState::new_with_capacity(4);
+        let object = ObjectInfo {
+            bucket: "manual-task-journal-fail-bucket".to_string(),
+            name: "logs/object".to_string(),
+            ..Default::default()
+        };
+        let event = crate::bucket::lifecycle::lifecycle::Event {
+            action: IlmAction::TransitionAction,
+            storage_class: "WARM".to_string(),
+            ..Default::default()
+        };
+
+        let failed = state
+            .queue_transition_task_outcome(Some(ecstore.clone()), &object, &event, &LcEventSrc::Scanner, Some(Uuid::nil()))
+            .await;
+
+        assert_eq!(failed, TransitionEnqueueOutcome::TaskJournalFailed);
+        assert_eq!(state.transition_rx.len(), 0);
+
+        let retried = state
+            .queue_transition_task_outcome(Some(ecstore), &object, &event, &LcEventSrc::Scanner, Some(Uuid::new_v4()))
+            .await;
+        assert_eq!(retried, TransitionEnqueueOutcome::Queued);
+        assert_eq!(state.transition_rx.len(), 1);
+    }
+
     #[tokio::test]
     #[serial]
     async fn queue_transition_task_dedupes_immediate_and_scanner_sources_for_same_version() {
@@ -7322,7 +7443,8 @@ mod tests {
         };
         let mut report = ManualTransitionRunReport::new(&object.bucket, &options);
 
-        let handled = enqueue_transition_with_lifecycle_report(&object, &lc, &LcEventSrc::Scanner, &options, &mut report).await;
+        let handled =
+            enqueue_transition_with_lifecycle_report(None, &object, &lc, &LcEventSrc::Scanner, &options, &mut report).await;
 
         assert!(handled);
         assert_eq!(report.eligible, 1);
@@ -7343,7 +7465,8 @@ mod tests {
         };
         let mut report = ManualTransitionRunReport::new(&object.bucket, &options);
 
-        let handled = enqueue_transition_with_lifecycle_report(&object, &lc, &LcEventSrc::Scanner, &options, &mut report).await;
+        let handled =
+            enqueue_transition_with_lifecycle_report(None, &object, &lc, &LcEventSrc::Scanner, &options, &mut report).await;
 
         assert!(!handled);
         assert_eq!(report.eligible, 0);
@@ -7361,7 +7484,8 @@ mod tests {
         };
         let mut report = ManualTransitionRunReport::new(&object.bucket, &options);
 
-        let handled = enqueue_transition_with_lifecycle_report(&object, &lc, &LcEventSrc::Scanner, &options, &mut report).await;
+        let handled =
+            enqueue_transition_with_lifecycle_report(None, &object, &lc, &LcEventSrc::Scanner, &options, &mut report).await;
 
         assert!(!handled);
         assert_eq!(report.eligible, 0);
@@ -7375,7 +7499,8 @@ mod tests {
         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;
+        let handled =
+            enqueue_transition_with_lifecycle_report(None, &object, &lc, &LcEventSrc::Scanner, &options, &mut report).await;
 
         assert!(!handled);
         assert_eq!(report.eligible, 0);
@@ -7426,7 +7551,8 @@ mod tests {
         };
         let mut report = ManualTransitionRunReport::new(&object.bucket, &options);
 
-        let handled = enqueue_transition_with_lifecycle_report(&object, &lc, &LcEventSrc::Scanner, &options, &mut report).await;
+        let handled =
+            enqueue_transition_with_lifecycle_report(None, &object, &lc, &LcEventSrc::Scanner, &options, &mut report).await;
 
         assert!(!handled);
         assert_eq!(report.eligible, 0);
@@ -7920,13 +8046,11 @@ mod tests {
             .await
             .expect("expired scope admission should save");
 
-        let stats = recover_manual_transition_jobs_once(ecstore.clone(), 10, None)
+        let outcome = recover_manual_transition_job(ecstore.clone(), job_id, ManualTransitionQueueSnapshot::default())
             .await
             .expect("manual transition recovery should run");
 
-        assert!(stats.scanned >= 1);
-        assert_eq!(stats.resumed, 1);
-        assert_eq!(stats.failed, 0);
+        assert_eq!(outcome, ManualTransitionJobRecoveryOutcome::Resumed);
         let recovered = load_manual_transition_job_record(ecstore.clone(), job_id)
             .await
             .expect("recovered job should load");
@@ -8214,14 +8338,11 @@ mod tests {
             .await
             .expect("expired cancelled scope admission should save");
 
-        let stats = recover_manual_transition_jobs(ecstore.clone(), 10)
+        let outcome = recover_manual_transition_job(ecstore.clone(), job_id, ManualTransitionQueueSnapshot::default())
             .await
             .expect("manual transition recovery should process cancelled jobs");
 
-        assert_eq!(stats.cancelled, 1);
-        assert_eq!(stats.unknown, 0);
-        assert_eq!(stats.resumed, 0);
-        assert_eq!(stats.failed, 0);
+        assert_eq!(outcome, ManualTransitionJobRecoveryOutcome::Cancelled);
         let recovered = load_manual_transition_job_record(ecstore.clone(), job_id)
             .await
             .expect("cancelled job should load");
@@ -8424,6 +8545,52 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    #[serial]
+    async fn manual_transition_worker_result_heartbeat_uses_task_journal_enqueued_floor() {
+        let (_paths, ecstore) = setup_test_env().await;
+        let job_id = Uuid::new_v4();
+        let bucket = format!("manual-task-heartbeat-{}", job_id.simple());
+        let mut record = ManualTransitionJobRecord::new(job_id, &bucket, &ManualTransitionRunOptions::default(), "worker-owner");
+        record.scan_completed = true;
+        save_manual_transition_job_record(ecstore.clone(), &record)
+            .await
+            .expect("worker heartbeat task journal job record should save");
+        save_manual_transition_scope_admission_if_absent(ecstore.clone(), &ManualTransitionScopeAdmission::from_job(&record))
+            .await
+            .expect("worker heartbeat task journal admission should save");
+        let task_key = manual_transition_worker_result_task_key(&bucket, "logs/a", None);
+        let task_marker = ManualTransitionTaskRecord::new(job_id, &task_key, &bucket, "logs/a", None, "WARM");
+        assert!(
+            save_manual_transition_task_if_absent(ecstore.clone(), &task_marker)
+                .await
+                .expect("task journal marker should save"),
+            "new task journal marker must be created"
+        );
+        let result_marker = ManualTransitionWorkerResultRecord::new(job_id, &task_key, ManualTransitionWorkerResult::Completed);
+        assert!(
+            save_manual_transition_worker_result_if_absent(ecstore.clone(), &result_marker)
+                .await
+                .expect("worker result marker should save"),
+            "new worker result marker must be created"
+        );
+
+        let renewed = renew_manual_transition_job_lease(ecstore.clone(), job_id, ManualTransitionQueueSnapshot::default())
+            .await
+            .expect("heartbeat should reconcile task and result journals");
+
+        assert_eq!(renewed.state, ManualTransitionJobState::Completed);
+        assert_eq!(renewed.report.enqueued, 1);
+        assert_eq!(renewed.report.transition_completed, 1);
+        assert!(
+            matches!(
+                load_manual_transition_scope_admission(ecstore, &record.scope_key).await,
+                Err(Error::ConfigNotFound)
+            ),
+            "terminal task journal heartbeat must release the scope admission"
+        );
+    }
+
     #[tokio::test]
     #[serial]
     async fn manual_transition_worker_result_recovery_applies_marker_before_record_update() {
@@ -8474,6 +8641,108 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    #[serial]
+    async fn manual_transition_recovery_uses_task_journal_as_enqueued_floor() {
+        let (_paths, ecstore) = setup_test_env().await;
+        let job_id = Uuid::new_v4();
+        let bucket = format!("manual-task-floor-{}", job_id.simple());
+        let mut record = ManualTransitionJobRecord::new(job_id, &bucket, &ManualTransitionRunOptions::default(), "old-owner");
+        record.scan_completed = true;
+        record.lease_expires_at_unix_nanos = 0;
+        save_manual_transition_job_record(ecstore.clone(), &record)
+            .await
+            .expect("task journal recovery job record should save");
+        save_manual_transition_scope_admission_if_absent(ecstore.clone(), &ManualTransitionScopeAdmission::from_job(&record))
+            .await
+            .expect("task journal recovery admission should save");
+        let task_key = manual_transition_worker_result_task_key(&bucket, "logs/a", None);
+        let task_marker = ManualTransitionTaskRecord::new(job_id, &task_key, &bucket, "logs/a", None, "WARM");
+        assert!(
+            save_manual_transition_task_if_absent(ecstore.clone(), &task_marker)
+                .await
+                .expect("task journal marker should save"),
+            "new task journal marker must be created"
+        );
+        let result_marker = ManualTransitionWorkerResultRecord::new(job_id, &task_key, ManualTransitionWorkerResult::Completed);
+        assert!(
+            save_manual_transition_worker_result_if_absent(ecstore.clone(), &result_marker)
+                .await
+                .expect("worker result marker should save"),
+            "new worker result marker must be created"
+        );
+
+        let outcome = recover_manual_transition_job(ecstore.clone(), job_id, ManualTransitionQueueSnapshot::default())
+            .await
+            .expect("recovery should reconcile task and result journals");
+
+        assert_eq!(outcome, ManualTransitionJobRecoveryOutcome::Resumed);
+        let recovered = load_manual_transition_job_record(ecstore.clone(), job_id)
+            .await
+            .expect("reconciled task journal job should load");
+        assert_eq!(recovered.state, ManualTransitionJobState::Completed);
+        assert_eq!(recovered.report.enqueued, 1);
+        assert_eq!(recovered.report.transition_completed, 1);
+        assert!(
+            matches!(
+                load_manual_transition_scope_admission(ecstore, &record.scope_key).await,
+                Err(Error::ConfigNotFound)
+            ),
+            "terminal task journal recovery must release the scope admission"
+        );
+    }
+
+    #[tokio::test]
+    #[serial]
+    async fn manual_transition_recovery_marks_unknown_for_task_journal_without_result() {
+        let (_paths, ecstore) = setup_test_env().await;
+        let job_id = Uuid::new_v4();
+        let bucket = format!("manual-task-lost-result-{}", job_id.simple());
+        let mut record = ManualTransitionJobRecord::new(job_id, &bucket, &ManualTransitionRunOptions::default(), "old-owner");
+        record.scan_completed = true;
+        record.lease_expires_at_unix_nanos = 0;
+        save_manual_transition_job_record(ecstore.clone(), &record)
+            .await
+            .expect("lost-result job record should save");
+        save_manual_transition_scope_admission_if_absent(ecstore.clone(), &ManualTransitionScopeAdmission::from_job(&record))
+            .await
+            .expect("lost-result admission should save");
+        let task_key = manual_transition_worker_result_task_key(&bucket, "logs/a", None);
+        let task_marker = ManualTransitionTaskRecord::new(job_id, &task_key, &bucket, "logs/a", None, "WARM");
+        assert!(
+            save_manual_transition_task_if_absent(ecstore.clone(), &task_marker)
+                .await
+                .expect("task journal marker should save"),
+            "new task journal marker must be created"
+        );
+
+        let outcome = recover_manual_transition_job(ecstore.clone(), job_id, ManualTransitionQueueSnapshot::default())
+            .await
+            .expect("recovery should fail closed when a task journal marker has no worker result");
+
+        assert_eq!(outcome, ManualTransitionJobRecoveryOutcome::Unknown);
+        let recovered = load_manual_transition_job_record(ecstore.clone(), job_id)
+            .await
+            .expect("unknown task journal job should load");
+        assert_eq!(recovered.state, ManualTransitionJobState::Unknown);
+        assert_eq!(recovered.report.enqueued, 1);
+        assert_eq!(recovered.report.transition_completed, 0);
+        assert_eq!(recovered.report.transition_failed, 0);
+        assert!(
+            recovered
+                .error
+                .as_deref()
+                .is_some_and(|error| error.contains("worker result was not persisted"))
+        );
+        assert!(
+            matches!(
+                load_manual_transition_scope_admission(ecstore, &record.scope_key).await,
+                Err(Error::ConfigNotFound)
+            ),
+            "unknown task journal recovery must release the scope admission"
+        );
+    }
+
     #[tokio::test]
     #[serial]
     async fn manual_transition_worker_result_recovery_marks_unknown_for_corrupt_marker() {

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

@@ -31,14 +31,18 @@ use crate::storage_api_contracts::object::HTTPPreconditions;
 use crate::store::ECStore;
 
 pub const MANUAL_TRANSITION_JOB_SCHEMA: &str = "rustfs-manual-transition-job-v1";
+pub const MANUAL_TRANSITION_TASK_SCHEMA: &str = "rustfs-manual-transition-task-v1";
 pub const MANUAL_TRANSITION_WORKER_RESULT_SCHEMA: &str = "rustfs-manual-transition-worker-result-v1";
 pub const MANUAL_TRANSITION_JOB_RECORD_PREFIX: &str = "ilm/manual-transition/jobs";
 pub const MANUAL_TRANSITION_SCOPE_RECORD_PREFIX: &str = "ilm/manual-transition/scopes";
+pub const MANUAL_TRANSITION_TASK_PREFIX: &str = "ilm/manual-transition/tasks";
 pub const MANUAL_TRANSITION_WORKER_RESULT_PREFIX: &str = "ilm/manual-transition/results";
 pub const MAX_MANUAL_TRANSITION_JOB_RECORD_SIZE: usize = 64 * 1024;
+pub const MAX_MANUAL_TRANSITION_TASK_RECORD_SIZE: usize = 16 * 1024;
 pub const MAX_MANUAL_TRANSITION_WORKER_RESULT_RECORD_SIZE: usize = 8 * 1024;
 const MANUAL_TRANSITION_JOB_LEASE_SECONDS: i128 = 60;
 const MANUAL_TRANSITION_LEGACY_SCOPE_SCAN_LIMIT: i32 = 1000;
+const MANUAL_TRANSITION_TASK_SCAN_LIMIT: i32 = 1000;
 const MANUAL_TRANSITION_WORKER_RESULT_SCAN_LIMIT: i32 = 1000;
 
 fn is_false(value: &bool) -> bool {
@@ -160,6 +164,21 @@ impl ManualTransitionJobRecord {
         true
     }
 
+    pub fn mark_unknown_for_task_journal_error(
+        &mut self,
+        error: impl Into<String>,
+        queue_snapshot: ManualTransitionQueueSnapshot,
+    ) -> bool {
+        if self.state != ManualTransitionJobState::Running || !self.scan_completed {
+            return false;
+        }
+        self.queue_snapshot = queue_snapshot;
+        self.state = ManualTransitionJobState::Unknown;
+        self.error = Some(format!("manual transition task journal is corrupt: {}", error.into()));
+        self.mark_updated_terminal();
+        true
+    }
+
     pub fn cancel_after_recovery(&mut self, queue_snapshot: ManualTransitionQueueSnapshot) {
         if self.state == ManualTransitionJobState::Running && self.cancel_requested {
             self.scan_completed = true;
@@ -189,11 +208,18 @@ impl ManualTransitionJobRecord {
         self.mark_terminal_if_worker_drained();
     }
 
-    fn apply_worker_result_counts(&mut self, completed: u64, failed: u64, queue_snapshot: ManualTransitionQueueSnapshot) -> bool {
+    fn apply_worker_result_counts(
+        &mut self,
+        completed: u64,
+        failed: u64,
+        task_queued: u64,
+        queue_snapshot: ManualTransitionQueueSnapshot,
+    ) -> bool {
         if self.is_terminal() {
             return false;
         }
-        if completed.saturating_add(failed) > self.report.enqueued {
+        let enqueued = self.report.enqueued.max(task_queued);
+        if completed.saturating_add(failed) > enqueued {
             self.queue_snapshot = queue_snapshot;
             self.state = ManualTransitionJobState::Unknown;
             self.error = Some("manual transition worker result journal exceeds enqueued count".to_string());
@@ -202,10 +228,14 @@ impl ManualTransitionJobRecord {
         }
         let transition_completed = self.report.transition_completed.max(completed);
         let transition_failed = self.report.transition_failed.max(failed);
-        if transition_completed == self.report.transition_completed && transition_failed == self.report.transition_failed {
+        if enqueued == self.report.enqueued
+            && transition_completed == self.report.transition_completed
+            && transition_failed == self.report.transition_failed
+        {
             return false;
         }
         let scan_tier_failure = self.report.tier_failure.saturating_sub(self.report.transition_failed);
+        self.report.enqueued = enqueued;
         self.report.transition_completed = transition_completed;
         self.report.transition_failed = transition_failed;
         self.report.tier_failure = scan_tier_failure.saturating_add(transition_failed);
@@ -415,6 +445,111 @@ pub enum ManualTransitionWorkerResult {
     TierFailure,
 }
 
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct ManualTransitionTaskRecord {
+    pub schema: String,
+    pub job_id: Uuid,
+    pub task_key: String,
+    pub bucket: String,
+    pub object: String,
+    pub version_id: Option<Uuid>,
+    pub storage_class: String,
+    pub queued_at_unix_nanos: i128,
+}
+
+impl ManualTransitionTaskRecord {
+    pub fn new(
+        job_id: Uuid,
+        task_key: impl Into<String>,
+        bucket: impl Into<String>,
+        object: impl Into<String>,
+        version_id: Option<Uuid>,
+        storage_class: impl Into<String>,
+    ) -> Self {
+        Self {
+            schema: MANUAL_TRANSITION_TASK_SCHEMA.to_string(),
+            job_id,
+            task_key: task_key.into(),
+            bucket: bucket.into(),
+            object: object.into(),
+            version_id,
+            storage_class: storage_class.into(),
+            queued_at_unix_nanos: OffsetDateTime::now_utc().unix_timestamp_nanos(),
+        }
+    }
+
+    pub fn encode(&self) -> Result<Vec<u8>, ManualTransitionJobError> {
+        self.validate()?;
+        let record_bytes = serde_json::to_vec(self)?;
+        let content_sha256 = hex_sha256(&record_bytes, ToOwned::to_owned);
+        let persisted = PersistedManualTransitionTaskRecord {
+            schema: MANUAL_TRANSITION_TASK_SCHEMA.to_string(),
+            content_sha256,
+            record: self.clone(),
+        };
+        let encoded = serde_json::to_vec(&persisted)?;
+        if encoded.len() > MAX_MANUAL_TRANSITION_TASK_RECORD_SIZE {
+            return Err(ManualTransitionJobError::Corrupt("encoded task record exceeds maximum size"));
+        }
+        Ok(encoded)
+    }
+
+    pub fn decode(expected_job_id: Uuid, expected_task_key: &str, data: &[u8]) -> Result<Self, ManualTransitionJobError> {
+        if data.len() > MAX_MANUAL_TRANSITION_TASK_RECORD_SIZE {
+            return Err(ManualTransitionJobError::Corrupt("encoded task record exceeds maximum size"));
+        }
+        let persisted: PersistedManualTransitionTaskRecord = serde_json::from_slice(data)?;
+        if persisted.schema != MANUAL_TRANSITION_TASK_SCHEMA {
+            return Err(ManualTransitionJobError::UnsupportedSchema(persisted.schema));
+        }
+        if !is_sha256_checksum(&persisted.content_sha256) {
+            return Err(ManualTransitionJobError::Corrupt("task record content checksum is not a sha256 checksum"));
+        }
+        let record = persisted.record;
+        let record_bytes = serde_json::to_vec(&record)?;
+        let actual_checksum = hex_sha256(&record_bytes, ToOwned::to_owned);
+        if persisted.content_sha256 != actual_checksum {
+            return Err(ManualTransitionJobError::ChecksumMismatch);
+        }
+        if record.job_id != expected_job_id {
+            return Err(ManualTransitionJobError::Corrupt("task record job_id does not match record key"));
+        }
+        if record.task_key != expected_task_key {
+            return Err(ManualTransitionJobError::Corrupt("task record task_key does not match record key"));
+        }
+        record.validate()?;
+        Ok(record)
+    }
+
+    fn validate(&self) -> Result<(), ManualTransitionJobError> {
+        if self.job_id.is_nil() {
+            return Err(ManualTransitionJobError::Corrupt("task record job_id is nil"));
+        }
+        if !is_sha256_checksum(&self.task_key) {
+            return Err(ManualTransitionJobError::Corrupt("task record task_key is not a sha256 checksum"));
+        }
+        if self.bucket.is_empty() {
+            return Err(ManualTransitionJobError::Corrupt("task record bucket is empty"));
+        }
+        if self.object.is_empty() {
+            return Err(ManualTransitionJobError::Corrupt("task record object is empty"));
+        }
+        if self.storage_class.trim().is_empty() {
+            return Err(ManualTransitionJobError::Corrupt("task record storage_class is empty"));
+        }
+        Ok(())
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+struct PersistedManualTransitionTaskRecord {
+    schema: String,
+    content_sha256: String,
+    record: ManualTransitionTaskRecord,
+}
+
 #[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
 pub struct ManualTransitionWorkerResultStats {
     pub completed: u64,
@@ -661,6 +796,17 @@ pub fn manual_transition_worker_result_object_prefix(job_id: Uuid) -> Result<Str
     manual_transition_job_sharded_prefix(MANUAL_TRANSITION_WORKER_RESULT_PREFIX, job_id)
 }
 
+pub fn manual_transition_task_object_prefix(job_id: Uuid) -> Result<String, ManualTransitionJobError> {
+    manual_transition_job_sharded_prefix(MANUAL_TRANSITION_TASK_PREFIX, job_id)
+}
+
+pub fn manual_transition_task_object_name(job_id: Uuid, task_key: &str) -> Result<String, ManualTransitionJobError> {
+    if !is_sha256_checksum(task_key) {
+        return Err(ManualTransitionJobError::Corrupt("task record task_key is not a sha256 checksum"));
+    }
+    Ok(format!("{}/{}.json", manual_transition_task_object_prefix(job_id)?, task_key))
+}
+
 pub fn manual_transition_worker_result_object_name(job_id: Uuid, task_key: &str) -> Result<String, ManualTransitionJobError> {
     if !is_sha256_checksum(task_key) {
         return Err(ManualTransitionJobError::Corrupt("worker result task_key is not a sha256 checksum"));
@@ -687,6 +833,22 @@ fn manual_transition_worker_result_task_key_from_object_name(
     Ok(task_key.to_string())
 }
 
+fn manual_transition_task_key_from_object_name(job_id: Uuid, object_name: &str) -> Result<String, ManualTransitionJobError> {
+    let prefix = manual_transition_task_object_prefix(job_id)?;
+    let rest = object_name
+        .strip_prefix(&prefix)
+        .ok_or(ManualTransitionJobError::Corrupt("task record object prefix is invalid"))?
+        .strip_prefix('/')
+        .ok_or(ManualTransitionJobError::Corrupt("task record object path is invalid"))?;
+    let task_key = rest
+        .strip_suffix(".json")
+        .ok_or(ManualTransitionJobError::Corrupt("task record suffix is invalid"))?;
+    if task_key.contains('/') || !is_sha256_checksum(task_key) {
+        return Err(ManualTransitionJobError::Corrupt("task record task_key is invalid"));
+    }
+    Ok(task_key.to_string())
+}
+
 pub fn manual_transition_job_id_from_record_object_name(object_name: &str) -> Result<Uuid, ManualTransitionJobError> {
     let Some(rest) = object_name.strip_prefix(MANUAL_TRANSITION_JOB_RECORD_PREFIX) else {
         return Err(ManualTransitionJobError::Corrupt("job record object prefix is invalid"));
@@ -812,6 +974,97 @@ pub(crate) async fn save_manual_transition_worker_result_if_absent(
     }
 }
 
+pub(crate) async fn save_manual_transition_task_if_absent(
+    api: Arc<ECStore>,
+    record: &ManualTransitionTaskRecord,
+) -> EcstoreResult<bool> {
+    let object =
+        manual_transition_task_object_name(record.job_id, &record.task_key).map_err(manual_transition_job_store_error)?;
+    let data = record.encode().map_err(manual_transition_job_store_error)?;
+    match config_boundary::save_config_with_opts(
+        api,
+        &object,
+        data,
+        &ObjectOptions {
+            max_parity: true,
+            http_preconditions: Some(HTTPPreconditions {
+                if_none_match: Some("*".to_string()),
+                ..Default::default()
+            }),
+            ..Default::default()
+        },
+    )
+    .await
+    {
+        Ok(()) => Ok(true),
+        Err(Error::PreconditionFailed) => Ok(false),
+        Err(err) => Err(err),
+    }
+}
+
+pub async fn load_manual_transition_task_record(
+    api: Arc<ECStore>,
+    job_id: Uuid,
+    task_key: &str,
+) -> EcstoreResult<ManualTransitionTaskRecord> {
+    let object = manual_transition_task_object_name(job_id, task_key).map_err(manual_transition_job_store_error)?;
+    let data = config_boundary::read_config(api, &object).await?;
+    ManualTransitionTaskRecord::decode(job_id, task_key, &data).map_err(manual_transition_job_store_error)
+}
+
+#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
+struct ManualTransitionTaskJournalStats {
+    queued: u64,
+}
+
+enum ManualTransitionTaskJournal {
+    Stats(ManualTransitionTaskJournalStats),
+    Corrupt(String),
+}
+
+async fn scan_manual_transition_task_journal(api: Arc<ECStore>, job_id: Uuid) -> EcstoreResult<ManualTransitionTaskJournal> {
+    let prefix = manual_transition_task_object_prefix(job_id).map_err(manual_transition_job_store_error)?;
+    let mut marker = None;
+    let mut stats = ManualTransitionTaskJournalStats::default();
+    loop {
+        let page = api
+            .clone()
+            .list_objects_v2(
+                RUSTFS_META_BUCKET,
+                &prefix,
+                marker,
+                None,
+                MANUAL_TRANSITION_TASK_SCAN_LIMIT,
+                false,
+                None,
+                false,
+            )
+            .await?;
+        for object in page.objects {
+            let task_key = match manual_transition_task_key_from_object_name(job_id, &object.name) {
+                Ok(task_key) => task_key,
+                Err(err) => return Ok(ManualTransitionTaskJournal::Corrupt(err.to_string())),
+            };
+            let object_name = manual_transition_task_object_name(job_id, &task_key).map_err(manual_transition_job_store_error)?;
+            let data = match config_boundary::read_config(api.clone(), &object_name).await {
+                Ok(data) => data,
+                Err(err) => return Err(err),
+            };
+            if let Err(err) = ManualTransitionTaskRecord::decode(job_id, &task_key, &data) {
+                return Ok(ManualTransitionTaskJournal::Corrupt(err.to_string()));
+            }
+            stats.queued = stats.queued.saturating_add(1);
+        }
+        if !page.is_truncated {
+            return Ok(ManualTransitionTaskJournal::Stats(stats));
+        }
+        let Some(next_marker) = page.next_continuation_token else {
+            return Err(Error::other("manual transition task journal page is truncated without a next marker"));
+        };
+        marker = Some(next_marker);
+    }
+}
+
 pub async fn load_manual_transition_worker_result_stats(
     api: Arc<ECStore>,
     job_id: Uuid,
@@ -880,6 +1133,12 @@ pub async fn reconcile_manual_transition_worker_results(
     job_id: Uuid,
     queue_snapshot: ManualTransitionQueueSnapshot,
 ) -> EcstoreResult<ManualTransitionJobRecord> {
+    let task_stats = match scan_manual_transition_task_journal(api.clone(), job_id).await? {
+        ManualTransitionTaskJournal::Stats(stats) => stats,
+        ManualTransitionTaskJournal::Corrupt(error) => {
+            return mark_manual_transition_job_unknown_for_task_journal_error(api, job_id, error, queue_snapshot).await;
+        }
+    };
     let stats = match scan_manual_transition_worker_result_journal(api.clone(), job_id).await? {
         ManualTransitionWorkerResultJournal::Stats(stats) => stats,
         ManualTransitionWorkerResultJournal::Corrupt(error) => {
@@ -888,7 +1147,7 @@ pub async fn reconcile_manual_transition_worker_results(
     };
     for _ in 0..4 {
         let (mut record, etag) = load_manual_transition_job_record_with_etag(api.clone(), job_id).await?;
-        let changed = record.apply_worker_result_counts(stats.completed, stats.failed, queue_snapshot);
+        let changed = record.apply_worker_result_counts(stats.completed, stats.failed, task_stats.queued, queue_snapshot);
         if !changed {
             return Ok(record);
         }
@@ -914,6 +1173,30 @@ pub async fn reconcile_manual_transition_worker_results(
     Err(Error::PreconditionFailed)
 }
 
+async fn mark_manual_transition_job_unknown_for_task_journal_error(
+    api: Arc<ECStore>,
+    job_id: Uuid,
+    error: String,
+    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.mark_unknown_for_task_journal_error(error.clone(), queue_snapshot) {
+            return Ok(record);
+        }
+        match save_manual_transition_job_record_if_current(api.clone(), &record, &etag).await {
+            Ok(()) => {
+                delete_manual_transition_scope_admission_if_current(api, &record.scope_key, record.job_id, record.lease_id)
+                    .await?;
+                return Ok(record);
+            }
+            Err(Error::PreconditionFailed) => continue,
+            Err(err) => return Err(err),
+        }
+    }
+    Err(Error::PreconditionFailed)
+}
+
 async fn mark_manual_transition_job_unknown_for_worker_result_journal_error(
     api: Arc<ECStore>,
     job_id: Uuid,
@@ -1199,11 +1482,7 @@ pub async fn renew_manual_transition_job_lease(
 ) -> EcstoreResult<ManualTransitionJobRecord> {
     let (mut record, mut etag) = load_manual_transition_job_record_with_etag(api.clone(), job_id).await?;
     if record.state == ManualTransitionJobState::Running {
-        if record.scan_completed
-            && record.report.worker_transition_pending()
-            && queue_snapshot.queued == 0
-            && queue_snapshot.active == 0
-        {
+        if record.scan_completed && queue_snapshot.queued == 0 && queue_snapshot.active == 0 {
             record = reconcile_manual_transition_worker_results(api.clone(), job_id, queue_snapshot).await?;
             if record.is_terminal() || !record.report.worker_transition_pending() {
                 return Ok(record);
@@ -1588,6 +1867,22 @@ mod tests {
         assert!(matches!(err, ManualTransitionJobError::ChecksumMismatch));
     }
 
+    #[test]
+    fn manual_transition_task_record_rejects_checksum_drift() {
+        let job_id = Uuid::new_v4();
+        let task_key = manual_transition_worker_result_task_key("bucket", "logs/a", Some(Uuid::new_v4()));
+        let record = ManualTransitionTaskRecord::new(job_id, &task_key, "bucket", "logs/a", Some(Uuid::new_v4()), "WARM");
+        let encoded = record.encode().expect("task record should encode");
+        let mut value: serde_json::Value = serde_json::from_slice(&encoded).expect("encoded task record should be json");
+        value["record"]["storage_class"] = serde_json::Value::String("COLD".to_string());
+        let mutated = serde_json::to_vec(&value).expect("mutated task record should encode");
+
+        let err =
+            ManualTransitionTaskRecord::decode(job_id, &task_key, &mutated).expect_err("task checksum drift must fail closed");
+
+        assert!(matches!(err, ManualTransitionJobError::ChecksumMismatch));
+    }
+
     #[test]
     fn manual_transition_job_record_rejects_unknown_report_fields() {
         let options = ManualTransitionRunOptions::default();