fix(ilm): deduplicate manual transition worker results

Persist per-task manual transition worker result markers before applying job counters so duplicate worker completion reports are no-ops. Reconcile persisted markers during drained-queue lease renewal and recovery to restore marker-before-record crash windows. Fail closed when persisted worker result markers are corrupt. Co-Authored-By: heihutu <heihutu@gmail.com>
2026-07-26 08:18:18 +00:00 · 2026-07-26 12:29:18 +08:00
parent b4e3c7117e
commit 12fcac61d2
2 changed files with 756 additions and 12 deletions
@@ -22,12 +22,13 @@ use crate::bucket::lifecycle::lifecycle::{
    self, Lifecycle, ObjectOpts, TransitionOptions, abort_incomplete_multipart_upload_due,
 };
 use crate::bucket::lifecycle::manual_transition_job::{
-    MANUAL_TRANSITION_JOB_RECORD_PREFIX, ManualTransitionJobRecord, ManualTransitionScopeAdmission,
+    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, persist_manual_transition_job_progress, record_manual_transition_worker_result,
-    renew_manual_transition_job_lease, save_manual_transition_job_record_if_current,
+    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,
 };
 use crate::bucket::lifecycle::replication_sink;
 use crate::bucket::lifecycle::replication_sink::{
@@ -1097,6 +1098,7 @@ struct TransitionTask {
    src: LcEventSrc,
    event: lifecycle::Event,
    manual_job_id: Option<Uuid>,
+    manual_result_key: Option<String>,
 }

 impl ExpiryOp for TransitionTask {
@@ -1425,6 +1427,8 @@ impl TransitionState {
            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)),
        };
        if is_immediate_transition_source(src) {
            let outcome = match self.transition_tx.try_send(Some(task)) {
@@ -1603,10 +1607,11 @@ impl TransitionState {
                            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 {
+                            if let (Some(job_id), Some(result_key)) = (task.manual_job_id, task.manual_result_key.as_deref()) {
                                record_manual_transition_worker_result_for_task(
                                    api.clone(),
                                    job_id,
+                                    result_key,
                                    ManualTransitionWorkerResult::TierFailure,
                                )
                                .await;
@@ -1628,10 +1633,11 @@ impl TransitionState {
                            }
                            emit_transition_failed_event(obj_info_for_event);
                        } else {
-                            if let Some(job_id) = task.manual_job_id {
+                            if let (Some(job_id), Some(result_key)) = (task.manual_job_id, task.manual_result_key.as_deref()) {
                                record_manual_transition_worker_result_for_task(
                                    api.clone(),
                                    job_id,
+                                    result_key,
                                    ManualTransitionWorkerResult::Completed,
                                )
                                .await;
@@ -1741,8 +1747,15 @@ 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 {
+async fn record_manual_transition_worker_result_for_task(
+    api: Arc<ECStore>,
+    job_id: Uuid,
+    result_key: &str,
+    result: ManualTransitionWorkerResult,
+) {
+    if let Err(err) =
+        record_manual_transition_worker_result(api, job_id, result_key, result, manual_transition_queue_snapshot()).await
+    {
        warn!(
            event = EVENT_LIFECYCLE_WORKER_STATE,
            component = LOG_COMPONENT_ECSTORE,
@@ -1940,7 +1953,7 @@ async fn recover_manual_transition_job(
    job_id: Uuid,
    queue_snapshot: ManualTransitionQueueSnapshot,
 ) -> Result<ManualTransitionJobRecoveryOutcome, Error> {
-    let (mut record, etag) = match load_manual_transition_job_record_with_etag(api.clone(), job_id).await {
+    let (mut record, mut etag) = match load_manual_transition_job_record_with_etag(api.clone(), job_id).await {
        Ok(record) => record,
        Err(Error::ConfigNotFound) => return Ok(ManualTransitionJobRecoveryOutcome::Skipped),
        Err(err) => return Err(err),
@@ -1950,6 +1963,21 @@ async fn recover_manual_transition_job(
    }

    let recovery_unknown_snapshot = ManualTransitionQueueSnapshot::default();
+    if record.scan_completed && record.report.worker_transition_pending() {
+        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;
+            return match reconciled.state {
+                ManualTransitionJobState::Cancelled => Ok(ManualTransitionJobRecoveryOutcome::Cancelled),
+                ManualTransitionJobState::Unknown => Ok(ManualTransitionJobRecoveryOutcome::Unknown),
+                _ => Ok(ManualTransitionJobRecoveryOutcome::Resumed),
+            };
+        }
+        if reconciled != record {
+            (record, etag) = load_manual_transition_job_record_with_etag(api.clone(), job_id).await?;
+        }
+    }
+
    if record.mark_unknown_if_worker_results_lost(recovery_unknown_snapshot)
        || record.mark_unknown_if_recovery_would_skip_pending_page(recovery_unknown_snapshot)
    {
@@ -4594,12 +4622,15 @@ mod tests {
    use crate::bucket::lifecycle::config_boundary;
    use crate::bucket::lifecycle::manual_transition_job::{
        ManualTransitionJobRecord, ManualTransitionJobState, ManualTransitionScopeAdmission, ManualTransitionScopeAdmissionClaim,
-        ManualTransitionWorkerResult, claim_manual_transition_scope_admission,
+        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,
-        renew_manual_transition_job_lease, request_manual_transition_job_cancel, save_manual_transition_job_record,
+        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_worker_result_if_absent,
    };
    use crate::bucket::lifecycle::replication_sink::{
        ReplicateDecision, ReplicateTargetDecision, ReplicationStatusType, VersionPurgeStatusType,
@@ -8002,6 +8033,303 @@ mod tests {
        );
    }

+    #[tokio::test]
+    #[serial]
+    async fn manual_transition_worker_result_duplicate_marker_is_noop() {
+        let (_paths, ecstore) = setup_test_env().await;
+        let job_id = Uuid::new_v4();
+        let bucket = format!("manual-worker-result-{}", job_id.simple());
+        let options = ManualTransitionRunOptions::default();
+        let mut record = ManualTransitionJobRecord::new(job_id, &bucket, &options, "worker-owner");
+        record.complete(
+            ManualTransitionRunReport {
+                bucket: bucket.clone(),
+                enqueued: 2,
+                ..Default::default()
+            },
+            ManualTransitionQueueSnapshot::default(),
+        );
+        save_manual_transition_job_record(ecstore.clone(), &record)
+            .await
+            .expect("worker result job record should save");
+        save_manual_transition_scope_admission_if_absent(ecstore.clone(), &ManualTransitionScopeAdmission::from_job(&record))
+            .await
+            .expect("worker result admission should save");
+
+        let first_key = manual_transition_worker_result_task_key(&bucket, "logs/a", None);
+        let first = record_manual_transition_worker_result(
+            ecstore.clone(),
+            job_id,
+            &first_key,
+            ManualTransitionWorkerResult::Completed,
+            ManualTransitionQueueSnapshot::default(),
+        )
+        .await
+        .expect("first worker result should persist");
+        assert_eq!(first.state, ManualTransitionJobState::Running);
+        assert_eq!(first.report.transition_completed, 1);
+        assert_eq!(first.report.transition_failed, 0);
+
+        let duplicate = record_manual_transition_worker_result(
+            ecstore.clone(),
+            job_id,
+            &first_key,
+            ManualTransitionWorkerResult::Completed,
+            ManualTransitionQueueSnapshot::default(),
+        )
+        .await
+        .expect("duplicate worker result should be idempotent");
+        assert_eq!(duplicate.state, ManualTransitionJobState::Running);
+        assert_eq!(duplicate.report.transition_completed, 1);
+        assert_eq!(duplicate.report.transition_failed, 0);
+
+        let second_key = manual_transition_worker_result_task_key(&bucket, "logs/b", None);
+        let final_record = record_manual_transition_worker_result(
+            ecstore.clone(),
+            job_id,
+            &second_key,
+            ManualTransitionWorkerResult::TierFailure,
+            ManualTransitionQueueSnapshot::default(),
+        )
+        .await
+        .expect("second distinct worker result should persist");
+
+        assert_eq!(final_record.state, ManualTransitionJobState::Partial);
+        assert_eq!(final_record.report.transition_completed, 1);
+        assert_eq!(final_record.report.transition_failed, 1);
+        assert_eq!(final_record.report.tier_failure, 1);
+        assert!(
+            matches!(
+                load_manual_transition_scope_admission(ecstore, &record.scope_key).await,
+                Err(Error::ConfigNotFound)
+            ),
+            "terminal worker result must release the scope admission"
+        );
+    }
+
+    #[tokio::test]
+    #[serial]
+    async fn manual_transition_worker_result_reconcile_applies_marker_and_releases_admission() {
+        let (_paths, ecstore) = setup_test_env().await;
+        let job_id = Uuid::new_v4();
+        let bucket = format!("manual-worker-reconcile-{}", job_id.simple());
+        let options = ManualTransitionRunOptions::default();
+        let mut record = ManualTransitionJobRecord::new(job_id, &bucket, &options, "worker-owner");
+        record.complete(
+            ManualTransitionRunReport {
+                bucket: bucket.clone(),
+                enqueued: 1,
+                ..Default::default()
+            },
+            ManualTransitionQueueSnapshot::default(),
+        );
+        save_manual_transition_job_record(ecstore.clone(), &record)
+            .await
+            .expect("worker reconcile job record should save");
+        save_manual_transition_scope_admission_if_absent(ecstore.clone(), &ManualTransitionScopeAdmission::from_job(&record))
+            .await
+            .expect("worker reconcile admission should save");
+
+        let task_key = manual_transition_worker_result_task_key(&bucket, "logs/a", None);
+        let marker = ManualTransitionWorkerResultRecord::new(job_id, &task_key, ManualTransitionWorkerResult::Completed);
+        assert!(
+            save_manual_transition_worker_result_if_absent(ecstore.clone(), &marker)
+                .await
+                .expect("worker result marker should save"),
+            "new worker result marker must be created once"
+        );
+        assert!(
+            !save_manual_transition_worker_result_if_absent(ecstore.clone(), &marker)
+                .await
+                .expect("duplicate worker result marker should not fail"),
+            "duplicate worker result marker must be reported as existing"
+        );
+        let duplicate_noop = record_manual_transition_worker_result(
+            ecstore.clone(),
+            job_id,
+            &task_key,
+            ManualTransitionWorkerResult::Completed,
+            ManualTransitionQueueSnapshot::default(),
+        )
+        .await
+        .expect("duplicate worker result should not apply journal counts");
+        assert_eq!(duplicate_noop.state, ManualTransitionJobState::Running);
+        assert_eq!(duplicate_noop.report.transition_completed, 0);
+        assert_eq!(duplicate_noop.report.transition_failed, 0);
+
+        let reconciled =
+            reconcile_manual_transition_worker_results(ecstore.clone(), job_id, ManualTransitionQueueSnapshot::default())
+                .await
+                .expect("worker result marker should reconcile into job record");
+
+        assert_eq!(reconciled.state, ManualTransitionJobState::Completed);
+        assert_eq!(reconciled.report.transition_completed, 1);
+        assert_eq!(reconciled.report.transition_failed, 0);
+        assert_eq!(reconciled.report.tier_failure, 0);
+        assert!(
+            matches!(
+                load_manual_transition_scope_admission(ecstore, &record.scope_key).await,
+                Err(Error::ConfigNotFound)
+            ),
+            "terminal reconcile must release the scope admission"
+        );
+    }
+
+    #[tokio::test]
+    #[serial]
+    async fn manual_transition_worker_result_heartbeat_applies_marker_before_record_update() {
+        let (_paths, ecstore) = setup_test_env().await;
+        let job_id = Uuid::new_v4();
+        let bucket = format!("manual-worker-heartbeat-{}", job_id.simple());
+        let mut record = ManualTransitionJobRecord::new(job_id, &bucket, &ManualTransitionRunOptions::default(), "worker-owner");
+        record.complete(
+            ManualTransitionRunReport {
+                bucket: bucket.clone(),
+                enqueued: 1,
+                ..Default::default()
+            },
+            ManualTransitionQueueSnapshot::default(),
+        );
+        save_manual_transition_job_record(ecstore.clone(), &record)
+            .await
+            .expect("worker heartbeat job record should save");
+        save_manual_transition_scope_admission_if_absent(ecstore.clone(), &ManualTransitionScopeAdmission::from_job(&record))
+            .await
+            .expect("worker heartbeat admission should save");
+        let task_key = manual_transition_worker_result_task_key(&bucket, "logs/a", None);
+        let marker = ManualTransitionWorkerResultRecord::new(job_id, &task_key, ManualTransitionWorkerResult::Completed);
+        assert!(
+            save_manual_transition_worker_result_if_absent(ecstore.clone(), &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 marker before unknown fallback");
+
+        assert_eq!(renewed.state, ManualTransitionJobState::Completed);
+        assert_eq!(renewed.report.transition_completed, 1);
+        assert!(
+            matches!(
+                load_manual_transition_scope_admission(ecstore, &record.scope_key).await,
+                Err(Error::ConfigNotFound)
+            ),
+            "terminal heartbeat reconcile must release the scope admission"
+        );
+    }
+
+    #[tokio::test]
+    #[serial]
+    async fn manual_transition_worker_result_recovery_applies_marker_before_record_update() {
+        let (_paths, ecstore) = setup_test_env().await;
+        let job_id = Uuid::new_v4();
+        let bucket = format!("manual-worker-recovery-{}", job_id.simple());
+        let mut record = ManualTransitionJobRecord::new(job_id, &bucket, &ManualTransitionRunOptions::default(), "old-owner");
+        record.complete(
+            ManualTransitionRunReport {
+                bucket: bucket.clone(),
+                enqueued: 1,
+                ..Default::default()
+            },
+            ManualTransitionQueueSnapshot::default(),
+        );
+        record.lease_expires_at_unix_nanos = 0;
+        save_manual_transition_job_record(ecstore.clone(), &record)
+            .await
+            .expect("worker recovery job record should save");
+        save_manual_transition_scope_admission_if_absent(ecstore.clone(), &ManualTransitionScopeAdmission::from_job(&record))
+            .await
+            .expect("worker recovery admission should save");
+        let task_key = manual_transition_worker_result_task_key(&bucket, "logs/a", None);
+        let marker = ManualTransitionWorkerResultRecord::new(job_id, &task_key, ManualTransitionWorkerResult::Completed);
+        assert!(
+            save_manual_transition_worker_result_if_absent(ecstore.clone(), &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 marker before unknown fallback");
+
+        assert_eq!(outcome, ManualTransitionJobRecoveryOutcome::Resumed);
+        let recovered = load_manual_transition_job_record(ecstore.clone(), job_id)
+            .await
+            .expect("reconciled job should load");
+        assert_eq!(recovered.state, ManualTransitionJobState::Completed);
+        assert_eq!(recovered.report.transition_completed, 1);
+        assert!(
+            matches!(
+                load_manual_transition_scope_admission(ecstore, &record.scope_key).await,
+                Err(Error::ConfigNotFound)
+            ),
+            "terminal recovery reconcile must release the scope admission"
+        );
+    }
+
+    #[tokio::test]
+    #[serial]
+    async fn manual_transition_worker_result_recovery_marks_unknown_for_corrupt_marker() {
+        let (_paths, ecstore) = setup_test_env().await;
+        let job_id = Uuid::new_v4();
+        let bucket = format!("manual-worker-corrupt-{}", job_id.simple());
+        let mut record = ManualTransitionJobRecord::new(job_id, &bucket, &ManualTransitionRunOptions::default(), "old-owner");
+        record.complete(
+            ManualTransitionRunReport {
+                bucket: bucket.clone(),
+                enqueued: 1,
+                ..Default::default()
+            },
+            ManualTransitionQueueSnapshot::default(),
+        );
+        record.lease_expires_at_unix_nanos = 0;
+        save_manual_transition_job_record(ecstore.clone(), &record)
+            .await
+            .expect("corrupt marker job record should save");
+        save_manual_transition_scope_admission_if_absent(ecstore.clone(), &ManualTransitionScopeAdmission::from_job(&record))
+            .await
+            .expect("corrupt marker admission should save");
+        let task_key = manual_transition_worker_result_task_key(&bucket, "logs/a", None);
+        let marker = ManualTransitionWorkerResultRecord::new(job_id, &task_key, ManualTransitionWorkerResult::Completed);
+        let encoded = marker.encode().expect("worker result marker should encode");
+        let mut value: serde_json::Value = serde_json::from_slice(&encoded).expect("worker result marker should be json");
+        value["record"]["result"] = serde_json::Value::String("tier_failure".to_string());
+        let object = manual_transition_worker_result_object_name(job_id, &task_key).expect("worker result object should encode");
+        config_boundary::save_config(
+            ecstore.clone(),
+            &object,
+            serde_json::to_vec(&value).expect("corrupt marker should encode"),
+        )
+        .await
+        .expect("corrupt worker result marker should save");
+
+        let outcome = recover_manual_transition_job(ecstore.clone(), job_id, ManualTransitionQueueSnapshot::default())
+            .await
+            .expect("recovery should fail closed on corrupt marker");
+
+        assert_eq!(outcome, ManualTransitionJobRecoveryOutcome::Unknown);
+        let recovered = load_manual_transition_job_record(ecstore.clone(), job_id)
+            .await
+            .expect("unknown job should load");
+        assert_eq!(recovered.state, ManualTransitionJobState::Unknown);
+        assert!(
+            recovered
+                .error
+                .as_deref()
+                .is_some_and(|error| error.contains("worker result journal is corrupt"))
+        );
+        assert!(
+            matches!(
+                load_manual_transition_scope_admission(ecstore, &record.scope_key).await,
+                Err(Error::ConfigNotFound)
+            ),
+            "corrupt marker unknown recovery must release the scope admission"
+        );
+    }
+
    #[test]
    fn manual_transition_cancelled_report_is_partial_and_resumable() {
        let report = ManualTransitionRunReport {
@@ -31,11 +31,15 @@ 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_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_WORKER_RESULT_PREFIX: &str = "ilm/manual-transition/results";
 pub const MAX_MANUAL_TRANSITION_JOB_RECORD_SIZE: usize = 64 * 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_WORKER_RESULT_SCAN_LIMIT: i32 = 1000;

 fn is_false(value: &bool) -> bool {
    !*value
@@ -141,6 +145,21 @@ impl ManualTransitionJobRecord {
        self.mark_updated_terminal();
    }

+    pub fn mark_unknown_for_worker_result_journal_error(
+        &mut self,
+        error: impl Into<String>,
+        queue_snapshot: ManualTransitionQueueSnapshot,
+    ) -> bool {
+        if self.state != ManualTransitionJobState::Running || !self.scan_completed || !self.report.worker_transition_pending() {
+            return false;
+        }
+        self.queue_snapshot = queue_snapshot;
+        self.state = ManualTransitionJobState::Unknown;
+        self.error = Some(format!("manual transition worker result 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;
@@ -170,6 +189,32 @@ impl ManualTransitionJobRecord {
        self.mark_terminal_if_worker_drained();
    }

+    fn apply_worker_result_counts(&mut self, completed: u64, failed: u64, queue_snapshot: ManualTransitionQueueSnapshot) -> bool {
+        if self.is_terminal() {
+            return false;
+        }
+        if completed.saturating_add(failed) > self.report.enqueued {
+            self.queue_snapshot = queue_snapshot;
+            self.state = ManualTransitionJobState::Unknown;
+            self.error = Some("manual transition worker result journal exceeds enqueued count".to_string());
+            self.mark_updated_terminal();
+            return true;
+        }
+        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 {
+            return false;
+        }
+        let scan_tier_failure = self.report.tier_failure.saturating_sub(self.report.transition_failed);
+        self.report.transition_completed = transition_completed;
+        self.report.transition_failed = transition_failed;
+        self.report.tier_failure = scan_tier_failure.saturating_add(transition_failed);
+        self.queue_snapshot = queue_snapshot;
+        self.updated_at_unix_nanos = OffsetDateTime::now_utc().unix_timestamp_nanos();
+        self.mark_terminal_if_worker_drained();
+        true
+    }
+
    pub fn mark_cancel_requested(&mut self) {
        self.cancel_requested = true;
        self.updated_at_unix_nanos = OffsetDateTime::now_utc().unix_timestamp_nanos();
@@ -363,12 +408,116 @@ impl ManualTransitionJobRecord {
    }
 }

-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
 pub enum ManualTransitionWorkerResult {
    Completed,
    TierFailure,
 }

+#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
+pub struct ManualTransitionWorkerResultStats {
+    pub completed: u64,
+    pub failed: u64,
+}
+
+impl ManualTransitionWorkerResultStats {
+    fn record(&mut self, result: ManualTransitionWorkerResult) {
+        match result {
+            ManualTransitionWorkerResult::Completed => self.completed = self.completed.saturating_add(1),
+            ManualTransitionWorkerResult::TierFailure => self.failed = self.failed.saturating_add(1),
+        }
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct ManualTransitionWorkerResultRecord {
+    pub schema: String,
+    pub job_id: Uuid,
+    pub task_key: String,
+    pub result: ManualTransitionWorkerResult,
+    pub completed_at_unix_nanos: i128,
+}
+
+impl ManualTransitionWorkerResultRecord {
+    pub fn new(job_id: Uuid, task_key: impl Into<String>, result: ManualTransitionWorkerResult) -> Self {
+        Self {
+            schema: MANUAL_TRANSITION_WORKER_RESULT_SCHEMA.to_string(),
+            job_id,
+            task_key: task_key.into(),
+            result,
+            completed_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 = PersistedManualTransitionWorkerResultRecord {
+            schema: MANUAL_TRANSITION_WORKER_RESULT_SCHEMA.to_string(),
+            content_sha256,
+            record: self.clone(),
+        };
+        let encoded = serde_json::to_vec(&persisted)?;
+        if encoded.len() > MAX_MANUAL_TRANSITION_WORKER_RESULT_RECORD_SIZE {
+            return Err(ManualTransitionJobError::Corrupt("encoded worker result 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_WORKER_RESULT_RECORD_SIZE {
+            return Err(ManualTransitionJobError::Corrupt("encoded worker result exceeds maximum size"));
+        }
+        let persisted: PersistedManualTransitionWorkerResultRecord = serde_json::from_slice(data)?;
+        if persisted.schema != MANUAL_TRANSITION_WORKER_RESULT_SCHEMA {
+            return Err(ManualTransitionJobError::UnsupportedSchema(persisted.schema));
+        }
+        if !is_sha256_checksum(&persisted.content_sha256) {
+            return Err(ManualTransitionJobError::Corrupt(
+                "worker result 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("worker result job_id does not match record key"));
+        }
+        if record.task_key != expected_task_key {
+            return Err(ManualTransitionJobError::Corrupt("worker result task_key does not match record key"));
+        }
+        record.validate()?;
+        Ok(record)
+    }
+
+    fn validate(&self) -> Result<(), ManualTransitionJobError> {
+        if self.schema != MANUAL_TRANSITION_WORKER_RESULT_SCHEMA {
+            return Err(ManualTransitionJobError::UnsupportedSchema(self.schema.clone()));
+        }
+        if self.job_id.is_nil() {
+            return Err(ManualTransitionJobError::Corrupt("worker result job_id is nil"));
+        }
+        if !is_sha256_checksum(&self.task_key) {
+            return Err(ManualTransitionJobError::Corrupt("worker result task_key is not a sha256 checksum"));
+        }
+        Ok(())
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+struct PersistedManualTransitionWorkerResultRecord {
+    schema: String,
+    content_sha256: String,
+    record: ManualTransitionWorkerResultRecord,
+}
+
 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))
 }
@@ -485,6 +634,59 @@ pub fn manual_transition_job_record_object_name(job_id: Uuid) -> Result<String,
    ))
 }

+fn manual_transition_job_sharded_prefix(prefix: &str, job_id: Uuid) -> Result<String, ManualTransitionJobError> {
+    if job_id.is_nil() {
+        return Err(ManualTransitionJobError::Corrupt("job_id is nil"));
+    }
+    let job_key = job_id.simple().to_string();
+    Ok(format!("{}/{}/{}/{}", prefix, &job_key[..2], &job_key[2..4], job_key))
+}
+
+pub fn manual_transition_worker_result_task_key(bucket: &str, object: &str, version_id: Option<Uuid>) -> String {
+    let version = version_id.map(|version| version.to_string()).unwrap_or_default();
+    let mut material = Vec::with_capacity(bucket.len() + object.len() + version.len() + 32);
+    push_len_prefixed(&mut material, bucket.as_bytes());
+    push_len_prefixed(&mut material, object.as_bytes());
+    push_len_prefixed(&mut material, version.as_bytes());
+    hex_sha256(&material, ToOwned::to_owned)
+}
+
+fn push_len_prefixed(out: &mut Vec<u8>, value: &[u8]) {
+    out.extend_from_slice(value.len().to_string().as_bytes());
+    out.push(b':');
+    out.extend_from_slice(value);
+}
+
+pub fn manual_transition_worker_result_object_prefix(job_id: Uuid) -> Result<String, ManualTransitionJobError> {
+    manual_transition_job_sharded_prefix(MANUAL_TRANSITION_WORKER_RESULT_PREFIX, job_id)
+}
+
+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"));
+    }
+    Ok(format!("{}/{}.json", manual_transition_worker_result_object_prefix(job_id)?, task_key))
+}
+
+fn manual_transition_worker_result_task_key_from_object_name(
+    job_id: Uuid,
+    object_name: &str,
+) -> Result<String, ManualTransitionJobError> {
+    let prefix = manual_transition_worker_result_object_prefix(job_id)?;
+    let rest = object_name
+        .strip_prefix(&prefix)
+        .ok_or(ManualTransitionJobError::Corrupt("worker result object prefix is invalid"))?
+        .strip_prefix('/')
+        .ok_or(ManualTransitionJobError::Corrupt("worker result object path is invalid"))?;
+    let task_key = rest
+        .strip_suffix(".json")
+        .ok_or(ManualTransitionJobError::Corrupt("worker result suffix is invalid"))?;
+    if task_key.contains('/') || !is_sha256_checksum(task_key) {
+        return Err(ManualTransitionJobError::Corrupt("worker result 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"));
@@ -582,6 +784,160 @@ pub async fn save_manual_transition_job_record_if_current(
    .await
 }

+pub(crate) async fn save_manual_transition_worker_result_if_absent(
+    api: Arc<ECStore>,
+    record: &ManualTransitionWorkerResultRecord,
+) -> EcstoreResult<bool> {
+    let object = manual_transition_worker_result_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_worker_result_stats(
+    api: Arc<ECStore>,
+    job_id: Uuid,
+) -> EcstoreResult<ManualTransitionWorkerResultStats> {
+    match scan_manual_transition_worker_result_journal(api, job_id).await? {
+        ManualTransitionWorkerResultJournal::Stats(stats) => Ok(stats),
+        ManualTransitionWorkerResultJournal::Corrupt(error) => Err(Error::other(error)),
+    }
+}
+
+enum ManualTransitionWorkerResultJournal {
+    Stats(ManualTransitionWorkerResultStats),
+    Corrupt(String),
+}
+
+async fn scan_manual_transition_worker_result_journal(
+    api: Arc<ECStore>,
+    job_id: Uuid,
+) -> EcstoreResult<ManualTransitionWorkerResultJournal> {
+    let prefix = manual_transition_worker_result_object_prefix(job_id).map_err(manual_transition_job_store_error)?;
+    let mut marker = None;
+    let mut stats = ManualTransitionWorkerResultStats::default();
+    loop {
+        let page = api
+            .clone()
+            .list_objects_v2(
+                RUSTFS_META_BUCKET,
+                &prefix,
+                marker,
+                None,
+                MANUAL_TRANSITION_WORKER_RESULT_SCAN_LIMIT,
+                false,
+                None,
+                false,
+            )
+            .await?;
+        for object in page.objects {
+            let task_key = match manual_transition_worker_result_task_key_from_object_name(job_id, &object.name) {
+                Ok(task_key) => task_key,
+                Err(err) => return Ok(ManualTransitionWorkerResultJournal::Corrupt(err.to_string())),
+            };
+            let object_name =
+                manual_transition_worker_result_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),
+            };
+            let result = match ManualTransitionWorkerResultRecord::decode(job_id, &task_key, &data) {
+                Ok(result) => result,
+                Err(err) => return Ok(ManualTransitionWorkerResultJournal::Corrupt(err.to_string())),
+            };
+            stats.record(result.result);
+        }
+        if !page.is_truncated {
+            return Ok(ManualTransitionWorkerResultJournal::Stats(stats));
+        }
+        let Some(next_marker) = page.next_continuation_token else {
+            return Err(Error::other("manual transition worker result page is truncated without a next marker"));
+        };
+        marker = Some(next_marker);
+    }
+}
+
+pub async fn reconcile_manual_transition_worker_results(
+    api: Arc<ECStore>,
+    job_id: Uuid,
+    queue_snapshot: ManualTransitionQueueSnapshot,
+) -> EcstoreResult<ManualTransitionJobRecord> {
+    let stats = match scan_manual_transition_worker_result_journal(api.clone(), job_id).await? {
+        ManualTransitionWorkerResultJournal::Stats(stats) => stats,
+        ManualTransitionWorkerResultJournal::Corrupt(error) => {
+            return mark_manual_transition_job_unknown_for_worker_result_journal_error(api, job_id, error, queue_snapshot).await;
+        }
+    };
+    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);
+        if !changed {
+            return Ok(record);
+        }
+        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)
+}
+
+async fn mark_manual_transition_job_unknown_for_worker_result_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_worker_result_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)
+}
+
 pub async fn save_manual_transition_scope_admission_if_absent(
    api: Arc<ECStore>,
    admission: &ManualTransitionScopeAdmission,
@@ -793,9 +1149,15 @@ pub async fn persist_manual_transition_job_progress(
 pub async fn record_manual_transition_worker_result(
    api: Arc<ECStore>,
    job_id: Uuid,
+    task_key: &str,
    result: ManualTransitionWorkerResult,
    queue_snapshot: ManualTransitionQueueSnapshot,
 ) -> EcstoreResult<ManualTransitionJobRecord> {
+    let result_record = ManualTransitionWorkerResultRecord::new(job_id, task_key, result);
+    if !save_manual_transition_worker_result_if_absent(api.clone(), &result_record).await? {
+        return load_manual_transition_job_record(api, job_id).await;
+    }
+
    for _ in 0..4 {
        let (mut record, etag) = load_manual_transition_job_record_with_etag(api.clone(), job_id).await?;
        if record.is_terminal() {
@@ -829,8 +1191,19 @@ pub async fn renew_manual_transition_job_lease(
    job_id: Uuid,
    queue_snapshot: ManualTransitionQueueSnapshot,
 ) -> EcstoreResult<ManualTransitionJobRecord> {
-    let (mut record, etag) = load_manual_transition_job_record_with_etag(api.clone(), job_id).await?;
+    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
+        {
+            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);
+            }
+            (record, etag) = load_manual_transition_job_record_with_etag(api.clone(), job_id).await?;
+        }
        let became_terminal = record.mark_unknown_if_worker_results_lost(queue_snapshot);
        if !became_terminal {
            record.renew_lease(queue_snapshot);
@@ -1100,6 +1473,33 @@ mod tests {
        assert!(record.completed_at_unix_nanos.is_some());
    }

+    #[test]
+    fn manual_transition_job_marks_unknown_for_corrupt_worker_result_journal() {
+        let options = ManualTransitionRunOptions::default();
+        let mut record = ManualTransitionJobRecord::new(Uuid::new_v4(), "bucket", &options, TEST_OWNER);
+
+        assert!(!record.mark_unknown_for_worker_result_journal_error("bad marker", ManualTransitionQueueSnapshot::default()));
+
+        record.complete(
+            ManualTransitionRunReport {
+                bucket: "bucket".to_string(),
+                enqueued: 1,
+                ..Default::default()
+            },
+            ManualTransitionQueueSnapshot::default(),
+        );
+        assert!(record.mark_unknown_for_worker_result_journal_error("bad marker", ManualTransitionQueueSnapshot::default()));
+
+        assert_eq!(record.state, ManualTransitionJobState::Unknown);
+        assert!(
+            record
+                .error
+                .as_deref()
+                .is_some_and(|error| error.contains("worker result journal is corrupt"))
+        );
+        assert!(record.completed_at_unix_nanos.is_some());
+    }
+
    #[test]
    fn manual_transition_job_record_builds_resume_options() {
        let options = ManualTransitionRunOptions {
@@ -1166,6 +1566,22 @@ mod tests {
        assert!(matches!(err, ManualTransitionJobError::ChecksumMismatch));
    }

+    #[test]
+    fn manual_transition_worker_result_record_rejects_checksum_drift() {
+        let job_id = Uuid::new_v4();
+        let task_key = manual_transition_worker_result_task_key("bucket", "logs/a", None);
+        let record = ManualTransitionWorkerResultRecord::new(job_id, &task_key, ManualTransitionWorkerResult::Completed);
+        let encoded = record.encode().expect("worker result record should encode");
+        let mut value: serde_json::Value = serde_json::from_slice(&encoded).expect("encoded worker result should be json");
+        value["record"]["result"] = serde_json::Value::String("tier_failure".to_string());
+        let mutated = serde_json::to_vec(&value).expect("mutated worker result should encode");
+
+        let err = ManualTransitionWorkerResultRecord::decode(job_id, &task_key, &mutated)
+            .expect_err("worker result checksum drift must fail closed");
+
+        assert!(matches!(err, ManualTransitionJobError::ChecksumMismatch));
+    }
+
    #[test]
    fn manual_transition_job_record_rejects_unknown_report_fields() {
        let options = ManualTransitionRunOptions::default();