cjw/Oxicloud
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

feat(consistency): merge-join backend_consistency, both directions

Browse Source 
The job walked the backend and probed the DB with WHERE hash = ANY($1)
over each page, so it could only ever see backend-only entries. A
registry row whose bytes are gone never appears in a backend listing —
it was invisible here by construction, and that half was left to
blobs_consistency's per-row HEAD probe, which does not survive the row
counts this plan produces.

Both sides are now ordered by hash — the backend by contract since
5343fdda, the DB by ORDER BY hash — so one pass yields both deltas:
orphan_blob for bytes with no row, and blob_missing_from_backend for a
row with no bytes. The latter is severity data_loss rather than
inconsistent: an orphan wastes space, this loses a file.

Three things the merge needs that a probe did not.

A horizon. The two pages cover different ranges, so only their overlap
can be judged — beyond it, a hash missing from one side may simply be on
the next page of the other, and emitting there would invent findings in
both directions at once. When a side is exhausted its entries cannot be
on a later page, so the other's tail becomes judgeable.

One cursor for both sides. They share an ordering, so "resume after H"
is start_after(H) on the backend and hash > H in the DB. The cursor
advances to the horizon, not the backend's own next_cursor, which would
skip the un-judged tail of whichever side reached further. Format is
unchanged, so paused runs resume.

And an ordering premise worth stating rather than assuming: hashes are
lowercase BLAKE3 hex of fixed length, so collation and byte order rank
identically over [0-9a-f]. A hash column admitting uppercase or variable
length would break this silently. Recorded in the module docs and beside
the query.

