From fc88a78055970caae82717376188ac9f2802c48a Mon Sep 17 00:00:00 2001 From: Edouard Vanbelle Date: Mon, 7 Sep 2026 13:19:18 +0200 Subject: [PATCH] fix(consistency): a mid-batch pause resumes at the last settled hash MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `04807464` made deep-mode cancellation responsive but paused at the BATCH-START cursor, which throws away everything done in the current batch. Ed caught the sharp edge while testing pause: the checkpoint only lands after a batch completes, so a run paused 27 s into its FIRST 63 s batch had `cursor_hex: ""` and would resume from scratch. Later batches lose 500 blob reads, about a minute against remote S3. Now the pause carries `settled` — the highest hash whose pair was fully handled. That is safe because the merge-join advances both sides in ascending hash order: at any point in the loop, everything at or below `settled` has had its findings recorded and, under `?deep=true`, its bytes re-hashed. So resume re-does one pair, not the whole batch. `settled` only advances after an arm finishes with its item, never on entry, which is what keeps that invariant true. Also checkpoints explicitly before returning `Paused`, with `delta_count = 0` since `scanned_count` is already credited per batch. The engine writes the cursor on the Paused row anyway; persisting it here means a restart racing that write still resumes from the right place rather than the previous batch. Strictly fewer duplicate findings on resume, too. Page-level `unknown_backend_file` notices are emitted before the join, so any resume re-emits those for the re-walked range — a shorter range is simply less of it. That duplication is pre-existing and orthogonal. Co-Authored-By: Claude Opus 5 (1M context) --- .../services/backend_consistency_service.rs | 64 ++++++++++++++----- 1 file changed, 48 insertions(+), 16 deletions(-) diff --git a/src/infrastructure/services/backend_consistency_service.rs b/src/infrastructure/services/backend_consistency_service.rs index e039d508..6107e193 100644 --- a/src/infrastructure/services/backend_consistency_service.rs +++ b/src/infrastructure/services/backend_consistency_service.rs @@ -635,6 +635,11 @@ impl RecoverableJobHandler for BackendConsistencyCheck { let mut bi = page.blobs.iter().peekable(); let mut di = db_hashes.iter().peekable(); + // Highest hash fully handled in THIS batch — findings + // recorded, bytes verified if deep. A mid-batch pause resumes + // here, so it must only advance once an arm is finished with + // its item, never on entry. + let mut settled: Option = None; loop { match (bi.peek(), di.peek()) { // Present on both sides. Shallow: nothing to say — the @@ -648,37 +653,62 @@ impl RecoverableJobHandler for BackendConsistencyCheck { // verified then. (Some(b), Some(d)) if b.hash == **d => { if deep && in_range(&b.hash) { - // Cancel poll INSIDE the verify loop. The - // per-batch poll above bounds latency by one - // batch, which is ~63 s of remote reads in - // deep mode — a minute of an apparently - // ignored Cancel on a run that may last - // hours. Pausing here is safe and cheap: the - // cursor still points at the last completed - // batch, so resume re-verifies this batch's - // handful of blobs rather than skipping them. + // Cancel poll INSIDE the verify loop, because + // the per-batch poll bounds latency by one + // batch — ~63 s of remote reads in deep mode, + // a minute of an apparently ignored Cancel on + // a run that may last hours. + // + // Pauses at `settled`, the last pair fully + // handled, NOT at the batch's start cursor. + // The merge-join walks both sides in + // ascending hash order, so everything at or + // below `settled` has had its findings + // recorded and its bytes verified — resuming + // there re-does one pair, not five hundred. + // + // The batch-start cursor would have been + // correct but wasteful: in the FIRST batch it + // is empty, so a pause 27 s into a 63 s batch + // threw away the whole scan. if verified_count.is_multiple_of(DEEP_CANCEL_POLL_EVERY) && matches!(store.status().await, Ok(RunStatus::CancelRequested)) { + let resume_at = settled.clone().or_else(|| cursor.clone()); tracing::info!( target: "oxicloud::consistency", event = "backend_consistency.cancelled_mid_verify", run_id = %store.run_id(), verified = verified_count, - "deep verify cancelled mid-batch, pausing at the last checkpoint" + resume_at = resume_at.as_deref().unwrap_or(""), + "deep verify cancelled mid-batch, pausing at the last settled hash" ); - return RunOutcome::Paused { - cursor: cursor - .as_ref() - .map(|s| s.as_bytes().to_vec()) - .unwrap_or_default(), - }; + // Checkpoint so the cursor survives even + // if the engine's Paused write races a + // restart; `scanned_count` is already + // counted per batch, so add nothing here. + let bytes = resume_at + .as_ref() + .map(|s| s.as_bytes().to_vec()) + .unwrap_or_default(); + if let Err(e) = store.checkpoint(bytes.clone(), 0).await { + tracing::warn!( + target: "oxicloud::consistency", + event = "backend_consistency.pause_checkpoint_failed", + run_id = %store.run_id(), + error = %e, + "could not persist the mid-batch cursor; resume will \ + restart from the previous batch" + ); + } + return RunOutcome::Paused { cursor: bytes }; } verified_count += 1; finding_count += self .verify_bytes(store, verify_backend.as_ref(), &b.hash) .await; } + settled = Some(b.hash.clone()); bi.next(); di.next(); } @@ -707,6 +737,7 @@ impl RecoverableJobHandler for BackendConsistencyCheck { ) .await; } + settled = Some(b.hash.clone()); bi.next(); } // DB-only: a row whose bytes are gone. Severity is @@ -729,6 +760,7 @@ impl RecoverableJobHandler for BackendConsistencyCheck { }), ) .await; + settled = Some((*d).clone()); di.next(); } // Past the horizon on both sides, or both exhausted.