diff --git a/crates/ecstore/src/bucket/lifecycle/durable_namespace.rs b/crates/ecstore/src/bucket/lifecycle/durable_namespace.rs index 606b5f536..2383196bc 100644 --- a/crates/ecstore/src/bucket/lifecycle/durable_namespace.rs +++ b/crates/ecstore/src/bucket/lifecycle/durable_namespace.rs @@ -388,6 +388,7 @@ impl ManualTransitionJobProgressProof { queue_snapshot: &ManualTransitionQueueSnapshot, cursor_revision: Option, ) -> Result { + 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 { 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" ); diff --git a/crates/ecstore/src/bucket/lifecycle/manual_transition_job.rs b/crates/ecstore/src/bucket/lifecycle/manual_transition_job.rs index 36a0120d5..b480ba468 100644 --- a/crates/ecstore/src/bucket/lifecycle/manual_transition_job.rs +++ b/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 { +pub(super) fn manual_transition_cursor_revision(report: &ManualTransitionRunReport) -> Option { 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, diff --git a/crates/ecstore/tests/manual_transition_job_matrix_test.rs b/crates/ecstore/tests/manual_transition_job_matrix_test.rs index 84c8e9f92..e631c2afb 100644 --- a/crates/ecstore/tests/manual_transition_job_matrix_test.rs +++ b/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());