blobs_consistency still emits its own blob_missing_from_backend; the two
now overlap. Retiring that probe is the follow-up, not folded in here.
This commit is contained in:
Edouard Vanbelle
2026-08-26 23:24:27 +02:00
parent d7de1c41e7
commit c656e684d4
1 changed files with 173 additions and 56 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/infrastructure/services/backend_consistency_service.rs
+173 -56
View File
@@ -1,18 +1,36 @@
//! Fifth tenant of Part 2 (recoverable-run engine).
//!
//! Iterates the storage backend's blob-enumeration surface and
//! reports every blob physically present on the backend that has NO
//! matching row in `storage.blobs`. Complements
//! `blobs_consistency` (which walks the DB and probes the backend):
//! together they close the reference graph.
//! **Merge-joins** the backend's blob enumeration against
//! `storage.blobs`, both ordered by hash, so a single pass yields the
//! delta in *both* directions rather than one.
//!
//! ### Per-row check
//! It previously walked the backend and probed the DB with
//! `WHERE hash = ANY($1)` over each page, which could only ever see
//! backend-only entries: a row whose bytes are gone never appears in a
//! backend listing, so it was invisible here by construction. That half
//! was left to `blobs_consistency`'s per-row HEAD probe, which does not
//! survive the row counts this plan produces — see
//! `docs/plan/derived-blobs.md`.
//!
//! ### Per-row checks
//!
//! * `orphan_blob` (severity `inconsistent`) — bytes on disk / S3 /
//! Azure with no registry row. Not data-loss (nothing broken —
//! just storage overhead), but points at dedup_gc or
//! ingest-path drift. Recovery = register-registry-row (if the
//! bytes are still needed) OR delete the file (if truly orphan).
//! * `blob_missing_from_backend` (severity `data_loss`) — a registry
//! row whose bytes are absent. The opposite direction and the more
//! serious one: an orphan wastes space, this loses a file.
//!
//! ### Why the two orderings agree
//!
//! The merge-join's premise is that the backend's byte order and the
//! database's `ORDER BY hash` rank identically. They do, because hashes
//! are lowercase BLAKE3 hex of fixed length: over `[0-9a-f]` digits
//! precede letters in both, and there is no case to fold. A hash column
//! that ever admitted uppercase or variable length would break this
//! silently and in both directions at once.
//!
//! ### Run-level check
//!
@@ -42,7 +60,6 @@
//! denominator (backend count ≈ blob count on a healthy install;
//! deviation IS the finding).
use std::collections::HashSet;
use std::sync::Arc;
use async_trait::async_trait;
@@ -376,60 +393,161 @@ impl RecoverableJobHandler for BackendConsistencyCheck {
return RunOutcome::completed();
}
// Batch DB probe: which of these hashes have a
// `storage.blobs` row? One `WHERE hash = ANY($1)` per
// batch — indexed lookup, cheap even on millions of
// rows.
let batch_hashes: Vec<String> = page.blobs.iter().map(|e| e.hash.clone()).collect();
let db_present: HashSet<String> = if batch_hashes.is_empty() {
HashSet::new()
} else {
match sqlx::query_as::<_, (String,)>(
r#"SELECT hash FROM storage.blobs WHERE hash = ANY($1)"#,
)
.bind(&batch_hashes[..])
.fetch_all(self.pool.as_ref())
.await
{
Ok(rows) => rows.into_iter().map(|(h,)| h).collect(),
Err(e) => {
return RunOutcome::Failed {
message: format!("db probe: {e}"),
};
}
// ── Merge-join, not a one-sided probe ────────────────
//
// Both sides are ordered by hash ascending — the backend by
// contract (`BlobStorageBackend::list_blob_hashes`), the DB by
// `ORDER BY hash` — so one pass yields BOTH deltas instead of
// one:
//
// * present on the backend, absent from the DB → `orphan_blob`
// * present in the DB, absent from the backend →
// `blob_missing_from_backend` (data loss, not overhead)
//
// The old form probed `WHERE hash = ANY($1)` over the backend
// page, so it could only ever see the first kind: a row whose
// bytes are gone never appears in a backend listing and was
// invisible here by construction.
//
// Ordering is the whole premise, so it is worth being explicit
// about why the two agree. Hashes are lowercase BLAKE3 hex of
// fixed length, and over `[0-9a-f]` the database collation and
// byte order rank identically (digits before letters in both,
// no case folding to disagree about). A hash column that ever
// admitted uppercase or variable length would break this
// silently, in both directions.
let db_hashes: Vec<String> = match sqlx::query_as::<_, (String,)>(
r#"SELECT hash FROM storage.blobs
WHERE ($1::text IS NULL OR hash > $1)
ORDER BY hash
LIMIT $2"#,
)
.bind(cursor.as_deref())
.bind(BATCH_SIZE as i64)
.fetch_all(self.pool.as_ref())
.await
{
Ok(rows) => rows.into_iter().map(|(h,)| h).collect(),
Err(e) => {
return RunOutcome::Failed {
message: format!("db page: {e}"),
};
}
};
for entry in &page.blobs {
if db_present.contains(&entry.hash) {
continue;
}
if let Some(mtime) = entry.mtime
&& mtime > grace_cutoff
{
continue;
}
// The two pages cover different ranges, so only the overlap can
// be judged. Beyond `horizon` a hash missing from one side may
// simply be on the next page of the other, and emitting there
// would invent findings in both directions. When a side is
// exhausted its entries cannot be "on a later page", so the
// other side's tail becomes judgeable.
let backend_last = page.blobs.last().map(|e| e.hash.as_str());
let db_last = db_hashes.last().map(|s| s.as_str());
let backend_done = page.next_cursor.is_none();
let db_done = db_hashes.len() < BATCH_SIZE;
finding_count += 1;
record_or_log(
store,
BACKEND_CONSISTENCY_JOB_NAME,
"orphan_blob",
"inconsistent",
None,
serde_json::json!({
"hash": entry.hash,
"mtime": entry.mtime.map(|t| t.to_rfc3339()),
"backend": backend.backend_type(),
}),
)
.await;
let horizon: Option<&str> = match (backend_last, db_last) {
_ if backend_done && db_done => None, // judge everything
(Some(b), Some(d)) if backend_done => Some(b.max(d)),
(Some(b), Some(d)) if db_done => Some(b.max(d)),
(Some(b), Some(d)) => Some(b.min(d)),
(Some(b), None) => Some(b),
(None, Some(d)) => Some(d),
(None, None) => None,
};
let in_range = |h: &str| horizon.is_none_or(|limit| h <= limit);
let mut bi = page.blobs.iter().peekable();
let mut di = db_hashes.iter().peekable();
loop {
match (bi.peek(), di.peek()) {
(Some(b), Some(d)) if b.hash == **d => {
bi.next();
di.next();
}
// Backend-only: bytes with no registry row.
(Some(b), d_opt)
if d_opt.is_none_or(|d| b.hash.as_str() < d.as_str())
&& in_range(&b.hash) =>
{
// Grace window: the write path is
// durability-before-visibility, so bytes exist
// briefly before their row does. Without this every
// in-flight upload reads as an orphan.
if !matches!(b.mtime, Some(m) if m > grace_cutoff) {
finding_count += 1;
record_or_log(
store,
BACKEND_CONSISTENCY_JOB_NAME,
"orphan_blob",
"inconsistent",
None,
serde_json::json!({
"hash": b.hash,
"mtime": b.mtime.map(|t| t.to_rfc3339()),
"backend": backend.backend_type(),
}),
)
.await;
}
bi.next();
}
// DB-only: a row whose bytes are gone. Severity is
// `data_loss`, not `inconsistent` — an orphan wastes
// space, this loses a file.
(b_opt, Some(d))
if b_opt.is_none_or(|b| d.as_str() < b.hash.as_str()) && in_range(d) =>
{
finding_count += 1;
record_or_log(
store,
BACKEND_CONSISTENCY_JOB_NAME,
"blob_missing_from_backend",
"data_loss",
None,
serde_json::json!({
"hash": d,
"backend": backend.backend_type(),
"note": "registry row with no bytes on the backend",
}),
)
.await;
di.next();
}
// Past the horizon on both sides, or both exhausted.
_ => break,
}
}
// Advance cursor + checkpoint. Scanned count tracks
// both blobs and unknowns since we walked both.
// Advance to the horizon, not the backend's own cursor.
//
// One hash serves both sides: they share an ordering, so "resume
// after H" means `start_after(H)` on the backend and
// `WHERE hash > H` in the DB. Advancing past the horizon would
// skip the un-judged tail of whichever side reached further.
//
// Scanned count covers blobs and unknowns, since both were
// walked.
let batch_len = (page.blobs.len() + page.unknowns.len()) as u64;
cursor = page.next_cursor;
let exhausted = backend_done && db_done;
cursor = if exhausted {
None
} else {
horizon.map(|h| h.to_string())
};
// Neither side exhausted yet no horizon means neither returned a
// row — nothing left to compare, and continuing would spin on the
// same empty pages forever.
if cursor.is_none() && !exhausted && horizon.is_none() {
tracing::debug!(
target: "oxicloud::consistency",
event = "backend_consistency.no_horizon",
run_id = %store.run_id(),
"both sides returned no rows before exhaustion; ending the sweep"
);
}
let cursor_bytes = cursor
.as_ref()
.map(|s| s.as_bytes().to_vec())
@@ -440,8 +558,7 @@ impl RecoverableJobHandler for BackendConsistencyCheck {
};
}
// Backend returned no next_cursor → enumeration
// complete. Emit the completion log and return.
// Both sides drained → the sweep is complete.
if cursor.is_none() {
tracing::info!(
target: "oxicloud::consistency",