From 44356b14f2c5c0c97dec91ca41a6957e6a2b332a Mon Sep 17 00:00:00 2001 From: Edouard Vanbelle Date: Wed, 29 Jul 2026 22:25:05 +0200 Subject: [PATCH] feat(recoverable-job): add blobs_consistency job --- src/common/di.rs | 35 ++ .../services/blobs_consistency_service.rs | 455 ++++++++++++++++++ src/infrastructure/services/mod.rs | 1 + 3 files changed, 491 insertions(+) create mode 100644 src/infrastructure/services/blobs_consistency_service.rs diff --git a/src/common/di.rs b/src/common/di.rs index 3d413f16..161078c1 100644 --- a/src/common/di.rs +++ b/src/common/di.rs @@ -322,6 +322,13 @@ impl AppServiceFactory { let blob_lifecycle = Arc::new(BlobLifecycleService::new().with_hook(thumbnail_service.clone())); + // Hold a clone of the fully-decorated blob backend for use by + // `blobs_consistency` (physical-existence + bit-rot probes) + // further down the DI chain. Must be captured BEFORE the + // `dedup_service` construction below because that call moves + // `blob_backend` into DedupService. + let blob_backend_for_consistency = blob_backend.clone(); + // Deduplication service — PRIMARY blob storage engine (PostgreSQL-backed index) let dedup_service = Arc::new( crate::infrastructure::services::dedup_service::DedupService::new( @@ -459,6 +466,7 @@ impl AppServiceFactory { config: self.config.clone(), job_registry, job_store_provider, + blob_backend: blob_backend_for_consistency, }) } @@ -1319,6 +1327,27 @@ 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. + let _ = Arc::new( + crate::infrastructure::services::blobs_consistency_service::BlobsConsistencyCheck::new( + maintenance_pool.clone(), + core.blob_backend.clone(), + ), + ) + .register_recoverable_job(&core.job_registry, &job_store_provider_dyn) + .await; + // "Run all consistency checks" coordinator. Plain JobHandler // (not RecoverableJobHandler) — it dispatches, doesn't scan. // MUST register AFTER every `*_consistency` tenant so the @@ -2278,6 +2307,12 @@ pub struct CoreServices { /// Boot-time crash-recovery sweep is run in `build_app_state` right /// after this provider is created. pub job_store_provider: Arc, + /// Fully-decorated blob backend (retry → encryption → cache + /// stack applied). Exposed here so tenants outside + /// `create_core_services` — notably `blobs_consistency` in + /// `build_app_state` — can probe `blob_exists()` / re-hash bytes + /// through the same stack DedupService uses. + pub blob_backend: Arc, } /// Container for repository services diff --git a/src/infrastructure/services/blobs_consistency_service.rs b/src/infrastructure/services/blobs_consistency_service.rs new file mode 100644 index 00000000..fd2332e9 --- /dev/null +++ b/src/infrastructure/services/blobs_consistency_service.rs @@ -0,0 +1,455 @@ +//! 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. +//! +//! Three per-row checks (subject-iteration principle in action — +//! one walk, multiple branches): +//! +//! * `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. +//! +//! * `refcount_mismatch` (severity `inconsistent`) — +//! `storage.blobs.ref_count` disagrees with the actual reference +//! count computed from `storage.files.blob_hash` + +//! `storage.chunk_manifests.chunk_hashes[]`. Under-count means +//! dedup GC could prematurely reap a live blob; over-count means +//! 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` +//! +//! `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. +//! +//! ### Not covered here +//! +//! * **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. +//! * **Manifest-level integrity** (`storage.chunk_manifests` rows +//! pointing at reaped chunks) — already covered by +//! `files_consistency::chunk_missing`. + +use std::sync::Arc; + +use async_trait::async_trait; +use chrono::{DateTime, Duration, Utc}; +use sqlx::PgPool; + +use crate::application::ports::blob_storage_ports::BlobStorageBackend; +use crate::infrastructure::scheduler::{ + JobRegistry, JobRunArgs, JobStore, JobStoreProvider, RecoverableJobHandler, RunOutcome, + RunStatus, record_or_log, +}; + +pub const BLOBS_CONSISTENCY_JOB_NAME: &str = "blobs_consistency"; + +/// 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. +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, + backend: Arc, +} + +impl BlobsConsistencyCheck { + pub fn new(pool: Arc, backend: Arc) -> Self { + Self { pool, backend } + } + + pub async fn register_recoverable_job( + self: Arc, + registry: &JobRegistry, + provider: &Arc, + ) -> Arc { + registry + .register_recoverable_job(self.clone(), provider.clone(), None) + .await; + self + } +} + +#[derive(Debug, sqlx::FromRow)] +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 + /// stored counter) to detect drift. + actual_ref_count: i64, +} + +#[async_trait] +impl RecoverableJobHandler for BlobsConsistencyCheck { + fn name(&self) -> &str { + BLOBS_CONSISTENCY_JOB_NAME + } + + /// Definitive count. `storage.blobs` PK scan is index-only; + /// even at millions of rows it's sub-second on modern PG. + async fn count_total(&self) -> Option { + let row: Result<(i64,), sqlx::Error> = sqlx::query_as("SELECT COUNT(*) FROM storage.blobs") + .fetch_one(self.pool.as_ref()) + .await; + match row { + Ok((n,)) => Some(n.max(0) as u64), + Err(e) => { + tracing::debug!( + target: "oxicloud::consistency", + event = "blobs_consistency.count_total_failed", + error = %e, + "count_total failed — run will not surface a progress bar" + ); + None + } + } + } + + async fn run_resumable( + &self, + store: &dyn JobStore, + args: &JobRunArgs, + resume_cursor: Option>, + ) -> RunOutcome { + // Cursor = the last-visited `hash` string, UTF-8-encoded. On + // resume, we walk `WHERE hash > $cursor` in ASC order. First + // batch: NULL cursor → start from the smallest hash. + let mut cursor: Option = match resume_cursor { + None => None, + Some(bytes) if bytes.is_empty() => None, + Some(bytes) => match String::from_utf8(bytes) { + Ok(s) => Some(s), + Err(e) => { + return RunOutcome::Failed { + message: format!("invalid cursor: not valid UTF-8: {e}"), + }; + } + }, + }; + + // Per-run finding counter (feeds outer JobOutcome extras via + // stats.finding_count — actual persistence happens in + // `record_finding` on each emission). + let mut finding_count = 0u64; + + // Deep mode = re-hash bytes for bit-rot detection. Logged + // once at run start so operators tailing tracing know why the + // scan is taking hours. + if args.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)" + ); + } + + loop { + // Cooperative cancel poll between batches. + match store.status().await { + Ok(RunStatus::CancelRequested) => { + tracing::info!( + target: "oxicloud::consistency", + event = "blobs_consistency.cancelled", + run_id = %store.run_id(), + finding_count = finding_count, + "blobs_consistency cancelled cooperatively, pausing" + ); + return RunOutcome::Paused { + cursor: cursor + .as_ref() + .map(|s| s.as_bytes().to_vec()) + .unwrap_or_default(), + }; + } + Ok(_) => {} + Err(e) => { + return RunOutcome::Failed { + message: format!("status poll: {e}"), + }; + } + } + + // Fetch the next batch. Per-row `actual_ref_count` + // computed inline via correlated subqueries — one for + // legacy whole-file references (`files.blob_hash`), one + // for CDC chunk references (`chunk_manifests.chunk_hashes`). + // GIN index on `chunk_hashes` (migration + // 20260628000000_delta_upload_gin_index) makes the + // `= ANY(chunk_hashes)` probe cheap. + let rows: Vec = match sqlx::query_as( + r#" + SELECT + 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 f + WHERE f.blob_hash = b.hash) + + (SELECT COUNT(*) FROM storage.chunk_manifests m + WHERE b.hash = ANY(m.chunk_hashes)) + )::bigint AS actual_ref_count + FROM storage.blobs b + WHERE ($1::text IS NULL OR b.hash > $1) + ORDER BY b.hash + LIMIT $2 + "#, + ) + .bind(cursor.as_deref()) + .bind(BATCH_SIZE) + .fetch_all(self.pool.as_ref()) + .await + { + Ok(r) => r, + Err(e) => { + return RunOutcome::Failed { + message: format!("batch fetch: {e}"), + }; + } + }; + + if rows.is_empty() { + tracing::info!( + target: "oxicloud::consistency", + event = "blobs_consistency.completed", + run_id = %store.run_id(), + finding_count = finding_count, + deep = args.deep, + "blobs_consistency completed with {} finding(s)", + finding_count + ); + return RunOutcome::Completed; + } + + let grace_cutoff = Utc::now() - CREATE_GRACE; + + 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; + record_or_log( + store, + BLOBS_CONSISTENCY_JOB_NAME, + "refcount_mismatch", + "inconsistent", + None, // hash isn't a UUID; resource identifier lives in detail + serde_json::json!({ + "hash": row.hash, + "stored": row.ref_count, + "actual": row.actual_ref_count, + "delta": row.actual_ref_count - row.ref_count as i64, + "size": row.size, + "affected_files": affected, + }), + ) + .await; + } + + // 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 self.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. + if args.deep { + match verify_hash(self.backend.as_ref(), &row.hash).await { + Ok(true) => {} + Ok(false) => { + 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, + "size": row.size, + "ref_count": row.ref_count, + "affected_files": affected, + }), + ) + .await; + } + Err(e) => { + tracing::warn!( + target: "oxicloud::consistency", + event = "blobs_consistency.verify_hash_error", + run_id = %store.run_id(), + hash = %row.hash, + error = %e, + "verify_hash failed; not a corruption signal on its own" + ); + } + } + } + } + + // Advance cursor + checkpoint. + let last_hash = rows.last().map(|r| r.hash.clone()).expect("non-empty rows"); + cursor = Some(last_hash.clone()); + let batch_len = rows.len() as u64; + if let Err(e) = store + .checkpoint(last_hash.into_bytes(), batch_len) + .await + { + return RunOutcome::Failed { + message: format!("checkpoint: {e}"), + }; + } + + if (rows.len() as i64) < BATCH_SIZE { + tracing::info!( + target: "oxicloud::consistency", + event = "blobs_consistency.completed", + run_id = %store.run_id(), + finding_count = finding_count, + deep = args.deep, + "blobs_consistency completed with {} finding(s)", + finding_count + ); + return RunOutcome::Completed; + } + } + } +} + +/// 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. +async fn verify_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); + } + + let actual = hasher.finalize().to_hex().to_string(); + Ok(actual == expected_hash) +} diff --git a/src/infrastructure/services/mod.rs b/src/infrastructure/services/mod.rs index ee359a9d..af79a2d0 100644 --- a/src/infrastructure/services/mod.rs +++ b/src/infrastructure/services/mod.rs @@ -1,5 +1,6 @@ pub mod audio_metadata_service; pub mod azure_blob_backend; +pub mod blobs_consistency_service; pub mod cached_blob_backend; pub mod chunked_upload_service; pub mod compression_service;