fix(ecstore): enforce monotonic transition cursors (#6522)

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

@@ -388,6 +388,7 @@ impl ManualTransitionJobProgressProof {
         queue_snapshot: &ManualTransitionQueueSnapshot,
         cursor_revision: Option<u64>,
     ) -> Result<Self> {
+        let cursor_revision = cursor_revision.or_else(|| manual_transition_job::manual_transition_cursor_revision(report));
         let cursor_marker = match report.continuation_token.as_deref() {
             Some(token) => {
                 let marker = decode_manual_transition_continuation_token(token)?
@@ -605,6 +606,11 @@ fn manual_job_cursor_reaches(
     if previous.continuation_token_sha256 == next.continuation_token_sha256 {
         return previous.cursor_marker == next.cursor_marker && previous.cursor_revision == next.cursor_revision;
     }
+    if let (Some(previous_marker), Some(next_marker)) = (&previous.cursor_marker, &next.cursor_marker)
+        && previous_marker != next_marker
+    {
+        return next.scanned > previous.scanned && next_marker > previous_marker;
+    }
     match (&previous.continuation_token_sha256, &next.continuation_token_sha256) {
         (None, Some(_)) => {
             next.scanned > previous.scanned
@@ -911,7 +917,7 @@ mod tests {
     fn try_manual_job_checkpoint(job: &manual_transition_job::ManualTransitionJobRecord) -> Result<DurableIlmRecordCheckpoint> {
         let path =
             manual_transition_job::manual_transition_job_record_object_name(job.job_id).expect("manual job path should build");
-        let encoded = job.encode().expect("manual job should encode");
+        let encoded = job.encode().map_err(|err| Error::other(err.to_string()))?;
         Ok(validate_durable_ilm_record(&path, &encoded)?.checkpoint)
     }
 
@@ -955,10 +961,10 @@ mod tests {
         let options = super::super::bucket_lifecycle_ops::ManualTransitionRunOptions::default();
         let mut job =
             manual_transition_job::ManualTransitionJobRecord::new(Uuid::new_v4(), "legacy-checkpoint-bucket", &options, "owner");
-        job.cursor_revision = None;
-        job.updated_at_unix_nanos += 1;
-        job.report.scanned = 1;
-        job.report.continuation_token = Some(continuation_token("logs/a"));
+        let mut report = job.report.clone();
+        report.scanned = 1;
+        report.continuation_token = Some(continuation_token("logs/a"));
+        job.update_running_progress(report, ManualTransitionQueueSnapshot::default());
         let compact = manual_job_checkpoint(&job);
         let mut legacy = compact.clone();
         let DurableIlmRecordCheckpoint::ManualTransitionJob {
@@ -1090,8 +1096,8 @@ mod tests {
         counter_rollback.updated_at_unix_nanos += 1;
         counter_rollback.report.scanned = 9;
         assert!(
-            previous_checkpoint
-                .validate_successor(&manual_job_checkpoint(&counter_rollback))
+            try_manual_job_checkpoint(&counter_rollback)
+                .and_then(|checkpoint| previous_checkpoint.validate_successor(&checkpoint))
                 .is_err()
         );
 
@@ -1100,8 +1106,8 @@ mod tests {
         cursor_rollback.report.scanned += 1;
         cursor_rollback.report.continuation_token = Some(continuation_token("logs/a"));
         assert!(
-            previous_checkpoint
-                .validate_successor(&manual_job_checkpoint(&cursor_rollback))
+            try_manual_job_checkpoint(&cursor_rollback)
+                .and_then(|checkpoint| previous_checkpoint.validate_successor(&checkpoint))
                 .is_err()
         );
 
@@ -1125,8 +1131,8 @@ mod tests {
         same_marker_version_rollback.report.continuation_token =
             Some(continuation_token_with_version("logs/b", Some("opaque-arbitrary-version")));
         assert!(
-            same_marker_version_previous_checkpoint
-                .validate_successor(&manual_job_checkpoint(&same_marker_version_rollback))
+            try_manual_job_checkpoint(&same_marker_version_rollback)
+                .and_then(|checkpoint| same_marker_version_previous_checkpoint.validate_successor(&checkpoint))
                 .is_err(),
             "a different opaque version marker without producer evidence must fail closed"
         );

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

@@ -588,6 +588,9 @@ impl ManualTransitionJobRecord {
         if self.state == ManualTransitionJobState::Cancelled && !self.cancel_requested {
             return Err(ManualTransitionJobError::Corrupt("cancelled job is missing cancel request"));
         }
+        if self.cursor_revision != manual_transition_cursor_revision(&self.report) {
+            return Err(ManualTransitionJobError::Corrupt("cursor revision does not match report"));
+        }
         Ok(())
     }
 }
@@ -1996,7 +1999,7 @@ fn manual_transition_job_store_error(err: ManualTransitionJobError) -> Error {
     Error::other(err)
 }
 
-fn manual_transition_cursor_revision(report: &ManualTransitionRunReport) -> Option<u64> {
+pub(super) fn manual_transition_cursor_revision(report: &ManualTransitionRunReport) -> Option<u64> {
     report.continuation_token.as_ref()?;
     (report.scanned > 0).then_some(report.scanned)
 }
@@ -2054,6 +2057,18 @@ mod tests {
         assert_eq!(decoded.max_objects, Some(17));
     }
 
+    #[test]
+    fn manual_transition_job_record_rejects_untracked_cursor_revision() {
+        let options = ManualTransitionRunOptions::default();
+        let mut record = ManualTransitionJobRecord::new(Uuid::new_v4(), "bucket", &options, TEST_OWNER);
+        record.report.scanned = 1;
+        record.report.continuation_token = encode_manual_transition_continuation_token(Some("logs/page-a".to_string()), None);
+
+        let err = record.encode().expect_err("untracked cursor progress must fail closed");
+
+        assert!(matches!(err, ManualTransitionJobError::Corrupt("cursor revision does not match report")));
+    }
+
     #[test]
     fn manual_transition_job_record_waits_for_worker_results() {
         let options = ManualTransitionRunOptions::default();
@@ -2734,6 +2749,7 @@ mod tests {
             lease_id: Uuid,
             lease_expires_at_unix_nanos: i128,
             state: ManualTransitionJobState,
+            #[serde(default)]
             scan_completed: bool,
             cancel_requested: bool,
             created_at_unix_nanos: i128,

crates/ecstore/tests/manual_transition_job_matrix_test.rs

@@ -88,17 +88,20 @@ fn manual_transition_record_marks_unknown_when_cursor_would_skip_pending_page()
     };
     let job_id = Uuid::new_v4();
     let mut record = ManualTransitionJobRecord::new(job_id, "manual-pending-page-bucket", &options, "owner-a");
-    record.report = ManualTransitionRunReport {
-        bucket: "manual-pending-page-bucket".to_string(),
-        prefix: options.prefix.clone(),
-        tier: options.tier,
-        scanned: 1000,
-        eligible: 2,
-        enqueued: 2,
-        transition_completed: 1,
-        continuation_token: Some("opaque-page-cursor".to_string()),
-        ..Default::default()
-    };
+    record.update_running_progress(
+        ManualTransitionRunReport {
+            bucket: "manual-pending-page-bucket".to_string(),
+            prefix: options.prefix.clone(),
+            tier: options.tier,
+            scanned: 1000,
+            eligible: 2,
+            enqueued: 2,
+            continuation_token: Some("opaque-page-cursor".to_string()),
+            ..Default::default()
+        },
+        ManualTransitionQueueSnapshot::default(),
+    );
+    record.report.transition_completed = 1;
 
     let marked = record.mark_unknown_if_recovery_would_skip_pending_page(ManualTransitionQueueSnapshot::default());