diff --git a/src/application/ports/blob_storage_ports.rs b/src/application/ports/blob_storage_ports.rs index ae8732d3..7f62d750 100644 --- a/src/application/ports/blob_storage_ports.rs +++ b/src/application/ports/blob_storage_ports.rs @@ -220,6 +220,34 @@ pub trait BlobStorageBackend: Send + Sync + 'static { /// return `None`; callers that need a local file must stream + spool. fn local_blob_path(&self, hash: &str) -> Option; + /// The same storage with any read-through cache peeled off, or + /// `None` when this backend is not a cache. + /// + /// **For verification only** — normal reads must keep going through + /// the cache, which is the point of having one. + /// + /// A cache answers reads from its own copy, so re-hashing through + /// one checks the cache rather than storage: rot on the remote is + /// hidden by a good cached copy, and rot in the cache is blamed on a + /// healthy remote. The second is worse, because it sends an operator + /// to the wrong layer. `backend_consistency ?deep=true` is the only + /// caller. + /// + /// Peels **only** the cache. The cache sits outside the encryption + /// decorator and stores plaintext, while the content hash is over + /// plaintext, so unwrapping further would hand back ciphertext and + /// fail every blob it checked. + /// + /// Implement by returning the inner backend. Pass-through wrappers + /// (hot-swap, retry) forward to whatever they wrap, so the unwrap + /// still reaches the cache — and, for hot-swap, resolves through + /// `current()` so it survives a migration cutover rather than + /// pinning the pre-cutover storage. Everything else inherits the + /// `None` default and is used as-is. + fn uncached(&self) -> Option> { + None + } + /// How many chunk fetches the CDC reader may run concurrently when /// reassembling a file (`read_blob_stream`'s `buffered(N)` read-ahead). /// diff --git a/src/infrastructure/services/backend_consistency_service.rs b/src/infrastructure/services/backend_consistency_service.rs index 38204047..01cd25ac 100644 --- a/src/infrastructure/services/backend_consistency_service.rs +++ b/src/infrastructure/services/backend_consistency_service.rs @@ -352,11 +352,33 @@ impl RecoverableJobHandler for BackendConsistencyCheck { // actually verified. let deep = args.get_bool("deep"); + // Verify through storage, never through a read-through cache. + // + // A cache answers from its own copy, so re-hashing through one + // checks the CACHE: rot on the remote is masked by a good cached + // copy, and rot in the cache is recorded as `blob_corrupted` + // against a healthy remote — sending an operator to the wrong + // layer. The finding names `backend.backend_type()`, so that + // attribution has to be true. + // + // Only the cache is peeled; the decryptor stays, because the + // cache holds plaintext and the content hash is over plaintext. + // `?storage=` already builds an uncached stack, so this + // only changes the live-backend path — which is the one that was + // silently fast. + let verify_backend = backend.uncached().unwrap_or_else(|| backend.clone()); + // Counter, not just a flag: a deep run that verified nothing and + // a deep run that verified everything are otherwise + // indistinguishable in the outcome, which is exactly the + // ambiguity that made a 1.5s "deep" sweep over 2022 chunks look + // plausible. + let mut verified_count = 0u64; if deep { tracing::info!( target: "oxicloud::consistency", event = "backend_consistency.deep_mode_active", run_id = %store.run_id(), + cached_read_bypassed = backend.uncached().is_some(), "deep mode: re-reading + re-hashing every matched blob (bit-rot detection)" ); } @@ -477,10 +499,14 @@ impl RecoverableJobHandler for BackendConsistencyCheck { event = "backend_consistency.completed", run_id = %store.run_id(), finding_count = finding_count, + verified = verified_count, "backend_consistency completed with {} finding(s)", finding_count ); - return RunOutcome::completed(); + return RunOutcome::completed_with(serde_json::json!({ + "deep": deep, + "verified": verified_count, + })); } // ── Merge-join, not a one-sided probe ──────────────── @@ -562,8 +588,10 @@ impl RecoverableJobHandler for BackendConsistencyCheck { // 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; + verified_count += 1; + finding_count += self + .verify_bytes(store, verify_backend.as_ref(), &b.hash) + .await; } bi.next(); di.next(); @@ -668,10 +696,18 @@ impl RecoverableJobHandler for BackendConsistencyCheck { event = "backend_consistency.completed", run_id = %store.run_id(), finding_count = finding_count, + verified = verified_count, "backend_consistency completed with {} finding(s)", finding_count ); - return RunOutcome::completed(); + // `verified` is reported on EVERY run, zero included: + // absent-vs-zero is exactly the distinction an operator + // needs, and omitting it on a shallow run would make + // "deep verified nothing" look like "this was shallow". + return RunOutcome::completed_with(serde_json::json!({ + "deep": deep, + "verified": verified_count, + })); } } } diff --git a/src/infrastructure/services/cached_blob_backend.rs b/src/infrastructure/services/cached_blob_backend.rs index b9ee0d95..87f23819 100644 --- a/src/infrastructure/services/cached_blob_backend.rs +++ b/src/infrastructure/services/cached_blob_backend.rs @@ -413,6 +413,14 @@ impl BlobStorageBackend for CachedBlobBackend { if path.exists() { Some(path) } else { None } } + /// This decorator IS the cache, so peeling it yields the real + /// storage. See [`BlobStorageBackend::uncached`] for why an + /// integrity check must not read through here — every other caller + /// keeps using the cache. + fn uncached(&self) -> Option> { + Some(self.inner.clone()) + } + /// Enumeration MUST delegate to the primary (inner) backend, not /// the local cache. The cache is by definition a subset (only /// recently-accessed blobs); walking the cache would look like diff --git a/src/infrastructure/services/retry_blob_backend.rs b/src/infrastructure/services/retry_blob_backend.rs index e06dcc30..0141a36f 100644 --- a/src/infrastructure/services/retry_blob_backend.rs +++ b/src/infrastructure/services/retry_blob_backend.rs @@ -356,6 +356,14 @@ impl BlobStorageBackend for RetryBlobBackend { self.inner.local_blob_path(hash) } + /// Pass-through wrapper — forward, so an unwrap started above still + /// reaches the cache. Inheriting the `None` default would silently + /// end the search at this layer and leave verification reading + /// through the cache after all. + fn uncached(&self) -> Option> { + self.inner.uncached() + } + /// Enumeration delegates to inner. Retry semantics apply per /// call, not per batch — a single list call that fails after /// exhausting retries surfaces the error to the tenant, which diff --git a/src/infrastructure/services/swappable_blob_backend.rs b/src/infrastructure/services/swappable_blob_backend.rs index e28abadb..8e8926c4 100644 --- a/src/infrastructure/services/swappable_blob_backend.rs +++ b/src/infrastructure/services/swappable_blob_backend.rs @@ -205,6 +205,15 @@ impl BlobStorageBackend for SwappableBlobBackend { self.current().local_blob_path(hash) } + /// Resolved through `current()`, not captured once: this wrapper sits + /// OUTSIDE the cache, so a migration cutover replaces the entire + /// cached stack beneath it. A handle taken at DI time would keep + /// pointing at the pre-cutover storage and audit the backend that + /// was just migrated away from. + fn uncached(&self) -> Option> { + self.current().uncached() + } + fn read_prefetch(&self) -> usize { self.current().read_prefetch() }