From 7f5ee7401f9eca9e189b797e6f5cb66b7cb2733d Mon Sep 17 00:00:00 2001 From: Edouard Vanbelle Date: Mon, 24 Aug 2026 21:25:08 +0200 Subject: [PATCH] refactor(storage): make blob enumeration ordered and hash-cursored MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Precondition for the merge-join in backend_consistency (step 6 / option A of docs/plan/derived-blobs.md), landed separately because it is independently useful and carries the risk. Two contract changes on BlobStorageBackend::list_blob_hashes: 1. Entries MUST be in ascending hash order. Every shipped backend already did this — local sorts within each shard and walks 00..ff, and since the shard IS the hash prefix that is globally sorted; S3 and Azure list lexicographically by key and blobs// sorts identically to . It was accidental, and a future backend enumerating in any other order would have silently made the merge-join emit bogus blob_missing_from_backend findings at data_loss severity. 2. The cursor is the last hash returned, not an opaque backend token. This is what lets a caller resume from a checkpoint it already holds — the merge-join keeps one cursor for both the DB walk and the backend walk instead of a compound one, which in turn means blobs_consistency's existing cursor format survives and no paused run is stranded. Local already derived its position from a hash; it now emits the bare hash instead of "/", and still accepts both legacy forms so a run paused across this deploy resumes. The bare-shard form works through the same path unchanged, since "3f" sorts before every 64-char hash beginning "3f". S3 moves from continuation_token to StartAfter, which supports this natively. One non-obvious case handled: a page can contain only non-canonical keys (.tmp spool files, .corrupt sidecars), which are filtered into `unknowns`, leaving `blobs` empty — a naive blobs.last() would return no cursor and silently end enumeration while is_truncated said otherwise, making an audit job under-report. It now falls back to the last key seen; StartAfter is a string comparison, so a non-hash resume point is fine. "Cursor is a hash" constrains what callers may synthesise, not what backends may return. Azure is unaffected — it does not implement list_blob_hashes (TODO, inherits the NotSupported default). Adds the first test for enumeration at all: ordering across shards with deliberately out-of-order inserts, complete paged traversal, and resume from a caller-synthesised cursor. NOT verified against real S3 — no bucket available here. The local path is covered by the new test; the StartAfter change is reasoned from the API contract and needs exercising against a real bucket before it is relied on. Co-Authored-By: Claude Opus 5 (1M context) --- src/application/ports/blob_storage_ports.rs | 21 +++- .../services/local_blob_backend.rs | 95 +++++++++++++++++-- .../services/s3_blob_backend.rs | 44 ++++++++- 3 files changed, 145 insertions(+), 15 deletions(-) diff --git a/src/application/ports/blob_storage_ports.rs b/src/application/ports/blob_storage_ports.rs index 262a1321..ae8732d3 100644 --- a/src/application/ports/blob_storage_ports.rs +++ b/src/application/ports/blob_storage_ports.rs @@ -244,11 +244,28 @@ pub trait BlobStorageBackend: Send + Sync + 'static { /// /// * `cursor` — opaque continuation token from a prior call, or /// `None` to start from the beginning. Format is per-backend - /// (local = last path visited; S3 = continuation token; Azure - /// = list marker); callers treat it as opaque. + /// — **the last blob hash returned by the previous page**. + /// Enumeration resumes strictly AFTER that hash. + /// + /// This is deliberately NOT an opaque backend token. Callers may + /// synthesise a cursor from any hash they hold, which is what lets a + /// consistency sweep merge-join this stream against a + /// `storage.blobs` walk and resume both sides from one checkpoint. + /// An opaque token would force the backend side to re-enumerate from + /// the beginning on every resume. /// * `limit` — soft cap on batch size; backends may return /// fewer (e.g. end of a shard directory). /// + /// **Entries MUST be returned in ascending hash order**, and pages must + /// be contiguous in that order. Every shipped backend already satisfies + /// this — local sorts within each shard and walks shards `00`..`ff` + /// (the shard IS the hash prefix, so that is globally sorted); S3 and + /// Azure list lexicographically by key, and `blobs//` sorts + /// identically to ``. It is stated here because the merge-join in + /// `backend_consistency` depends on it: an unordered backend would + /// silently emit bogus `blob_missing_from_backend` findings at + /// `data_loss` severity. + /// /// Returns `(entries, next_cursor)`. `next_cursor = None` means /// enumeration is complete. Each `BackendBlobEntry` carries the /// hash + optional mtime for grace-window filtering. diff --git a/src/infrastructure/services/local_blob_backend.rs b/src/infrastructure/services/local_blob_backend.rs index 20ed4eaf..4f0fac15 100644 --- a/src/infrastructure/services/local_blob_backend.rs +++ b/src/infrastructure/services/local_blob_backend.rs @@ -778,12 +778,30 @@ impl BlobStorageBackend for LocalBlobBackend { let blob_root = self.blob_root.clone(); Box::pin(async move { + // Cursor is the last hash returned (see the port contract). The + // shard is derivable from it — the shard name IS the hash's first + // two chars — so no composite is needed. + // + // Both legacy forms still resume correctly, so a consistency run + // paused across this deploy is not stranded: + // * "/" — what this backend used to emit; the + // hash half is taken and the shard re-derived from it. + // * "" — a bare 2-char shard. It flows through the same + // path: "3f" sorts BEFORE every 64-char hash beginning "3f", + // so using it as start_after skips nothing. let (start_shard, start_after_hash): (String, Option) = match cursor { None => (String::from("00"), None), - Some(c) => match c.split_once('/') { - Some((sh, h)) => (sh.to_string(), Some(h.to_string())), - None => (c, None), - }, + Some(c) => { + let hash = c.split_once('/').map(|(_, h)| h).unwrap_or(c.as_str()); + if hash.len() >= 2 { + (hash[..2].to_string(), Some(hash.to_string())) + } else { + // Under 2 chars — not a hash and not a shard. Should + // be unreachable; start from the beginning rather + // than index out of bounds. + (String::from("00"), None) + } + } }; let mut blobs: Vec = Vec::with_capacity(limit); @@ -879,11 +897,8 @@ impl BlobStorageBackend for LocalBlobBackend { continue; } if blobs.len() >= limit { - next_cursor = Some(format!( - "{}/{}", - prefix, - blobs.last().map(|e| e.hash.as_str()).unwrap_or("") - )); + // Just the hash — the shard is recoverable from it. + next_cursor = blobs.last().map(|e| e.hash.clone()); return Ok(BlobListPage { blobs, unknowns, @@ -1008,4 +1023,66 @@ mod tests { ); assert_eq!(hash_prefix_slot("gg"), None); } + + /// The port contract now REQUIRES ascending hash order and a cursor that + /// is the last hash returned. `backend_consistency`'s merge-join depends + /// on both: an out-of-order page would make it emit bogus + /// `blob_missing_from_backend` findings at `data_loss` severity, and a + /// non-hash cursor would stop a caller resuming from its own checkpoint. + /// + /// Nothing covered enumeration before this, so both properties were + /// accidental. + #[tokio::test] + async fn list_blob_hashes_is_ordered_and_hash_cursor_resumes() { + let dir = TempDir::new().unwrap(); + let backend = LocalBlobBackend::new(dir.path()); + backend.initialize().await.unwrap(); + + // Deliberately inserted out of order and across several shards, so a + // passing result cannot come from insertion order. + let mut written: Vec = ["f0", "0a", "9c", "0b", "ff", "12"] + .iter() + .map(|p| fake_hash(p)) + .collect(); + for h in &written { + backend + .put_blob_from_bytes(h, Bytes::from_static(b"x")) + .await + .unwrap(); + } + written.sort(); + + // Page with limit 2 so the cursor is exercised repeatedly. + let mut seen: Vec = Vec::new(); + let mut cursor: Option = None; + for _ in 0..20 { + let page = backend.list_blob_hashes(cursor.clone(), 2).await.unwrap(); + seen.extend(page.blobs.iter().map(|e| e.hash.clone())); + match page.next_cursor { + Some(c) => cursor = Some(c), + None => break, + } + } + + assert_eq!(seen, written, "enumeration must be complete and ascending"); + + // A cursor the CALLER synthesises from a hash it already holds must + // work — that is the property the merge-join resume relies on, and + // what an opaque backend token could not provide. + let midpoint = &written[2]; + let resumed = backend + .list_blob_hashes(Some(midpoint.clone()), 100) + .await + .unwrap(); + let expected: Vec = written[3..].to_vec(); + assert_eq!( + resumed + .blobs + .iter() + .map(|e| e.hash.clone()) + .collect::>(), + expected, + "resume must start STRICTLY after the given hash" + ); + } } diff --git a/src/infrastructure/services/s3_blob_backend.rs b/src/infrastructure/services/s3_blob_backend.rs index c58ef2ce..9dfc2591 100644 --- a/src/infrastructure/services/s3_blob_backend.rs +++ b/src/infrastructure/services/s3_blob_backend.rs @@ -497,8 +497,13 @@ impl BlobStorageBackend for S3BlobBackend { .list_objects_v2() .bucket(&self.bucket) .max_keys(limit.min(1000) as i32); + // Resume after a HASH, not a continuation token (port contract). + // ListObjectsV2 supports this natively via StartAfter, and it is + // what lets a caller resume the backend side of a merge-join from + // a checkpoint it holds — a continuation token would force a + // re-enumeration from the start on every resume. if let Some(c) = cursor { - req = req.continuation_token(c); + req = req.start_after(Self::object_key(&c)); } let resp = req.send().await.map_err(|e| { @@ -512,6 +517,9 @@ impl BlobStorageBackend for S3BlobBackend { let objects = resp.contents.unwrap_or_default(); let mut blobs: Vec = Vec::with_capacity(objects.len()); let mut unknowns: Vec = Vec::new(); + // Last key of the page regardless of classification — the resume + // fallback for an all-unknowns page (see next_cursor below). + let mut last_key: Option = None; for obj in objects { let Some(key) = obj.key else { continue }; @@ -539,13 +547,41 @@ impl BlobStorageBackend for S3BlobBackend { }); match is_canonical { - Some(hash) => blobs.push(BackendBlobEntry { hash, mtime }), - None => unknowns.push(BackendUnknownEntry { path: key, mtime }), + Some(hash) => { + last_key = Some(hash.clone()); + blobs.push(BackendBlobEntry { hash, mtime }) + } + None => { + last_key = Some(key.clone()); + unknowns.push(BackendUnknownEntry { path: key, mtime }) + } } } + // Resume point: the last hash of this page, not the continuation + // token — see the StartAfter note above. + // + // `blobs.last()` alone is NOT sufficient. A page can legitimately + // contain only non-canonical keys (`.tmp` spool files, `.corrupt` + // sidecars), which are filtered into `unknowns`; `blobs` is then + // empty and a naive `blobs.last()` yields None, silently ending + // enumeration while `is_truncated` says otherwise. A consistency + // sweep would under-report rather than fail — the worst shape of + // bug for an audit job. + // + // So fall back to the last KEY seen. StartAfter is a plain string + // comparison, so any key works as a resume point; it need not be + // a hash. The port contract's "cursor is a hash" is what CALLERS + // may synthesise, not a restriction on what backends may return. let next_cursor = if resp.is_truncated.unwrap_or(false) { - resp.next_continuation_token + match blobs.last() { + Some(entry) => Some(entry.hash.clone()), + None => last_key.map(|k| { + // Strip the `blobs//` prefix: object_key() re-adds + // it when this comes back as a cursor. + k.rsplit('/').next().unwrap_or(&k).to_string() + }), + } } else { None };