diff --git a/docs/architecture/backend-storage.md b/docs/architecture/backend-storage.md index a7b1d1b1..544ac22f 100644 --- a/docs/architecture/backend-storage.md +++ b/docs/architecture/backend-storage.md @@ -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=` 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=` 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 diff --git a/docs/config/admin-settings.md b/docs/config/admin-settings.md index 589eac3a..e80b6820 100644 --- a/docs/config/admin-settings.md +++ b/docs/config/admin-settings.md @@ -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=` 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=` 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= +POST /api/admin/jobs/backend_consistency/trigger?storage= ``` +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=`: it only reads the database, so there is no entry for it to scope. + Unknown names 400 at the HTTP layer. ## Data Storage diff --git a/docs/plan/derived-blobs.md b/docs/plan/derived-blobs.md index 6ffe378e..d959d663 100644 --- a/docs/plan/derived-blobs.md +++ b/docs/plan/derived-blobs.md @@ -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 | diff --git a/docs/plan/storage-key-rotation.md b/docs/plan/storage-key-rotation.md index 7b908f3b..2f4a18df 100644 --- a/docs/plan/storage-key-rotation.md +++ b/docs/plan/storage-key-rotation.md @@ -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 diff --git a/docs/plan/storage-multi-entry.md b/docs/plan/storage-multi-entry.md index 00c91046..c10c429c 100644 --- a/docs/plan/storage-multi-entry.md +++ b/docs/plan/storage-multi-entry.md @@ -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=` on blobs_consistency**: run against `s3_prod` before +7. **`?storage=` 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 diff --git a/frontend/src/lib/components/AdminJobsPanel.svelte b/frontend/src/lib/components/AdminJobsPanel.svelte index e7e0a778..60fee78d 100644 --- a/frontend/src/lib/components/AdminJobsPanel.svelte +++ b/frontend/src/lib/components/AdminJobsPanel.svelte @@ -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 diff --git a/src/common/di.rs b/src/common/di.rs index 92b1b572..dd422dc8 100644 --- a/src/common/di.rs +++ b/src/common/di.rs @@ -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(), diff --git a/src/infrastructure/services/backend_consistency_service.rs b/src/infrastructure/services/backend_consistency_service.rs index 7b5c462f..2b94f819 100644 --- a/src/infrastructure/services/backend_consistency_service.rs +++ b/src/infrastructure/services/backend_consistency_service.rs @@ -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 { + 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()) +} diff --git a/src/infrastructure/services/blob_diagnostics.rs b/src/infrastructure/services/blob_diagnostics.rs new file mode 100644 index 00000000..a5639910 --- /dev/null +++ b/src/infrastructure/services/blob_diagnostics.rs @@ -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 { + 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() +} diff --git a/src/infrastructure/services/blobs_consistency_service.rs b/src/infrastructure/services/blobs_consistency_service.rs index 8fe139b2..6f4d49b0 100644 --- a/src/infrastructure/services/blobs_consistency_service.rs +++ b/src/infrastructure/services/blobs_consistency_service.rs @@ -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=` 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, - /// The default backend to probe when `args.storage` is `None` — - /// the currently-active LIVE backend, injected at DI time. Runs - /// with `?storage=` build a fresh backend for the named - /// entry instead (via [`build_entry_backend`]). - backend: Arc, - /// Snapshot of `AppConfig.storage_entries` used to resolve - /// `args.storage` to a `NamedStorageEntry`. Empty for the - /// legacy zero-entries path — `?storage=` runs then - /// fail-fast with a clear "no entries declared" message. - storage_entries: Vec, - /// 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, - backend: Arc, - storage_entries: Vec, - storage_path_fallback: PathBuf, - reference_registry: Arc, - ) -> Self { + pub fn new(pool: Arc, reference_registry: Arc) -> 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, /// 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>, ) -> 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=` 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 = 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 = 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::>() - .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=) 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 { - 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 { - 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 ( diff --git a/src/infrastructure/services/mod.rs b/src/infrastructure/services/mod.rs index 5ad26293..4b75ac4c 100644 --- a/src/infrastructure/services/mod.rs +++ b/src/infrastructure/services/mod.rs @@ -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;