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

refactor(consistency): blobs_consistency reads only the database

Browse Source 
`blobs_consistency` probed `blob_exists` once per row and, under
`?deep=true`, read and re-hashed every blob. `backend_consistency`
already reports the same `blob_missing_from_backend` from its
merge-join — so the probe was duplicated work that found strictly less
(a DB walk cannot see backend-only orphans by construction) at N round
-trips instead of one enumeration. Every scheduled sweep paid for it.

All three physical checks move to `backend_consistency`:

* `blob_missing_from_backend` was already there; the duplicate is gone.
* `blob_corrupted` / `blob_unreadable` hook the matched arm of the
  merge-join, which holds exactly the key pairs worth reading. Guarded
  by `in_range` so a pair past the horizon is not read twice, and
  `params.deep` is persisted on a fresh run and read back on resume so
  a paused deep scan does not silently continue shallow.

Deep mode belongs there because it is backend work end to end: the
only DB input is the hash. Keeping it in `blobs_consistency` forced
that tenant to carry a backend for one flag.

What remains is the half that needs no backend: `refcount_mismatch`
and its repair. The constructor drops from five parameters to two —
no backend, no storage_entries, no storage_path_fallback — and
`?storage=<name>` / `?deep=true` are now inert there, which the
job description says outright.

`affected_files` is needed by both tenants, so it moves to a shared
`blob_diagnostics` module rather than being copied.
`PROBED_STORAGE_PARAM` moves to `backend_consistency`: it was defined
in `blobs_consistency` and re-exported, which is backwards once the
DB-only tenant has no entry to scope. The create-grace window goes
with the probe — it existed to avoid flagging a blob whose bytes had
landed before its row, and the refcount comparison reads one
consistent snapshot.

