fix(ecstore): replay manual transition task journal (#5329)

Recover accepted manual-transition tasks from the durable task journal when worker-result markers are missing after a restart.

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

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

@@ -25,7 +25,7 @@ use crate::bucket::lifecycle::manual_transition_job::{
     MANUAL_TRANSITION_JOB_RECORD_PREFIX, ManualTransitionJobRecord, ManualTransitionJobState, ManualTransitionScopeAdmission,
     ManualTransitionScopeAdmissionClaim, ManualTransitionTaskRecord, ManualTransitionWorkerFailureReason,
     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,
+    load_manual_transition_job_record, load_manual_transition_job_record_with_etag, load_manual_transition_pending_task_records,
     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, record_manual_transition_worker_result_with_reason,
@@ -1167,6 +1167,13 @@ impl TransitionEnqueueOutcome {
     }
 }
 
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum ManualTransitionPendingTaskReplay {
+    Empty,
+    Queued,
+    Deferred,
+}
+
 impl TransitionState {
     #[allow(clippy::new_ret_no_self)]
     pub fn new() -> Arc<Self> {
@@ -1453,7 +1460,8 @@ impl TransitionState {
                 oi.name.clone(),
                 oi.version_id,
                 event.storage_class.clone(),
-            );
+            )
+            .with_object_metadata(oi.etag.clone(), oi.mod_time, oi.size, oi.is_latest);
             if let Err(err) = save_manual_transition_task_if_absent(api, &task_record).await {
                 self.release_transition(oi);
                 warn!(
@@ -2103,19 +2111,6 @@ async fn recover_manual_transition_job(
         }
     }
 
-    if record.mark_unknown_if_worker_results_lost(recovery_unknown_snapshot)
-        || record.mark_unknown_if_recovery_would_skip_pending_page(recovery_unknown_snapshot)
-    {
-        return match save_manual_transition_job_record_if_current(api.clone(), &record, &etag).await {
-            Ok(()) => {
-                release_manual_transition_recovery_admission(api, &record).await;
-                Ok(ManualTransitionJobRecoveryOutcome::Unknown)
-            }
-            Err(Error::PreconditionFailed) => Ok(ManualTransitionJobRecoveryOutcome::Skipped),
-            Err(err) => Err(err),
-        };
-    }
-
     if record.cancel_requested {
         record.cancel_after_recovery(queue_snapshot);
         return match save_manual_transition_job_record_if_current(api.clone(), &record, &etag).await {
@@ -2128,6 +2123,7 @@ async fn recover_manual_transition_job(
         };
     }
 
+    let previous_lease_id = record.lease_id;
     record.claim_recovery_lease(manual_transition_recovery_owner_id(), queue_snapshot);
     let recovery_lease_id = record.lease_id;
     match save_manual_transition_job_record_if_current(api.clone(), &record, &etag).await {
@@ -2135,6 +2131,7 @@ async fn recover_manual_transition_job(
         Err(Error::PreconditionFailed) => return Ok(ManualTransitionJobRecoveryOutcome::Skipped),
         Err(err) => return Err(err),
     }
+    delete_manual_transition_scope_admission_if_current(api.clone(), &record.scope_key, record.job_id, previous_lease_id).await?;
 
     match claim_manual_transition_scope_admission(api.clone(), &ManualTransitionScopeAdmission::from_job(&record)).await {
         Ok(ManualTransitionScopeAdmissionClaim::Claimed) => {}
@@ -2148,6 +2145,29 @@ async fn recover_manual_transition_job(
         }
     }
 
+    let replay = replay_manual_transition_pending_tasks(api.clone(), job_id).await?;
+    if matches!(
+        replay,
+        ManualTransitionPendingTaskReplay::Queued | ManualTransitionPendingTaskReplay::Deferred
+    ) {
+        spawn_manual_transition_recovery_heartbeat(api, job_id);
+        return Ok(ManualTransitionJobRecoveryOutcome::Resumed);
+    }
+
+    let (mut 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)
+    {
+        return match save_manual_transition_job_record_if_current(api.clone(), &record, &etag).await {
+            Ok(()) => {
+                release_manual_transition_recovery_admission(api, &record).await;
+                Ok(ManualTransitionJobRecoveryOutcome::Unknown)
+            }
+            Err(Error::PreconditionFailed) => Ok(ManualTransitionJobRecoveryOutcome::Skipped),
+            Err(err) => Err(err),
+        };
+    }
+
     let mut options = record.resume_options();
     options.job_id = Some(job_id);
     options.cancel_check = Some(manual_transition_recovery_cancel_check(api.clone(), job_id));
@@ -2162,6 +2182,69 @@ async fn recover_manual_transition_job(
     Ok(ManualTransitionJobRecoveryOutcome::Resumed)
 }
 
+async fn replay_manual_transition_pending_tasks(
+    api: Arc<ECStore>,
+    job_id: Uuid,
+) -> Result<ManualTransitionPendingTaskReplay, Error> {
+    let transition_state = runtime_sources::transition_state_handle();
+    let limit = transition_state.transition_queue_capacity.max(1);
+    let pending = load_manual_transition_pending_task_records(api.clone(), job_id, limit).await?;
+    if pending.is_empty() {
+        return Ok(ManualTransitionPendingTaskReplay::Empty);
+    }
+
+    let mut queued = 0usize;
+    for task in pending {
+        let mod_time = match task.mod_time_unix_nanos {
+            Some(nanos) => Some(
+                OffsetDateTime::from_unix_timestamp_nanos(nanos)
+                    .map_err(|_| Error::other("manual transition task journal mod_time is invalid"))?,
+            ),
+            None => None,
+        };
+        let object = ObjectInfo {
+            bucket: task.bucket,
+            name: task.object,
+            version_id: task.version_id,
+            etag: task.etag,
+            mod_time,
+            size: task.size.map_or(0, |size| size),
+            is_latest: task.is_latest.unwrap_or(false),
+            ..Default::default()
+        };
+        let event = lifecycle::Event {
+            action: if object.version_id.is_some() {
+                IlmAction::TransitionVersionAction
+            } else {
+                IlmAction::TransitionAction
+            },
+            storage_class: task.storage_class,
+            ..Default::default()
+        };
+
+        match transition_state
+            .queue_transition_task_outcome(Some(api.clone()), &object, &event, &LcEventSrc::Scanner, Some(job_id))
+            .await
+        {
+            TransitionEnqueueOutcome::Queued | TransitionEnqueueOutcome::AlreadyInFlight => {
+                queued = queued.saturating_add(1);
+            }
+            TransitionEnqueueOutcome::QueueFull
+            | TransitionEnqueueOutcome::QueueClosed
+            | TransitionEnqueueOutcome::QueueSendTimedOut
+            | TransitionEnqueueOutcome::TaskJournalFailed => {
+                break;
+            }
+        }
+    }
+
+    if queued > 0 {
+        Ok(ManualTransitionPendingTaskReplay::Queued)
+    } else {
+        Ok(ManualTransitionPendingTaskReplay::Deferred)
+    }
+}
+
 fn manual_transition_recovery_owner_id() -> &'static str {
     "ecstore-manual-transition-recovery"
 }
@@ -7015,6 +7098,10 @@ mod tests {
             bucket: "manual-task-journal-bucket".to_string(),
             name: "logs/object".to_string(),
             version_id: Some(version_id),
+            etag: Some("task-etag".to_string()),
+            mod_time: Some(OffsetDateTime::now_utc()),
+            size: 42,
+            is_latest: true,
             ..Default::default()
         };
         let event = crate::bucket::lifecycle::lifecycle::Event {
@@ -7039,6 +7126,10 @@ mod tests {
         assert_eq!(task_record.object, object.name);
         assert_eq!(task_record.version_id, Some(version_id));
         assert_eq!(task_record.storage_class, "WARM");
+        assert_eq!(task_record.etag.as_deref(), Some("task-etag"));
+        assert_eq!(task_record.mod_time_unix_nanos, object.mod_time.map(|time| time.unix_timestamp_nanos()));
+        assert_eq!(task_record.size, Some(42));
+        assert_eq!(task_record.is_latest, Some(true));
     }
 
     #[tokio::test]
@@ -8327,7 +8418,7 @@ mod tests {
 
     #[tokio::test]
     #[serial]
-    async fn manual_transition_recovery_marks_unknown_before_cancel_for_cursor_pending_work() {
+    async fn manual_transition_recovery_cancels_cursor_pending_work_when_requested() {
         let (_paths, ecstore) = setup_test_env().await;
         let job_id = Uuid::new_v4();
         let continuation_token =
@@ -8353,20 +8444,15 @@ mod tests {
 
         let outcome = recover_manual_transition_job(ecstore.clone(), job_id, ManualTransitionQueueSnapshot::default())
             .await
-            .expect("manual transition recovery should fail closed before cancelled pending cursor jobs");
+            .expect("manual transition recovery should cancel requested pending cursor jobs");
 
-        assert_eq!(outcome, ManualTransitionJobRecoveryOutcome::Unknown);
+        assert_eq!(outcome, ManualTransitionJobRecoveryOutcome::Cancelled);
         let recovered = load_manual_transition_job_record(ecstore.clone(), job_id)
             .await
-            .expect("unknown cancelled job should load");
+            .expect("cancelled job should load");
-        assert_eq!(recovered.state, ManualTransitionJobState::Unknown);
+        assert_eq!(recovered.state, ManualTransitionJobState::Cancelled);
         assert!(recovered.cancel_requested);
-        assert!(
+        assert!(recovered.error.is_none());
-            recovered
-                .error
-                .as_deref()
-                .is_some_and(|error| error.contains("page/task journal"))
-        );
     }
 
     #[tokio::test]
@@ -8892,8 +8978,13 @@ mod tests {
 
     #[tokio::test]
     #[serial]
-    async fn manual_transition_recovery_marks_unknown_for_task_journal_without_result() {
+    async fn manual_transition_recovery_replays_task_journal_without_result() {
         let (_paths, ecstore) = setup_test_env().await;
+        let transition_state = runtime_sources::transition_state_handle();
+        let original_workers = transition_state.num_workers.load(Ordering::SeqCst);
+        let absolute_max = resolve_transition_workers_absolute_max();
+        TransitionState::resize_workers_to(ecstore.clone(), 0, 0, absolute_max);
+        let pending_before = transition_state.pending_tasks();
         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");
@@ -8916,29 +9007,27 @@ mod tests {
 
         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");
+            .expect("recovery should replay a task journal marker with no worker result");
 
-        assert_eq!(outcome, ManualTransitionJobRecoveryOutcome::Unknown);
+        assert_eq!(outcome, ManualTransitionJobRecoveryOutcome::Resumed);
         let recovered = load_manual_transition_job_record(ecstore.clone(), job_id)
             .await
-            .expect("unknown task journal job should load");
+            .expect("replayed task journal job should load");
-        assert_eq!(recovered.state, ManualTransitionJobState::Unknown);
+        assert_eq!(recovered.state, ManualTransitionJobState::Running);
         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,
+                load_manual_transition_scope_admission(ecstore.clone(), &record.scope_key).await,
-                Err(Error::ConfigNotFound)
+                Ok(admission) if admission.job_id == job_id
             ),
-            "unknown task journal recovery must release the scope admission"
+            "replayed task journal recovery must keep the scope admission until worker results arrive"
         );
+        if transition_state.pending_tasks() > pending_before {
+            let _ = transition_state.transition_rx.try_recv();
+        }
+        TransitionState::resize_workers_to(ecstore, original_workers, original_workers, absolute_max);
     }
 
     #[tokio::test]

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

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::collections::BTreeMap;
+use std::collections::{BTreeMap, BTreeSet};
 use std::sync::Arc;
 
 use rustfs_utils::crypto::{hex_sha256, is_sha256_checksum};
@@ -505,6 +505,14 @@ pub struct ManualTransitionTaskRecord {
     pub object: String,
     pub version_id: Option<Uuid>,
     pub storage_class: String,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub etag: Option<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub mod_time_unix_nanos: Option<i128>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub size: Option<i64>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub is_latest: Option<bool>,
     pub queued_at_unix_nanos: i128,
 }
 
@@ -525,10 +533,28 @@ impl ManualTransitionTaskRecord {
             object: object.into(),
             version_id,
             storage_class: storage_class.into(),
+            etag: None,
+            mod_time_unix_nanos: None,
+            size: None,
+            is_latest: None,
             queued_at_unix_nanos: OffsetDateTime::now_utc().unix_timestamp_nanos(),
         }
     }
 
+    pub fn with_object_metadata(
+        mut self,
+        etag: Option<String>,
+        mod_time: Option<OffsetDateTime>,
+        size: i64,
+        is_latest: bool,
+    ) -> Self {
+        self.etag = etag;
+        self.mod_time_unix_nanos = mod_time.map(|time| time.unix_timestamp_nanos());
+        self.size = Some(size);
+        self.is_latest = Some(is_latest);
+        self
+    }
+
     pub fn encode(&self) -> Result<Vec<u8>, ManualTransitionJobError> {
         self.validate()?;
         let record_bytes = serde_json::to_vec(self)?;
@@ -588,6 +614,9 @@ impl ManualTransitionTaskRecord {
         if self.storage_class.trim().is_empty() {
             return Err(ManualTransitionJobError::Corrupt("task record storage_class is empty"));
         }
+        if self.size.is_some_and(|size| size < 0) {
+            return Err(ManualTransitionJobError::Corrupt("task record size is negative"));
+        }
         Ok(())
     }
 }
@@ -1151,23 +1180,95 @@ pub async fn load_manual_transition_worker_result_stats(
     job_id: Uuid,
 ) -> EcstoreResult<ManualTransitionWorkerResultStats> {
     match scan_manual_transition_worker_result_journal(api, job_id).await? {
-        ManualTransitionWorkerResultJournal::Stats(stats) => Ok(stats),
+        ManualTransitionWorkerResultJournal::Stats(stats) => Ok(stats.stats),
         ManualTransitionWorkerResultJournal::Corrupt(error) => Err(Error::other(error)),
     }
 }
 
 enum ManualTransitionWorkerResultJournal {
-    Stats(ManualTransitionWorkerResultStats),
+    Stats(ManualTransitionWorkerResultJournalStats),
     Corrupt(String),
 }
 
+#[derive(Debug, Clone, Default, PartialEq, Eq)]
+struct ManualTransitionWorkerResultJournalStats {
+    stats: ManualTransitionWorkerResultStats,
+    task_keys: BTreeSet<String>,
+}
+
+impl ManualTransitionWorkerResultJournalStats {
+    fn record(&mut self, result: ManualTransitionWorkerResultRecord) {
+        self.task_keys.insert(result.task_key.clone());
+        self.stats.record(result.result, result.failure_reason);
+    }
+}
+
+pub async fn load_manual_transition_pending_task_records(
+    api: Arc<ECStore>,
+    job_id: Uuid,
+    limit: usize,
+) -> EcstoreResult<Vec<ManualTransitionTaskRecord>> {
+    if limit == 0 {
+        return Ok(Vec::new());
+    }
+
+    let result_keys = match scan_manual_transition_worker_result_journal(api.clone(), job_id).await? {
+        ManualTransitionWorkerResultJournal::Stats(stats) => stats.task_keys,
+        ManualTransitionWorkerResultJournal::Corrupt(error) => return Err(Error::other(error)),
+    };
+    let prefix = manual_transition_task_object_prefix(job_id).map_err(manual_transition_job_store_error)?;
+    let mut marker = None;
+    let scan_limit = usize::try_from(MANUAL_TRANSITION_TASK_SCAN_LIMIT)
+        .map_err(|_| Error::other("manual transition task scan limit is invalid"))?;
+    let mut pending = Vec::with_capacity(limit.min(scan_limit));
+
+    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 =
+                manual_transition_task_key_from_object_name(job_id, &object.name).map_err(manual_transition_job_store_error)?;
+            if result_keys.contains(&task_key) {
+                continue;
+            }
+            let object_name = manual_transition_task_object_name(job_id, &task_key).map_err(manual_transition_job_store_error)?;
+            let data = config_boundary::read_config(api.clone(), &object_name).await?;
+            let task = ManualTransitionTaskRecord::decode(job_id, &task_key, &data).map_err(manual_transition_job_store_error)?;
+            pending.push(task);
+            if pending.len() == limit {
+                return Ok(pending);
+            }
+        }
+
+        if !page.is_truncated {
+            return Ok(pending);
+        }
+        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);
+    }
+}
+
 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();
+    let mut stats = ManualTransitionWorkerResultJournalStats::default();
     loop {
         let page = api
             .clone()
@@ -1197,7 +1298,7 @@ async fn scan_manual_transition_worker_result_journal(
                 Ok(result) => result,
                 Err(err) => return Ok(ManualTransitionWorkerResultJournal::Corrupt(err.to_string())),
             };
-            stats.record(result.result, result.failure_reason);
+            stats.record(result);
         }
         if !page.is_truncated {
             return Ok(ManualTransitionWorkerResultJournal::Stats(stats));
@@ -1229,9 +1330,9 @@ 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.stats.completed,
-            stats.failed,
+            stats.stats.failed,
-            &stats.tier_failure_by_reason,
+            &stats.stats.tier_failure_by_reason,
             task_stats.queued,
             queue_snapshot,
         );