Known cost: `backend_consistency` returns `backend_unenumerable` on
Azure and mid-migration, so on those configs missing bytes now go
unreported where the per-row probe caught them. That argues for the
Azure enumeration impl, not for keeping the probe.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Edouard Vanbelle
2026-08-29 01:44:08 +02:00
parent 1ea3826660
commit f1f327a6c4
11 changed files with 361 additions and 474 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/architecture/backend-storage.md
+25 -14
View File
@@ -234,27 +234,38 @@ can be safely dropped.
---
## 4. Blob consistency (`blobs_consistency`)
## 4. Blob consistency — two jobs, split by what they read
Read-only recoverable job that walks `storage.blobs` and reports
divergence between the DB registry and the physical backend.
The registry side and the physical side are separate tenants. They
used to be one, with `blobs_consistency` probing the backend once per
row; that probe found strictly less than the merge-join below, at N
round-trips instead of one enumeration, so it was removed.
### Shallow mode (default)
### `blobs_consistency` — database only
Per row:
Walks `storage.blobs` and compares `ref_count` against the reference
count computed from `storage.files.blob_hash` +
`chunk_manifests.chunk_hashes[]`. On mismatch: `refcount_mismatch`
(severity `inconsistent`), repairable under `?repair=true`.
- `blob_exists(hash)` on the active backend → if false, record
`blob_missing_from_backend` (severity `data_loss`)
- Compare `ref_count` against the actual reference count computed
from `SUM` over `storage.files.blob_hash` + `chunk_manifests.chunk_hashes[]`
→ if mismatch, record `refcount_mismatch` (severity `inconsistent`)
It opens no backend and makes no network call. `?storage=<name>` and
`?deep=true` are inert. Cost is one aggregate SQL per row.
Cost: one existence probe + one aggregate SQL per row. Fast on
S3/Azure (single HEAD).
### `backend_consistency` — everything physical
### Deep mode (`?deep=true`)
Merge-joins the backend's enumeration against `storage.blobs`, both
ordered by hash, yielding both deltas in one pass:
Adds a full read of every blob:
- bytes with no registry row → `orphan_blob` (severity `inconsistent`)
- a registry row with no bytes → `blob_missing_from_backend`
(severity `data_loss`)
`?storage=<name>` scopes it to any declared entry rather than the live
backend.
### Deep mode (`?deep=true`, on `backend_consistency`)
For every hash present on both sides, adds a full read:
- Stream the blob through `EncryptedBlobBackend::get_blob_stream`
(strips header, decrypts if needed, applies BLAKE3 rescue for
docs/config/admin-settings.md
+6 -2
View File
@@ -100,12 +100,16 @@ This one-shot repair command re-runs the same env-parse the server does at boot,
### Auditing entries other than the active one
`blobs_consistency` and `backend_consistency` (recoverable jobs on the Jobs tab) accept `?storage=<name>` to probe any declared entry — not just the live one. Use this to verify a migration target before cutover, or to audit an old backend after cutover but before decommissioning:
`backend_consistency` (a recoverable job on the Jobs tab) accepts `?storage=<name>` to audit any declared entry — not just the live one. Use this to verify a migration target before cutover, or to audit an old backend after cutover but before decommissioning:
```
POST /api/admin/jobs/blobs_consistency/trigger?storage=<name>
POST /api/admin/jobs/backend_consistency/trigger?storage=<name>
```
Add `?deep=true` to also read every blob back and re-hash it, which catches silent bit-rot. That is a full read of the entry and can take hours.
`blobs_consistency` does *not* accept `?storage=<name>`: it only reads the database, so there is no entry for it to scope.
Unknown names 400 at the HTTP layer.
## Data Storage
docs/plan/derived-blobs.md
+2 -2
View File
@@ -949,8 +949,8 @@ Findings each job reports today, and where the new tables land:
| # | Edge | Direction | Mechanism | Status |
|---|---|---|---|---|
| 1 | backend → `storage.blobs` | orphan bytes | `orphan_blob` (backend_consistency) | ✓ |
| 2 | `storage.blobs` → backend | missing bytes | `blob_missing_from_backend` | ✓ |
| 3 | chunk bytes | corruption | `blob_corrupted`, `blob_unreadable` | ✓ |
| 2 | `storage.blobs` → backend | missing bytes | `blob_missing_from_backend` (backend_consistency) | ✓ |
| 3 | chunk bytes | corruption | `blob_corrupted`, `blob_unreadable` (backend_consistency, `?deep=true`) | ✓ |
| 4 | manifest → chunks | chunk reaped | `chunk_missing` (files_consistency) | ✓ |
| 5 | `files` → Blob | dangling | `missing_blob` (files_consistency) | ✓ |
| 6 | `storage.blobs.ref_count` | recompute | `refcount_mismatch` | ✓ chunk level only |
docs/plan/storage-key-rotation.md
+11 -1
View File
@@ -179,7 +179,17 @@ Guardrail:
head pair's fingerprint differs from the second entry's — signals "you
added new keys but haven't rotated legacy blobs yet".
* A **legacy-blob counter** is surfaced in the admin panel per storage entry.
The counter is maintained by `blobs_consistency`: during its normal walk it
> **Retarget (post-split).** This plan names `blobs_consistency` as the
> host for the magic-byte branch throughout. That is no longer the right
> tenant: `blobs_consistency` became database-only and opens no backend,
> while `backend_consistency` owns every physical check and already holds
> the enumeration. Read `backend_consistency` wherever the sections below
> say `blobs_consistency`. Nothing else about the design changes — the
> magic-byte check still rides along on an existing walk, and still lands
> in the run's `stats` bag.
The counter is maintained by that scan: during its normal walk it
branches on the magic-byte check and records the legacy count as a run
statistic on `jobs.recoverable_runs` (existing surface, no schema hit). The
admin panel reads the most recent count and displays it. Refresh cadence is
docs/plan/storage-multi-entry.md
+4 -2
View File
@@ -407,10 +407,12 @@ Per slice, plus these end-to-end scenarios in Hurl:
6. **In-place encryption rotation refused**: two entries, same S3 bucket,
different encryption keys. Trigger migration → refuses with the specific
error message pointing at the encryption case and the two-step workaround.
7. **`?storage=<name>` on blobs_consistency**: run against `s3_prod` before
7. **`?storage=<name>` on backend_consistency**: run against `s3_prod` before
cutover. Full walk, `probed_storage` in run row. Then cutover, then rerun
against `local_main` — verifies old backend still has everything.
8. **Unknown storage name**: `POST /admin/jobs/blobs_consistency/trigger?storage=nope`
(This scenario named `blobs_consistency` until that tenant became
database-only; entry scoping belongs to whichever job opens a backend.)
8. **Unknown storage name**: `POST /admin/jobs/backend_consistency/trigger?storage=nope`
→ 400 with known-names list. No run row created.
9. **Missing entry at boot**: `active_backend_name = "gone"` but `_ENTRIES`
doesn't include it → boot aborts with the specific message pointing at
frontend/src/lib/components/AdminJobsPanel.svelte
+7 -6
View File
@@ -637,13 +637,14 @@
// Jobs that respect `?deep=true`:
// * `consistency_batch` — propagates deep to every child that
// understands it
// * `blobs_consistency` — deep mode re-reads + re-hashes every
// blob for silent bit-rot detection (severity `data_loss`).
// Full read of storage; can take hours on big installs — the
// "Run" (normal) button on the same row does the cheap
// existence probes only.
// * `backend_consistency` — deep mode re-reads + re-hashes every
// matched blob for silent bit-rot detection (severity
// `data_loss`). Full read of storage; can take hours on big
// installs — the "Run" button on the same row does the
// enumeration merge-join only. This was `blobs_consistency`
// until that tenant became database-only.
function supportsDeep(name: string): boolean {
return name === 'consistency_batch' || name === 'blobs_consistency';
return name === 'consistency_batch' || name === 'backend_consistency';
}
// Whether `?repair=true` does anything for this job — declared by the
src/common/di.rs
+12 -15
View File
@@ -1536,24 +1536,21 @@ impl AppServiceFactory {
.register_recoverable_job(&core.job_registry, &job_store_provider_dyn)
.await;
// Fourth recoverable-run tenant. Iterates `storage.blobs`
// and verifies each row against the physical backend AND
// against the reference-counting invariants that `dedup_gc`
// relies on. Three per-row checks (subject-iteration in
// action): `blob_missing_from_backend` (data_loss, bytes
// gone from disk), `refcount_mismatch` (inconsistent,
// dedup counter drift), and `blob_corrupted` (data_loss,
// deep mode only — bit-rot). Complements
// `files_consistency` without doubling work: probing
// per-unique-blob preserves dedup savings vs probing
// per-file-chunk. See memory
// `project_cdc_dual_storage_registries` for the rationale.
// Fourth recoverable-run tenant. Iterates `storage.blobs` and
// checks the reference-counting invariant `dedup_gc` relies on:
// `refcount_mismatch` (inconsistent — an under-count lets GC reap
// a live blob, an over-count pins a dead one), repairable under
// `?repair=true`.
//
// DB-only, and takes no backend. Physical checks — missing bytes,
// orphaned bytes, bit-rot — all belong to `backend_consistency`,
// which merge-joins the backend enumeration against this same
// table in one pass. This tenant used to probe the backend once
// per row for missing bytes, which found strictly less than the
// merge-join at N round-trips instead of one enumeration.
let _ = Arc::new(
crate::infrastructure::services::blobs_consistency_service::BlobsConsistencyCheck::new(
maintenance_pool.clone(),
core.blob_backend.clone(),
core.config.storage_entries.clone(),
self.storage_path.clone(),
// Same registry instance GC reaps from — see
// DedupService::reference_registry.
core.dedup_service.reference_registry(),
src/infrastructure/services/backend_consistency_service.rs
+190 -8
View File
@@ -10,7 +10,8 @@
//! 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`.
//! `docs/plan/derived-blobs.md`. That probe is now gone: this tenant
//! owns every backend-side check, and `blobs_consistency` is DB-only.
//!
//! ### Per-row checks
//!
@@ -22,6 +23,22 @@
//! * `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.
//! * `blob_corrupted` (severity `data_loss`, `?deep=true` only) —
//! the key exists on both sides but the bytes behind it no longer
//! hash to it. Silent bit-rot.
//! * `blob_unreadable` (severity `data_loss`, `?deep=true` only) —
//! the key exists but the bytes cannot be read at all: decrypt
//! failure (missing key), transport error, permissions. Same impact
//! as corruption from a file's point of view, different remedy,
//! hence a separate kind. Triage on the recorded `error`.
//!
//! ### Deep mode
//!
//! The last two moved here from `blobs_consistency`, which used to
//! carry a backend solely for them. Re-hashing is backend work end to
//! end — the only DB input is the hash — and this walk already holds
//! the matched key pairs, which is exactly the set worth reading. It
//! costs a full read of every blob, so it is opt-in.
//!
//! ### Why the two orderings agree
//!
@@ -71,14 +88,18 @@ use crate::infrastructure::scheduler::{
JobRegistry, JobRunArgs, JobStore, JobStoreProvider, ProgressKind, RecoverableJobHandler,
RunOutcome, RunStatus, record_or_log,
};
use crate::infrastructure::services::blob_diagnostics::affected_files;
pub const BACKEND_CONSISTENCY_JOB_NAME: &str = "backend_consistency";
/// Same `params` JSONB key `blobs_consistency` uses — kept identical
/// so operators grepping run rows see the same convention across
/// both storage-audit tenants.
pub const PROBED_STORAGE_PARAM: &str =
crate::infrastructure::services::blobs_consistency_service::PROBED_STORAGE_PARAM;
/// `params` JSONB key under which the entry name being enumerated is
/// stashed on a Fresh run (matches `TARGET_NAME_PARAM` on
/// `backend_migration`). Resumed runs re-read it so a paused audit
/// survives restart without the admin re-specifying the target.
///
/// Defined here rather than in `blobs_consistency`, which no longer
/// touches a backend and so has no entry to scope.
pub const PROBED_STORAGE_PARAM: &str = "probed_storage";
/// Batch size for backend enumeration + DB probe. 500 is enough to
/// amortise the DB round-trip while keeping the cancel-poll cadence
@@ -151,8 +172,10 @@ impl RecoverableJobHandler for BackendConsistencyCheck {
"Merge-joins the storage backend's blob enumeration against \
storage.blobs, both ordered by hash, so one pass yields the delta \
in both directions: bytes on the backend no DB row claims, and \
rows whose bytes are gone. Read-only — nothing is uploaded or \
deleted."
rows whose bytes are gone. Add ?deep=true to also read every \
matched blob back and re-hash it, catching silent bit-rot — that \
is a full read of storage and can take hours. Read-only in both \
modes: nothing is uploaded or deleted."
}
/// Approximate total: on a healthy install every backend blob
@@ -259,6 +282,49 @@ impl RecoverableJobHandler for BackendConsistencyCheck {
);
}
// Deep mode — read every matched blob back and re-hash it, rather
// than trusting that a key present on both sides means the bytes
// behind it are still the bytes that key names.
//
// It lives here rather than in `blobs_consistency` because it is
// a backend operation end to end: the only DB input is the hash,
// which this merge-join already holds. Keeping it there forced
// that tenant to carry a backend for one flag, which is the
// overlap this split removes.
//
// Persisted to `params.deep` on a Fresh run so a Resume picks up
// the same mode (a Paused deep scan must not silently continue
// shallow) and the admin run-detail view can show what the scan
// actually verified. Written BEFORE the walk so a crash mid-batch
// still leaves the marker.
let deep = if is_fresh {
let v = if args.deep { "true" } else { "false" };
if let Err(e) = store.set_string_param("deep", v).await {
return RunOutcome::Failed {
message: format!("failed to persist deep flag to params: {e}"),
};
}
args.deep
} else {
match store.get_string_param("deep").await {
Ok(Some(v)) => v == "true",
Ok(None) => false,
Err(e) => {
return RunOutcome::Failed {
message: format!("read `deep` from params: {e}"),
};
}
}
};
if deep {
tracing::info!(
target: "oxicloud::consistency",
event = "backend_consistency.deep_mode_active",
run_id = %store.run_id(),
"deep mode: re-reading + re-hashing every matched blob (bit-rot detection)"
);
}
// Cursor = opaque backend continuation token, UTF-8-encoded.
// Each backend defines its own format (local = shard/hash,
// S3 = ListObjectsV2 continuation token, Azure = list
@@ -469,7 +535,20 @@ impl RecoverableJobHandler for BackendConsistencyCheck {
let mut di = db_hashes.iter().peekable();
loop {
match (bi.peek(), di.peek()) {
// Present on both sides. Shallow: nothing to say — the
// key exists where the registry claims. Deep: the key
// matching says nothing about the bytes behind it, so
// read them back and re-hash.
//
// Guarded by `in_range` so a pair past the horizon is
// not read twice — the cursor stops at the horizon, so
// that pair comes round again next batch and is
// verified then.
(Some(b), Some(d)) if b.hash == **d => {
if deep && in_range(&b.hash) {
finding_count +=
self.verify_bytes(store, backend.as_ref(), &b.hash).await;
}
bi.next();
di.next();
}
@@ -581,3 +660,106 @@ impl RecoverableJobHandler for BackendConsistencyCheck {
}
}
}
impl BackendConsistencyCheck {
/// Deep-mode per-blob verification. Reads the blob back, re-hashes it,
/// and records what it finds. Returns the number of findings recorded
/// (0 or 1) so the caller's counter stays the single tally.
///
/// Moved here from `blobs_consistency` along with the rest of the
/// backend-touching work: the merge-join already holds a verified
/// key pair, which is exactly the set worth reading.
async fn verify_bytes(
&self,
store: &dyn JobStore,
backend: &dyn BlobStorageBackend,
hash: &str,
) -> u64 {
match recompute_hash(backend, hash).await {
// The bytes still hash to the key they are filed under.
Ok(computed) if computed == hash => 0,
// Silent bit-rot. `computed_hash` is reported rather than a
// bare "mismatch" because the value is diagnostic: a one-bit
// flip, a truncation and a whole-object swap leave distinct
// signatures.
Ok(computed) => {
let affected = affected_files(self.pool.as_ref(), hash).await;
record_or_log(
store,
BACKEND_CONSISTENCY_JOB_NAME,
"blob_corrupted",
"data_loss",
None,
serde_json::json!({
"hash": hash,
"computed_hash": computed,
"backend": backend.backend_type(),
"affected_files": affected,
}),
)
.await;
1
}
// Bytes are there by key but cannot be read at all: decrypt
// failure (missing key), transport error, permissions. Same
// impact as corruption from a file's point of view — the
// content is inaccessible — but a different remedy, which is
// why it is a separate kind rather than folded into
// `blob_corrupted`. Operators triage on `error`.
Err(e) => {
let affected = affected_files(self.pool.as_ref(), hash).await;
record_or_log(
store,
BACKEND_CONSISTENCY_JOB_NAME,
"blob_unreadable",
"data_loss",
None,
serde_json::json!({
"hash": hash,
"backend": backend.backend_type(),
"affected_files": affected,
"error": e.to_string(),
}),
)
.await;
tracing::warn!(
target: "oxicloud::consistency",
event = "backend_consistency.blob_unreadable",
run_id = %store.run_id(),
hash = %hash,
error = %e,
"🚨 blob unreadable in deep mode — recorded finding, continuing"
);
1
}
}
}
}
/// Deep-mode helper — read the blob from the backend and recompute its
/// BLAKE3 hash. Returns the recomputed hex string; callers compare it
/// against the expected hash themselves. Returning the actual hash (not
/// a bool) lets the finding surface WHAT the bytes now hash to, which is
/// diagnostic gold: a one-bit flip has a very different signature from a
/// chunk-boundary corruption or a truncated read. `Err(_)` on any
/// backend-side error — the caller records that as `blob_unreadable`
/// rather than as corruption.
async fn recompute_hash(
backend: &dyn BlobStorageBackend,
expected_hash: &str,
) -> Result<String, crate::common::errors::DomainError> {
use crate::common::errors::DomainError;
use futures::StreamExt;
let mut stream = backend.get_blob_stream(expected_hash).await?;
let mut hasher = blake3::Hasher::new();
while let Some(chunk) = stream.next().await {
let bytes = chunk.map_err(|e| {
DomainError::internal_error("BackendConsistency", format!("stream read: {e}"))
})?;
hasher.update(&bytes);
}
Ok(hasher.finalize().to_hex().to_string())
}
src/infrastructure/services/blob_diagnostics.rs
+46
View File
@@ -0,0 +1,46 @@
//! Reverse-lookup helpers shared by the storage consistency tenants.
//!
//! `blobs_consistency` (DB-side: refcount drift) and
//! `backend_consistency` (backend-side: missing / orphaned / corrupted
//! bytes) both answer the same operator question when they emit a
//! finding — *which files does this hash break?* — so the query lives
//! here rather than in whichever tenant happened to need it first.
use sqlx::PgPool;
/// Cap on reverse-lookup file names surfaced in a finding's detail.
/// Keeps detail JSON bounded when a broken blob is referenced by
/// hundreds of files.
const AFFECTED_FILES_SAMPLE: i64 = 5;
/// Sample of file names that reference this blob — either directly
/// (`files.blob_hash = $hash`, legacy pre-CDC) or transitively via a
/// manifest (`chunk_hashes @> ARRAY[$hash]`, the post-CDC dominant
/// path). Capped so a chunk shared by 10 000 files doesn't blow up the
/// finding detail JSON. Order is arbitrary — this samples for
/// diagnosis, it does not enumerate.
///
/// Returns an empty vec on query error: a finding with no sample is
/// still a finding, and failing the sweep because the diagnostic
/// garnish didn't load would trade the whole scan for a nicety.
pub(crate) async fn affected_files(pool: &PgPool, hash: &str) -> Vec<String> {
let rows: Vec<(String,)> = sqlx::query_as(
r#"
SELECT DISTINCT f.name
FROM storage.files f
WHERE f.blob_hash = $1
OR EXISTS (
SELECT 1 FROM storage.chunk_manifests m
WHERE m.file_hash = f.blob_hash
AND $1 = ANY(m.chunk_hashes)
)
LIMIT $2
"#,
)
.bind(hash)
.bind(AFFECTED_FILES_SAMPLE)
.fetch_all(pool)
.await
.unwrap_or_default();
rows.into_iter().map(|(n,)| n).collect()
}
src/infrastructure/services/blobs_consistency_service.rs
+57 -424
View File
@@ -1,32 +1,12 @@
//! Fourth tenant of Part 2 (recoverable-run engine).
//!
//! Iterates `storage.blobs` — the content-addressable registry —
//! and verifies each row against the physical backend AND against
//! the reference-counting invariants that `dedup_gc` relies on.
//! Iterates `storage.blobs` — the content-addressable registry — and
//! checks the reference-counting invariant `dedup_gc` relies on.
//!
//! Three per-row checks (subject-iteration principle in action —
//! one walk, multiple branches):
//! **Database only.** It opens no backend and makes no network call;
//! `?storage=<name>` and `?deep=true` are both inert here.
//!
//! * `blob_missing_from_backend` (severity `data_loss`) — the DB
//! row says the hash exists but `BlobStorageBackend::blob_exists`
//! returns false. Bytes gone from disk / S3 / Azure. Any file
//! whose manifest references this hash (or whose whole-file
//! `blob_hash` points at it) will fail to read.
//!
//! * `blob_corrupted` (severity `data_loss`, deep mode only) —
//! bytes exist on the backend but their BLAKE3 no longer matches
//! the hash under which they're indexed. Silent bit-rot. Only
//! runs when the operator passes `?deep=true` because it costs a
//! full read of every blob.
//!
//! * `blob_unreadable` (severity `data_loss`, deep mode only) —
//! `blob_exists` returned true but the read pipeline errored (can't
//! decrypt, network glitch, permission error, etc.). Distinct from
//! `blob_corrupted` (which requires successful read + hash mismatch);
//! here we can't get bytes out at all. Same operator impact — any
//! file referencing this hash is inaccessible — but the remedy
//! differs (key recovery, retry, or blob replacement, depending on
//! the recorded `error` field).
//! One per-row check:
//!
//! * `refcount_mismatch` (severity `inconsistent`) —
//! `storage.blobs.ref_count` disagrees with the actual reference
@@ -36,85 +16,57 @@
//! a blob is being pinned longer than needed. Content-safe either
//! way (the storage.blobs row is fine, the counter is wrong).
//!
//! ### Complements `files_consistency`
//! ### Why nothing physical lives here any more
//!
//! `files_consistency` (Slice 6/10) iterates files and verifies DB
//! integrity. `blobs_consistency` iterates the storage registry and
//! verifies physical existence + counter integrity. Together they
//! cover both sides of the reference graph. Neither doubles the
//! other's work — probing per-blob (here) instead of per-file-chunk
//! preserves dedup savings: a chunk shared by 5 files gets probed
//! ONCE.
//! This tenant used to probe `BlobStorageBackend::blob_exists` once
//! per row for `blob_missing_from_backend`, and under `?deep=true`
//! read and re-hashed every blob for `blob_corrupted` /
//! `blob_unreadable`.
//!
//! ### Not covered here
//! All three moved to `backend_consistency`, which merge-joins the
//! backend's enumeration against this same table in one ordered pass.
//! It reports the same missing bytes, plus the backend-only orphans a
//! DB walk cannot see by construction, at one enumeration instead of
//! N round-trips — and a deep pass there re-hashes the matched pairs
//! it already holds. Keeping the probe here bought nothing and made
//! every scheduled sweep pay for it.
//!
//! * **Orphan bytes on the backend** (files on disk with no DB row)
//! — belongs in the future `backend_consistency` tenant which
//! iterates the backend itself. Requires the `list_blob_hashes`
//! trait extension and per-backend enumeration impls.
//! What is left is the half that needs no backend at all: a counter,
//! and the two tables that determine what it should be.
//!
//! ### Elsewhere in the graph
//!
//! * **Physical existence, orphan bytes, bit-rot** —
//! `backend_consistency`.
//! * **File-side DB integrity** (parent folder, blob reference,
//! denormalised size) — `files_consistency`.
//! * **Manifest-level integrity** (`storage.chunk_manifests` rows
//! pointing at reaped chunks) — already covered by
//! `files_consistency::chunk_missing`.
//! pointing at reaped chunks) — `files_consistency::chunk_missing`.
//! * **The OTHER refcount** (`chunk_manifests.ref_count`, which every
//! whole-Blob reference lands on) — `manifests_consistency`.
use std::path::PathBuf;
use std::sync::Arc;
use async_trait::async_trait;
use chrono::{DateTime, Duration, Utc};
use sqlx::PgPool;
use crate::application::ports::blob_reference_ports::{BlobReferenceRegistry, RefLevel};
use crate::application::ports::blob_storage_ports::BlobStorageBackend;
use crate::common::config::NamedStorageEntry;
use crate::infrastructure::scheduler::{
JobRegistry, JobRunArgs, JobStore, JobStoreProvider, Mutates, RecoverableJobHandler,
RunOutcome, RunStatus, record_or_log,
};
use crate::infrastructure::services::entry_backend::build_entry_backend;
use crate::infrastructure::services::blob_diagnostics::affected_files;
pub const BLOBS_CONSISTENCY_JOB_NAME: &str = "blobs_consistency";
/// `params` JSONB key under which the entry name being probed is
/// stashed on a Fresh run (matches `TARGET_NAME_PARAM` on
/// `backend_migration`). Resumed runs re-read it so a paused audit
/// survives restart without the admin re-specifying the target.
pub const PROBED_STORAGE_PARAM: &str = "probed_storage";
/// Rows per batch. Blobs are numerous (millions on a busy install)
/// but per-row work is one indexed backend probe + one indexed SQL
/// ref-count query. 200 balances cancel-poll cadence against
/// round-trip amortisation.
/// but per-row work is now a single indexed SQL ref-count comparison,
/// with no backend round-trip. 200 balances cancel-poll cadence
/// against round-trip amortisation.
const BATCH_SIZE: i64 = 200;
/// Grace window — rows created within this window are skipped by
/// the physical-existence probe because the write path is
/// durability-before-visibility: `dedup_service` writes bytes, then
/// registers the row a few ms later. A scan catching a row
/// mid-write would false-positive it as `blob_missing_from_backend`.
/// Same shape `dedup_gc` uses (see its `grace_secs`).
const CREATE_GRACE: Duration = Duration::hours(1);
/// Cap on reverse-lookup file names surfaced in a finding's detail.
/// Keeps detail JSON size bounded when a broken blob is referenced
/// by hundreds of files.
const AFFECTED_FILES_SAMPLE: i64 = 5;
pub struct BlobsConsistencyCheck {
pool: Arc<PgPool>,
/// The default backend to probe when `args.storage` is `None` —
/// the currently-active LIVE backend, injected at DI time. Runs
/// with `?storage=<name>` build a fresh backend for the named
/// entry instead (via [`build_entry_backend`]).
backend: Arc<dyn BlobStorageBackend>,
/// Snapshot of `AppConfig.storage_entries` used to resolve
/// `args.storage` to a `NamedStorageEntry`. Empty for the
/// legacy zero-entries path — `?storage=<name>` runs then
/// fail-fast with a clear "no entries declared" message.
storage_entries: Vec<NamedStorageEntry>,
/// Ambient `AppConfig.storage_path` — used as the `root_dir`
/// fallback for a Local target entry with no `_ROOT_DIR`. Same
/// fallback rule the boot path uses.
storage_path_fallback: PathBuf,
/// The chunk-level page query, assembled once from the blob-reference
/// registry so this recompute and `dedup_gc` agree on what "referenced"
/// means. Built at construction rather than per page so the sweep runs a
@@ -166,7 +118,6 @@ fn chunk_page_sql(registry: &BlobReferenceRegistry) -> String {
b.hash AS hash,
b.size AS size,
b.ref_count AS ref_count,
b.created_at AS created_at,
({expected})::bigint AS actual_ref_count
FROM storage.blobs b
WHERE ($1::text IS NULL OR b.hash > $1)
@@ -176,18 +127,9 @@ fn chunk_page_sql(registry: &BlobReferenceRegistry) -> String {
}
impl BlobsConsistencyCheck {
pub fn new(
pool: Arc<PgPool>,
backend: Arc<dyn BlobStorageBackend>,
storage_entries: Vec<NamedStorageEntry>,
storage_path_fallback: PathBuf,
reference_registry: Arc<BlobReferenceRegistry>,
) -> Self {
pub fn new(pool: Arc<PgPool>, reference_registry: Arc<BlobReferenceRegistry>) -> Self {
Self {
pool,
backend,
storage_entries,
storage_path_fallback,
chunk_page_sql: chunk_page_sql(&reference_registry),
}
}
@@ -209,7 +151,6 @@ struct BlobRow {
hash: String,
size: i64,
ref_count: i32,
created_at: DateTime<Utc>,
/// Real reference count derived from the actual references —
/// files' whole-file `blob_hash` PLUS every chunk hash across
/// `storage.chunk_manifests`. Compared to `ref_count` (the
@@ -224,13 +165,12 @@ impl RecoverableJobHandler for BlobsConsistencyCheck {
}
fn description(&self) -> &'static str {
"Walks storage.blobs and checks each row against the reference- \
counting invariants dedup_gc relies on, and probes the backend \
once per row for missing bytes. That probe costs one backend \
call per blob — backend_consistency finds the same missing \
bytes in a single merge-join pass, and orphaned bytes too, so \
prefer it on large installs. Add ?deep=true to re-read and \
re-hash every blob for bit-rot; that is a full read of storage."
"Walks storage.blobs and reports rows whose ref_count disagrees \
with the references that actually exist. An under-count lets \
dedup_gc reap a blob that is still in use; an over-count pins \
one nothing needs. Database only — it never touches the storage \
backend, so it is cheap and safe to run at any time. Missing, \
orphaned or corrupted bytes are backend_consistency's job."
}
fn mutates(&self) -> Mutates {
@@ -272,78 +212,12 @@ impl RecoverableJobHandler for BlobsConsistencyCheck {
args: &JobRunArgs,
resume_cursor: Option<Vec<u8>>,
) -> RunOutcome {
// Resolve the backend to probe. Two paths, mirroring the
// Fresh/Resumed split the backend_migration handler uses:
// No backend is resolved here, and `?storage=<name>` is inert:
// this tenant reads nothing but the database. Everything physical
// — existence, orphan bytes, bit-rot — belongs to
// `backend_consistency`, which finds it in one enumeration pass
// instead of one probe per row.
//
// * Fresh + args.storage=Some — probe that named entry
// instead of the live backend. Stamp probed_storage in
// params so a mid-audit restart resumes against the same
// entry without re-input.
// * Fresh + args.storage=None — probe the live backend
// (today's default; audit of what the app is actually
// using).
// * Resumed — read probed_storage from params; None means
// the original run was against the live backend.
let is_fresh = resume_cursor.is_none();
let probed_storage: Option<String> = if is_fresh {
let name = args.storage.clone();
if let Some(n) = &name
&& let Err(e) = store.set_string_param(PROBED_STORAGE_PARAM, n).await
{
return RunOutcome::Failed {
message: format!("persist {PROBED_STORAGE_PARAM} to params: {e}"),
};
}
name
} else {
match store.get_string_param(PROBED_STORAGE_PARAM).await {
Ok(v) => v,
Err(e) => {
return RunOutcome::Failed {
message: format!("read {PROBED_STORAGE_PARAM} from params: {e}"),
};
}
}
};
let backend: Arc<dyn BlobStorageBackend> = match &probed_storage {
None => self.backend.clone(),
Some(name) => match self.storage_entries.iter().find(|e| &e.name == name) {
Some(entry) => build_entry_backend(entry, &self.storage_path_fallback),
None => {
let available = if self.storage_entries.is_empty() {
"(none)".to_string()
} else {
self.storage_entries
.iter()
.map(|e| e.name.as_str())
.collect::<Vec<_>>()
.join(", ")
};
return RunOutcome::Failed {
message: format!(
"storage entry `{name}` not found in OXICLOUD_STORAGE_ENTRIES. \
Available: [{available}]"
),
};
}
},
};
if let Err(e) = backend.initialize().await {
return RunOutcome::Failed {
message: format!("probed backend init: {e}"),
};
}
if let Some(name) = &probed_storage {
tracing::info!(
target: "audit",
event = "blobs_consistency.probe_scoped",
run_id = %store.run_id(),
probed_storage = %name,
"blobs_consistency probing entry `{name}` (via ?storage=<name>) instead of \
live backend"
);
}
// Snapshot "is this a Fresh run?" BEFORE the resume_cursor
// match consumes it — otherwise the `is_none()` check later
// borrows a partially-moved value. Fresh = no cursor bytes
@@ -375,60 +249,14 @@ impl RecoverableJobHandler for BlobsConsistencyCheck {
// `extra_stats` so operators see "found N, fixed M" in one line.
let mut repaired_count = 0u64;
// Deep mode is a per-run flag with two consumers:
// 1. This handler — decides whether to re-hash bytes.
// 2. The admin panel — needs to display whether the run
// was deep so operators know what the scan actually
// verified.
//
// On a Fresh run we take it from `deep` (the trigger
// endpoint stamps `?deep=true` onto the args) and stash it
// in `params.deep` so:
// * Resume picks up the same mode (would previously become
// non-deep on Resume — a Paused deep scan silently lost
// its `deep` intent).
// * The admin panel run-detail view can render
// `params.deep = "true"` alongside `target_name`,
// `progress_kind`, etc.
//
// Persist BEFORE the walk so a mid-fresh-batch crash still
// leaves a Paused row with the right mode marker.
let deep = if is_fresh {
let deep = args.deep;
let v = if deep { "true" } else { "false" };
if let Err(e) = store.set_string_param("deep", v).await {
return RunOutcome::Failed {
message: format!("failed to persist deep flag to params: {e}"),
};
}
deep
} else {
// Resumed run — read the persisted flag. Default to
// false (fast mode) if the row is a pre-K3.5 Paused
// scan without the param stashed.
match store.get_string_param("deep").await {
Ok(Some(v)) => v == "true",
Ok(None) => false,
Err(e) => {
return RunOutcome::Failed {
message: format!("read `deep` from params: {e}"),
};
}
}
};
// `?deep=true` is not handled here. Re-reading and re-hashing
// bytes is backend work end to end, so it moved to
// `backend_consistency`, where the merge-join already holds the
// matched key pairs worth verifying. A deep flag on this tenant
// would be a flag with nothing to do.
if deep {
tracing::info!(
target: "oxicloud::consistency",
event = "blobs_consistency.deep_mode_active",
run_id = %store.run_id(),
"deep mode: re-reading + re-hashing every blob (bit-rot detection)"
);
}
// Repair mode: same shape as `deep` above so the admin run-
// detail view can display `params.repair = "true"` alongside
// `params.deep`. Fresh persists what the trigger asked for;
// Repair mode persisted to `params.repair` so the admin run-detail
// view can display it. Fresh persists what the trigger asked for;
// Resume reads back so a paused repair scan stays a repair
// scan (a mid-scan crash mustn't silently downgrade to
// discovery-only for the remaining rows).
@@ -512,7 +340,6 @@ impl RecoverableJobHandler for BlobsConsistencyCheck {
finding_count = finding_count,
repaired_count = repaired_count,
repair_requested = repair,
deep = deep,
"blobs_consistency completed with {} finding(s), {} repaired",
finding_count,
repaired_count
@@ -523,12 +350,12 @@ impl RecoverableJobHandler for BlobsConsistencyCheck {
}));
}
let grace_cutoff = Utc::now() - CREATE_GRACE;
// No grace window here any more. It existed to keep the
// physical probe from flagging a blob whose bytes had landed
// but whose row hadn't — a write-path race this tenant no
// longer looks at. The refcount comparison reads one
// consistent DB snapshot, so there is nothing to wait for.
for row in &rows {
// (1) refcount_mismatch — content-safe check, cheap,
// always runs. Emitted BEFORE the physical probe so
// a broken-and-miscounted blob shows both findings.
if row.ref_count as i64 != row.actual_ref_count {
finding_count += 1;
let affected = affected_files(self.pool.as_ref(), &row.hash).await;
@@ -610,135 +437,6 @@ impl RecoverableJobHandler for BlobsConsistencyCheck {
}
}
}
// Skip physical probes for rows within the write
// grace window — writes-in-flight would false-positive.
if row.created_at > grace_cutoff {
continue;
}
// (2) blob_missing_from_backend — normal mode
// physical existence probe. Fails-open on backend
// error (log + skip): a transient S3 network blip
// shouldn't produce a flood of false data_loss
// findings.
let exists = match backend.blob_exists(&row.hash).await {
Ok(v) => v,
Err(e) => {
tracing::warn!(
target: "oxicloud::consistency",
event = "blobs_consistency.blob_exists_error",
run_id = %store.run_id(),
hash = %row.hash,
error = %e,
"blob_exists probe failed; skipping this row"
);
continue;
}
};
if !exists {
finding_count += 1;
let affected = affected_files(self.pool.as_ref(), &row.hash).await;
record_or_log(
store,
BLOBS_CONSISTENCY_JOB_NAME,
"blob_missing_from_backend",
"data_loss",
None,
serde_json::json!({
"hash": row.hash,
"size": row.size,
"ref_count": row.ref_count,
"affected_files": affected,
}),
)
.await;
// No point re-hashing bytes that aren't there.
continue;
}
// (3) blob_corrupted — DEEP MODE only. Read the
// whole blob, recompute BLAKE3, compare to the hash
// it's indexed under. Any mismatch = silent bit-rot.
//
// Finding fields:
// * `hash` — expected hash (the key the blob is
// indexed under in `storage.blobs`).
// * `computed_hash` — what BLAKE3 of the current
// bytes actually produces. Diagnostic: a
// one-bit flip vs a truncation vs a whole-file
// swap all leave distinctive signatures.
// `expected_hash` was NOT reused as a name to
// avoid mistaking it for "the hash we expect to
// see on disk (i.e. what will fix this)".
if deep {
match recompute_hash(backend.as_ref(), &row.hash).await {
Ok(computed_hash) if computed_hash == row.hash => {}
Ok(computed_hash) => {
finding_count += 1;
let affected = affected_files(self.pool.as_ref(), &row.hash).await;
record_or_log(
store,
BLOBS_CONSISTENCY_JOB_NAME,
"blob_corrupted",
"data_loss",
None,
serde_json::json!({
"hash": row.hash,
"computed_hash": computed_hash,
"size": row.size,
"ref_count": row.ref_count,
"affected_files": affected,
}),
)
.await;
}
Err(e) => {
// Blob can't be read at all — record as
// `blob_unreadable`. Distinct from
// `blob_corrupted` (hash mismatch = we
// can read but content differs): here
// we can't get bytes out to hash. Common
// causes: decrypt failure (missing key),
// network glitch on S3/Azure, missing
// file on Local, permission error.
//
// Recorded as `data_loss` because from
// the file's perspective the outcome is
// the same as corruption: content is
// inaccessible. Admins triage the error
// string to distinguish transient
// (retry-safe) from permanent (needs
// key recovery or blob replacement).
finding_count += 1;
let affected = affected_files(self.pool.as_ref(), &row.hash).await;
record_or_log(
store,
BLOBS_CONSISTENCY_JOB_NAME,
"blob_unreadable",
"data_loss",
None,
serde_json::json!({
"hash": row.hash,
"size": row.size,
"ref_count": row.ref_count,
"affected_files": affected,
"error": e.to_string(),
}),
)
.await;
tracing::warn!(
target: "oxicloud::consistency",
event = "blobs_consistency.blob_unreadable",
run_id = %store.run_id(),
hash = %row.hash,
error = %e,
"🚨 blob unreadable in deep mode — recorded finding, continuing"
);
}
}
}
}
// Advance cursor + checkpoint.
@@ -759,7 +457,6 @@ impl RecoverableJobHandler for BlobsConsistencyCheck {
finding_count = finding_count,
repaired_count = repaired_count,
repair_requested = repair,
deep = deep,
"blobs_consistency completed with {} finding(s), {} repaired",
finding_count,
repaired_count
@@ -773,69 +470,6 @@ impl RecoverableJobHandler for BlobsConsistencyCheck {
}
}
/// Sample of file names that reference this blob — either directly
/// (`files.blob_hash = $hash`, legacy pre-CDC) or transitively via a
/// manifest (`chunk_hashes @> ARRAY[$hash]`, post-CDC dominant path).
/// Capped so a chunk shared by 10 000 files doesn't blow up the
/// finding detail JSON. Order is arbitrary — sampling for
/// diagnosis, not enumeration.
async fn affected_files(pool: &PgPool, hash: &str) -> Vec<String> {
let rows: Vec<(String,)> = sqlx::query_as(
r#"
SELECT DISTINCT f.name
FROM storage.files f
WHERE f.blob_hash = $1
OR EXISTS (
SELECT 1 FROM storage.chunk_manifests m
WHERE m.file_hash = f.blob_hash
AND $1 = ANY(m.chunk_hashes)
)
LIMIT $2
"#,
)
.bind(hash)
.bind(AFFECTED_FILES_SAMPLE)
.fetch_all(pool)
.await
.unwrap_or_default();
rows.into_iter().map(|(n,)| n).collect()
}
/// Deep-mode helper — read the blob from the backend and recompute
/// its BLAKE3 hash. Returns `Ok(true)` when the recomputed hash
/// matches `expected_hash` (byte for byte), `Ok(false)` on mismatch
/// (bit-rot), `Err(_)` on any backend-side error (network blip,
/// permission issue) — callers log-and-skip errors since a transient
/// failure isn't a corruption signal.
/// Deep-mode helper — read the blob from the backend and recompute
/// its BLAKE3 hash. Returns the recomputed hex string; callers
/// compare against the expected hash themselves. Returning the
/// actual hash (not just a bool) lets the finding surface WHAT the
/// bytes now hash to, which is diagnostic gold: a specific one-bit
/// flip has a very different signature from a chunk-boundary
/// corruption or a truncated read. `Err(_)` on backend-side error
/// (network blip, permission issue) — callers log-and-skip since
/// transient failure isn't a corruption signal.
async fn recompute_hash(
backend: &dyn BlobStorageBackend,
expected_hash: &str,
) -> Result<String, crate::common::errors::DomainError> {
use crate::common::errors::DomainError;
use futures::StreamExt;
let mut stream = backend.get_blob_stream(expected_hash).await?;
let mut hasher = blake3::Hasher::new();
while let Some(chunk) = stream.next().await {
let bytes = chunk.map_err(|e| {
DomainError::internal_error("BlobsConsistency", format!("stream read: {e}"))
})?;
hasher.update(&bytes);
}
Ok(hasher.finalize().to_hex().to_string())
}
#[cfg(test)]
mod tests {
use super::*;
@@ -865,7 +499,6 @@ mod tests {
b.hash AS hash,
b.size AS size,
b.ref_count AS ref_count,
b.created_at AS created_at,
((SELECT COUNT(*) FROM storage.files cnt_f
WHERE cnt_f.blob_hash = b.hash
AND NOT EXISTS (
src/infrastructure/services/mod.rs
+1
View File
@@ -3,6 +3,7 @@ pub mod azure_blob_backend;
pub mod backend_consistency_service;
pub mod backend_migration_service;
pub mod backend_rotate_service;
pub mod blob_diagnostics;
pub mod blobs_consistency_service;
pub mod cached_blob_backend;
pub mod chunked_upload_service;