diff --git a/src/application/ports/blob_storage_ports.rs b/src/application/ports/blob_storage_ports.rs index 91f7d428..29fa87fa 100644 --- a/src/application/ports/blob_storage_ports.rs +++ b/src/application/ports/blob_storage_ports.rs @@ -10,6 +10,7 @@ //! and PostgreSQL index logic in `DedupService` itself. use bytes::Bytes; +use chrono::{DateTime, Utc}; use futures::Stream; use serde::Serialize; use std::future::Future; @@ -18,6 +19,53 @@ use std::pin::Pin; use crate::domain::errors::DomainError; +/// One row returned by [`BlobStorageBackend::list_blob_hashes`] — the +/// hash of a blob physically present on the backend, plus its +/// last-modified timestamp when the backend can supply one. `mtime` +/// is used by `backend_consistency` to skip freshly-created files +/// still within the write grace window (avoids false-positive +/// orphans during the durability-before-visibility window that +/// `dedup_service` opens). +#[derive(Debug, Clone)] +pub struct BackendBlobEntry { + pub hash: String, + /// `None` when the backend doesn't track mtime — the consistency + /// scan then falls back to treating the entry as "old enough" and + /// will emit an orphan finding without a grace check. + pub mtime: Option>, +} + +/// A file present in the blob-storage namespace but NOT matching the +/// canonical `<64-hex>.blob` shape. Sidecars (`.blob.orig`, +/// `.blob.lost`, `.blob.tmp`), wrong extensions, non-hex names — +/// anything the enumeration filter skips for the blob list. Surfaced +/// so `backend_consistency` can emit them as `anomaly` notices +/// (informational only — they don't hurt anything but the operator +/// should know they're there). +#[derive(Debug, Clone)] +pub struct BackendUnknownEntry { + /// Backend-relative path (`04/04f48c...blob.orig` on local FS or + /// as an S3 key). Included in the finding detail so operators + /// can locate it. + pub path: String, + pub mtime: Option>, +} + +/// Return type of [`BlobStorageBackend::list_blob_hashes`] — one +/// batch of the enumeration. Struct (not tuple) so adding future +/// per-batch metadata (e.g. `truncated: bool`) doesn't break every +/// backend impl. `next_cursor = None` signals end of enumeration. +/// +/// Backends that don't track sidecar/unknown files leave `unknowns` +/// empty; the tenant just doesn't emit any `unknown_backend_file` +/// notices from that batch. +#[derive(Debug, Clone)] +pub struct BlobListPage { + pub blobs: Vec, + pub unknowns: Vec, + pub next_cursor: Option, +} + /// Boxed future alias used by [`BlobStorageBackend`] to keep the trait dyn-compatible. type BoxFut<'a, T> = Pin + Send + 'a>>; @@ -145,4 +193,40 @@ pub trait BlobStorageBackend: Send + Sync + 'static { fn read_prefetch(&self) -> usize { 1 } + + /// Enumerate blob entries physically present on this backend, in + /// implementation-defined order — cursor-based paging. + /// + /// * `cursor` — opaque continuation token from a prior call, or + /// `None` to start from the beginning. Format is per-backend + /// (local = last path visited; S3 = continuation token; Azure + /// = list marker); callers treat it as opaque. + /// * `limit` — soft cap on batch size; backends may return + /// fewer (e.g. end of a shard directory). + /// + /// Returns `(entries, next_cursor)`. `next_cursor = None` means + /// enumeration is complete. Each `BackendBlobEntry` carries the + /// hash + optional mtime for grace-window filtering. + /// + /// The trait default returns + /// [`DomainError::NotSupported`](DomainError::not_supported) + /// — future backends that genuinely can't enumerate (some + /// write-only queue, some read-only mirror) can inherit it. All + /// currently-shipped backends (local, S3, Azure) override. + /// + /// Filtering out non-blob artifacts (temp files, `.corrupt` / + /// `.lost` sidecars, encryption metadata) is the backend's + /// responsibility — the tenant walks whatever this returns. + fn list_blob_hashes( + &self, + _cursor: Option, + _limit: usize, + ) -> BoxFut<'_, Result> { + Box::pin(async { + Err(DomainError::operation_not_supported( + "list_blob_hashes", + "this backend does not implement enumeration", + )) + }) + } } diff --git a/src/common/di.rs b/src/common/di.rs index 161078c1..be862c6f 100644 --- a/src/common/di.rs +++ b/src/common/di.rs @@ -1348,6 +1348,24 @@ impl AppServiceFactory { .register_recoverable_job(&core.job_registry, &job_store_provider_dyn) .await; + // Fifth recoverable-run tenant. Iterates the storage + // backend (via `BlobStorageBackend::list_blob_hashes`) and + // reports every physical blob that has no matching row in + // `storage.blobs`. Closes the reference graph together with + // `blobs_consistency`: this tenant walks backend→DB, that + // one walks DB→backend. Enumeration is backend-specific but + // the tenant is fully backend-agnostic — each backend owns + // its own layout knowledge (local walks `.blobs/`, S3 uses + // ListObjectsV2, migration wrapper refuses mid-migration). + let _ = Arc::new( + crate::infrastructure::services::backend_consistency_service::BackendConsistencyCheck::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 diff --git a/src/infrastructure/services/azure_blob_backend.rs b/src/infrastructure/services/azure_blob_backend.rs index 353a5aa3..3be87330 100644 --- a/src/infrastructure/services/azure_blob_backend.rs +++ b/src/infrastructure/services/azure_blob_backend.rs @@ -389,4 +389,14 @@ impl BlobStorageBackend for AzureBlobBackend { fn local_blob_path(&self, _hash: &str) -> Option { None } + + // TODO: implement `list_blob_hashes` via + // `container_client.list_blobs()` (`azure_storage_blobs` + // paginator). Same filter as local + S3 impls: + // `/<64-hex>.blob` naming. Currently inherits the trait + // default which returns `operation_not_supported` — the + // `backend_consistency` tenant handles that by emitting a + // single run-level `backend_unenumerable` finding and + // completing without per-blob probes. Ship as a follow-up once + // there's an Azure test environment to validate against. } diff --git a/src/infrastructure/services/backend_consistency_service.rs b/src/infrastructure/services/backend_consistency_service.rs new file mode 100644 index 00000000..71d5d8a7 --- /dev/null +++ b/src/infrastructure/services/backend_consistency_service.rs @@ -0,0 +1,369 @@ +//! Fifth tenant of Part 2 (recoverable-run engine). +//! +//! Iterates the storage backend's blob-enumeration surface and +//! reports every blob physically present on the backend that has NO +//! matching row in `storage.blobs`. Complements +//! `blobs_consistency` (which walks the DB and probes the backend): +//! together they close the reference graph. +//! +//! ### Per-row check +//! +//! * `orphan_blob` (severity `inconsistent`) — bytes on disk / S3 / +//! Azure with no registry row. Not data-loss (nothing broken — +//! just storage overhead), but points at dedup_gc or +//! ingest-path drift. Recovery = register-registry-row (if the +//! bytes are still needed) OR delete the file (if truly orphan). +//! +//! ### Run-level check +//! +//! * `backend_unenumerable` (severity `anomaly`) — the backend +//! returned `operation_not_supported` on the first +//! `list_blob_hashes` call. Currently this fires when a +//! `MigrationBlobBackend` is active (refuses enumeration +//! mid-migration by design) or on an Azure backend (Azure impl +//! deferred). Informational — operators know they can't rely on +//! this scan under that config. +//! +//! ### Grace window +//! +//! Skip orphans whose backend mtime is within the last hour. Same +//! shape as `blobs_consistency` + `dedup_gc`: matches the +//! durability-before-visibility gap in the write path. +//! +//! ### Cost profile +//! +//! Batched: fetch N hashes from the backend, do one +//! `WHERE hash = ANY($1)` DB probe per batch, set-difference in +//! Rust. Dedup savings: yes — a chunk shared by 5 files still +//! walks once. On local backend the walk is +//! `walkdir + fs::metadata` per file (fast). On S3 the walk is +//! `ListObjectsV2` (rate-limited but paginated). Progress bar +//! uses `COUNT(*) FROM storage.blobs` as the approximate +//! denominator (backend count ≈ blob count on a healthy install; +//! deviation IS the finding). + +use std::collections::HashSet; +use std::sync::Arc; + +use async_trait::async_trait; +use chrono::{Duration, Utc}; +use sqlx::PgPool; + +use crate::application::ports::blob_storage_ports::BlobStorageBackend; +use crate::infrastructure::scheduler::{ + JobRegistry, JobRunArgs, JobStore, JobStoreProvider, ProgressKind, RecoverableJobHandler, + RunOutcome, RunStatus, record_or_log, +}; + +pub const BACKEND_CONSISTENCY_JOB_NAME: &str = "backend_consistency"; + +/// Batch size for backend enumeration + DB probe. 500 is enough to +/// amortise the DB round-trip while keeping the cancel-poll cadence +/// sub-second (each batch = one backend list + one DB probe + Rust +/// set-difference). Larger batches on S3 hit ListObjectsV2's +/// per-request limit (1000) with wasted rows filtered client-side; +/// smaller batches over-poll the DB. +const BATCH_SIZE: usize = 500; + +/// Grace window — orphans younger than this are skipped, since the +/// write path is durability-before-visibility: bytes hit disk before +/// the `storage.blobs` row is inserted. A scan catching a blob +/// mid-write would false-positive it as orphan. Matches +/// `blobs_consistency` + `dedup_gc`. +const CREATE_GRACE: Duration = Duration::hours(1); + +/// Cap on affected-blob examples surfaced in the run-level +/// `backend_unenumerable` finding. Keeps the finding detail bounded. +const _MAX_EXAMPLES: usize = 5; + +pub struct BackendConsistencyCheck { + pool: Arc, + backend: Arc, +} + +impl BackendConsistencyCheck { + 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 + } +} + +#[async_trait] +impl RecoverableJobHandler for BackendConsistencyCheck { + fn name(&self) -> &str { + BACKEND_CONSISTENCY_JOB_NAME + } + + /// Approximate total: on a healthy install every backend blob + /// has a `storage.blobs` row, so the DB count is a proxy for + /// the backend count. The fraction deviating from 1.0 at run + /// end IS informative — a fraction of 1.05 means the backend + /// holds ~5% orphan bytes, which is exactly what this check + /// surfaces per-row. + 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 = "backend_consistency.count_total_failed", + error = %e, + "count_total failed — run will not surface a progress bar" + ); + None + } + } + } + + fn progress_kind(&self) -> ProgressKind { + // Approximate — the denominator (DB count) is a proxy for + // the backend count. Deviation is meaningful (see the + // count_total doc). + ProgressKind::Approximate + } + + async fn run_resumable( + &self, + store: &dyn JobStore, + _args: &JobRunArgs, + resume_cursor: Option>, + ) -> RunOutcome { + // Cursor = opaque backend continuation token, UTF-8-encoded. + // Each backend defines its own format (local = shard/hash, + // S3 = ListObjectsV2 continuation token, Azure = list + // marker); the tenant just passes it through. + 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}"), + }; + } + }, + }; + + let mut finding_count = 0u64; + + loop { + // Cancel poll between batches. + match store.status().await { + Ok(RunStatus::CancelRequested) => { + tracing::info!( + target: "oxicloud::consistency", + event = "backend_consistency.cancelled", + run_id = %store.run_id(), + finding_count = finding_count, + "backend_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 next batch from the backend. `BlobListPage` + // splits canonical blobs (checked for orphan) from + // "unknown" entries (sidecar files, foreign namespaces — + // emitted as informational notices). + let page = match self + .backend + .list_blob_hashes(cursor.clone(), BATCH_SIZE) + .await + { + Ok(v) => v, + Err(e) => { + // Backend refuses / can't enumerate. First-batch + // failure = we emit ONE run-level anomaly and + // complete cleanly (the run stays useful — the + // operator learns why nothing was checked + // instead of getting a red error). Mid-scan + // failure = we fail the run. + + let is_first_batch = cursor.is_none() && finding_count == 0; + if is_first_batch { + // No local increment — the local + // `finding_count` is only used for the + // completion log below, but this branch + // returns immediately. The finding IS + // persisted + counted in `stats.finding_count` + // by `record_or_log` → `store.record_finding`. + record_or_log( + store, + BACKEND_CONSISTENCY_JOB_NAME, + "backend_unenumerable", + "anomaly", + None, + serde_json::json!({ + "backend": self.backend.backend_type(), + "error": format!("{e}"), + "note": "backend refused enumeration; no per-blob orphan probes attempted", + }), + ) + .await; + tracing::info!( + target: "oxicloud::consistency", + event = "backend_consistency.unenumerable", + run_id = %store.run_id(), + backend = self.backend.backend_type(), + "backend refused enumeration (typical during migration or on backends without list support)" + ); + return RunOutcome::Completed; + } + return RunOutcome::Failed { + message: format!("backend list failed mid-scan: {e}"), + }; + } + }; + + let grace_cutoff = Utc::now() - CREATE_GRACE; + + // Non-canonical files in the blob namespace — sidecars, + // wrong extensions, foreign namespaces. Informational + // only (severity `anomaly`, blue notice pill). Emitted + // BEFORE the blob orphan probes so operators see them + // grouped near the top of the findings list per batch. + // Grace-window filter applies here too — a temp file + // being written should not fire a notice. + for unknown in &page.unknowns { + if let Some(mtime) = unknown.mtime + && mtime > grace_cutoff + { + continue; + } + finding_count += 1; + record_or_log( + store, + BACKEND_CONSISTENCY_JOB_NAME, + "unknown_backend_file", + "anomaly", + None, + serde_json::json!({ + "path": unknown.path, + "mtime": unknown.mtime.map(|t| t.to_rfc3339()), + "backend": self.backend.backend_type(), + "note": "non-canonical file in blob namespace (sidecar / wrong extension); not managed by dedup", + }), + ) + .await; + } + + if page.blobs.is_empty() && page.next_cursor.is_none() { + // Nothing more to enumerate. Empty-blobs batches + // with unknowns still emitted above are fine — we + // fall through to completion. + tracing::info!( + target: "oxicloud::consistency", + event = "backend_consistency.completed", + run_id = %store.run_id(), + finding_count = finding_count, + "backend_consistency completed with {} finding(s)", + finding_count + ); + return RunOutcome::Completed; + } + + // Batch DB probe: which of these hashes have a + // `storage.blobs` row? One `WHERE hash = ANY($1)` per + // batch — indexed lookup, cheap even on millions of + // rows. + let batch_hashes: Vec = page.blobs.iter().map(|e| e.hash.clone()).collect(); + let db_present: HashSet = if batch_hashes.is_empty() { + HashSet::new() + } else { + match sqlx::query_as::<_, (String,)>( + r#"SELECT hash FROM storage.blobs WHERE hash = ANY($1)"#, + ) + .bind(&batch_hashes[..]) + .fetch_all(self.pool.as_ref()) + .await + { + Ok(rows) => rows.into_iter().map(|(h,)| h).collect(), + Err(e) => { + return RunOutcome::Failed { + message: format!("db probe: {e}"), + }; + } + } + }; + + for entry in &page.blobs { + if db_present.contains(&entry.hash) { + continue; + } + if let Some(mtime) = entry.mtime + && mtime > grace_cutoff + { + continue; + } + + finding_count += 1; + record_or_log( + store, + BACKEND_CONSISTENCY_JOB_NAME, + "orphan_blob", + "inconsistent", + None, + serde_json::json!({ + "hash": entry.hash, + "mtime": entry.mtime.map(|t| t.to_rfc3339()), + "backend": self.backend.backend_type(), + }), + ) + .await; + } + + // Advance cursor + checkpoint. Scanned count tracks + // both blobs and unknowns since we walked both. + let batch_len = (page.blobs.len() + page.unknowns.len()) as u64; + cursor = page.next_cursor; + let cursor_bytes = cursor + .as_ref() + .map(|s| s.as_bytes().to_vec()) + .unwrap_or_default(); + if let Err(e) = store.checkpoint(cursor_bytes, batch_len).await { + return RunOutcome::Failed { + message: format!("checkpoint: {e}"), + }; + } + + // Backend returned no next_cursor → enumeration + // complete. Emit the completion log and return. + if cursor.is_none() { + tracing::info!( + target: "oxicloud::consistency", + event = "backend_consistency.completed", + run_id = %store.run_id(), + finding_count = finding_count, + "backend_consistency completed with {} finding(s)", + finding_count + ); + return RunOutcome::Completed; + } + } + } +} diff --git a/src/infrastructure/services/cached_blob_backend.rs b/src/infrastructure/services/cached_blob_backend.rs index 25d6fb28..b9ee0d95 100644 --- a/src/infrastructure/services/cached_blob_backend.rs +++ b/src/infrastructure/services/cached_blob_backend.rs @@ -412,6 +412,29 @@ impl BlobStorageBackend for CachedBlobBackend { let path = self.cached_path(hash); if path.exists() { Some(path) } else { None } } + + /// 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 + /// "most of my blobs are orphans" from the tenant's perspective. + /// The inner backend is the authoritative "what exists" source. + fn list_blob_hashes( + &self, + cursor: Option, + limit: usize, + ) -> Pin< + Box< + dyn std::future::Future< + Output = Result< + crate::application::ports::blob_storage_ports::BlobListPage, + DomainError, + >, + > + Send + + '_, + >, + > { + self.inner.list_blob_hashes(cursor, limit) + } } // ── Cache internals (miss path + population) ─────────────────────── diff --git a/src/infrastructure/services/encrypted_blob_backend.rs b/src/infrastructure/services/encrypted_blob_backend.rs index 272d7806..d4d3de05 100644 --- a/src/infrastructure/services/encrypted_blob_backend.rs +++ b/src/infrastructure/services/encrypted_blob_backend.rs @@ -351,6 +351,29 @@ impl BlobStorageBackend for EncryptedBlobBackend { // Encrypted blobs cannot be served directly from disk None } + + /// Enumeration = plaintext hashes, same as the inner backend. + /// Encryption operates on payload bytes, not on the hash key: + /// blob objects on the inner backend are stored under the + /// PLAINTEXT hash so dedup works. Delegating list to the inner + /// backend therefore returns exactly the right identifiers. + fn list_blob_hashes( + &self, + cursor: Option, + limit: usize, + ) -> Pin< + Box< + dyn std::future::Future< + Output = Result< + crate::application::ports::blob_storage_ports::BlobListPage, + DomainError, + >, + > + Send + + '_, + >, + > { + self.inner.list_blob_hashes(cursor, limit) + } } /// Collect a byte stream into a single `Vec`. diff --git a/src/infrastructure/services/local_blob_backend.rs b/src/infrastructure/services/local_blob_backend.rs index dc714132..af79cd2f 100644 --- a/src/infrastructure/services/local_blob_backend.rs +++ b/src/infrastructure/services/local_blob_backend.rs @@ -10,6 +10,7 @@ use tokio::io::{AsyncSeekExt, AsyncWriteExt}; use tokio_util::io::ReaderStream; use bytes::Bytes; +use chrono::{DateTime, Utc}; use crate::application::ports::blob_storage_ports::{ BlobStorageBackend, BlobStream, StorageHealthStatus, @@ -607,6 +608,165 @@ impl BlobStorageBackend for LocalBlobBackend { fn read_prefetch(&self) -> usize { self.read_prefetch } + + /// Enumerate `.blob` files under `.blobs//`. Cursor format: + /// + /// * `None` — start from the first shard (`00`) at file offset 0 + /// * `Some("/")` — resume: skip shards `< shard` + /// entirely, and within `shard` skip files whose hash `≤ hash`. + /// + /// Ordering: shards ascending (00–ff), files within a shard + /// ascending by hash. Stable across calls given the sorting. + /// + /// Filter: basename must be exactly 64 hex chars + `.blob`. This + /// excludes `.tmp` staging files, `.orig`/`.lost`/`.corrupt` + /// sidecars from manual admin work, and any other non-canonical + /// artefacts. Backend consistency scans the DB-registered + /// content-addressable set only. + fn list_blob_hashes( + &self, + cursor: Option, + limit: usize, + ) -> Pin< + Box< + dyn std::future::Future< + Output = Result< + crate::application::ports::blob_storage_ports::BlobListPage, + DomainError, + >, + > + Send + + '_, + >, + > { + use crate::application::ports::blob_storage_ports::{ + BackendBlobEntry, BackendUnknownEntry, BlobListPage, + }; + + let blob_root = self.blob_root.clone(); + Box::pin(async move { + let (start_shard, start_after_hash): (String, Option) = match cursor { + None => (String::from("00"), None), + Some(c) => match c.split_once('/') { + Some((sh, h)) => (sh.to_string(), Some(h.to_string())), + None => (c, None), + }, + }; + + let mut blobs: Vec = Vec::with_capacity(limit); + let mut unknowns: Vec = Vec::new(); + let mut next_cursor: Option = None; + + for prefix in &HEX_PREFIXES { + let prefix = *prefix; + if prefix < start_shard.as_str() { + continue; + } + let shard_dir = blob_root.join(prefix); + let mut entries = match fs::read_dir(&shard_dir).await { + Ok(e) => e, + Err(e) if e.kind() == std::io::ErrorKind::NotFound => continue, + Err(e) => { + return Err(DomainError::new( + ErrorKind::InternalError, + "Blob", + format!("read shard {prefix}: {e}"), + )); + } + }; + + // Collect canonical blobs + unknowns for this shard. + // The distinction is filename shape: `<64-hex>.blob` + // → canonical blob; anything else → unknown sidecar. + // Unknowns are captured with their full basename so + // the tenant can surface them to operators as + // informational notices (severity `anomaly`). + let mut shard_blobs: Vec<(String, Option>)> = Vec::new(); + let mut shard_unknowns: Vec<(String, Option>)> = Vec::new(); + while let Some(dirent) = entries.next_entry().await.map_err(|e| { + DomainError::new( + ErrorKind::InternalError, + "Blob", + format!("read shard {prefix} entry: {e}"), + ) + })? { + let name = dirent.file_name(); + let name_str = match name.to_str() { + Some(s) => s, + None => continue, // non-UTF8 filename — skip entirely + }; + // Skip directories — the shard dir itself + // shouldn't contain any, but defensively. + if dirent + .file_type() + .await + .map(|t| t.is_dir()) + .unwrap_or(false) + { + continue; + } + let mtime = dirent + .metadata() + .await + .ok() + .and_then(|m| m.modified().ok()) + .map(DateTime::::from); + + // Canonical shape check: `<64-hex>.blob`. + let canonical = name_str + .strip_suffix(".blob") + .filter(|stem| { + stem.len() == 64 && stem.chars().all(|c| c.is_ascii_hexdigit()) + }) + .map(|s| s.to_string()); + + match canonical { + Some(hash) => shard_blobs.push((hash, mtime)), + None => shard_unknowns.push((name_str.to_string(), mtime)), + } + } + shard_blobs.sort_by(|a, b| a.0.cmp(&b.0)); + + // Unknowns don't need cursor-precise ordering — they + // ride alongside the blobs batch. Sort just for + // stable operator-facing output. + shard_unknowns.sort_by(|a, b| a.0.cmp(&b.0)); + for (name, mtime) in shard_unknowns { + unknowns.push(BackendUnknownEntry { + path: format!("{prefix}/{name}"), + mtime, + }); + } + + for (hash, mtime) in shard_blobs { + if prefix == start_shard.as_str() + && let Some(ref after) = start_after_hash + && hash.as_str() <= after.as_str() + { + continue; + } + if blobs.len() >= limit { + next_cursor = Some(format!( + "{}/{}", + prefix, + blobs.last().map(|e| e.hash.as_str()).unwrap_or("") + )); + return Ok(BlobListPage { + blobs, + unknowns, + next_cursor, + }); + } + blobs.push(BackendBlobEntry { hash, mtime }); + } + } + + Ok(BlobListPage { + blobs, + unknowns, + next_cursor, + }) + }) + } } #[cfg(test)] diff --git a/src/infrastructure/services/migration_blob_backend.rs b/src/infrastructure/services/migration_blob_backend.rs index e5a87fd0..dbe32231 100644 --- a/src/infrastructure/services/migration_blob_backend.rs +++ b/src/infrastructure/services/migration_blob_backend.rs @@ -229,4 +229,45 @@ impl BlobStorageBackend for MigrationBlobBackend { .local_blob_path(hash) .or_else(|| self.source.local_blob_path(hash)) } + + /// Enumeration during migration is intentionally REFUSED. Both + /// source and target legitimately hold bytes concurrently + /// mid-migration: a blob copied to target but not yet deleted + /// from source would be reported "twice"; a blob in-flight from + /// source to target could be flagged as orphan on whichever + /// side the consistency scan doesn't walk. There's no single + /// authoritative "what's on the backend" answer while a + /// migration is running. + /// + /// Operators wanting to run `backend_consistency` during a + /// migration should either wait for the migration to complete + /// (target becomes authoritative) or cancel it. The + /// `operation_not_supported` error is surfaced by the tenant as + /// a single run-level `backend_unenumerable` finding — no + /// per-blob probes attempted. + fn list_blob_hashes( + &self, + _cursor: Option, + _limit: usize, + ) -> Pin< + Box< + dyn std::future::Future< + Output = Result< + crate::application::ports::blob_storage_ports::BlobListPage, + DomainError, + >, + > + Send + + '_, + >, + > { + Box::pin(async { + Err(DomainError::operation_not_supported( + "list_blob_hashes", + "backend_consistency cannot enumerate while a storage \ + migration is in progress — source and target hold bytes \ + concurrently; wait for migration completion or cancel it \ + before running the scan", + )) + }) + } } diff --git a/src/infrastructure/services/mod.rs b/src/infrastructure/services/mod.rs index af79a2d0..457e0ef1 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 backend_consistency_service; pub mod blobs_consistency_service; pub mod cached_blob_backend; pub mod chunked_upload_service; diff --git a/src/infrastructure/services/retry_blob_backend.rs b/src/infrastructure/services/retry_blob_backend.rs index b09b4b3d..e06dcc30 100644 --- a/src/infrastructure/services/retry_blob_backend.rs +++ b/src/infrastructure/services/retry_blob_backend.rs @@ -355,4 +355,26 @@ impl BlobStorageBackend for RetryBlobBackend { fn local_blob_path(&self, hash: &str) -> Option { self.inner.local_blob_path(hash) } + + /// 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 + /// treats it as a transient backend issue and skips the batch. + fn list_blob_hashes( + &self, + cursor: Option, + limit: usize, + ) -> Pin< + Box< + dyn std::future::Future< + Output = Result< + crate::application::ports::blob_storage_ports::BlobListPage, + DomainError, + >, + > + Send + + '_, + >, + > { + self.inner.list_blob_hashes(cursor, limit) + } } diff --git a/src/infrastructure/services/s3_blob_backend.rs b/src/infrastructure/services/s3_blob_backend.rs index 7a910ea3..75c69721 100644 --- a/src/infrastructure/services/s3_blob_backend.rs +++ b/src/infrastructure/services/s3_blob_backend.rs @@ -421,4 +421,97 @@ impl BlobStorageBackend for S3BlobBackend { fn local_blob_path(&self, _hash: &str) -> Option { None // Remote backend — no local path } + + /// Enumerate blobs via S3 `ListObjectsV2`. Cursor is the S3 + /// continuation token verbatim (opaque). Filter: keys must + /// match `/<64-hex>.blob` — matches how `blob_key` writes + /// them — so any future non-blob namespace living in the same + /// bucket (e.g. `thumbnails/.jpg`) is skipped + /// automatically. No prefix passed to S3 so we get everything + /// in one paginated scan; the client-side filter enforces + /// correctness. + fn list_blob_hashes( + &self, + cursor: Option, + limit: usize, + ) -> Pin< + Box< + dyn std::future::Future< + Output = Result< + crate::application::ports::blob_storage_ports::BlobListPage, + DomainError, + >, + > + Send + + '_, + >, + > { + use crate::application::ports::blob_storage_ports::{ + BackendBlobEntry, BackendUnknownEntry, BlobListPage, + }; + + Box::pin(async move { + let mut req = self + .client + .list_objects_v2() + .bucket(&self.bucket) + .max_keys(limit.min(1000) as i32); + if let Some(c) = cursor { + req = req.continuation_token(c); + } + + let resp = req.send().await.map_err(|e| { + DomainError::new( + ErrorKind::InternalError, + "Blob", + format!("S3 ListObjectsV2 failed: {e}"), + ) + })?; + + let objects = resp.contents.unwrap_or_default(); + let mut blobs: Vec = Vec::with_capacity(objects.len()); + let mut unknowns: Vec = Vec::new(); + + for obj in objects { + let Some(key) = obj.key else { continue }; + let mtime = obj.last_modified.and_then(|ts| { + let secs = ts.secs(); + let nsecs = ts.subsec_nanos(); + chrono::DateTime::::from_timestamp(secs, nsecs) + }); + + // Canonical S3 key shape: `/<64-hex>.blob`. + // Anything else is a sidecar or foreign namespace + // (e.g. future `thumbnails/.jpg` if Ed adds + // that) — surface as an unknown so operators know + // it's there. Recovery framework can decide per- + // pattern how to act. + let is_canonical = key.split_once('/').and_then(|(prefix, rest)| { + if prefix.len() != 2 || !prefix.chars().all(|c| c.is_ascii_hexdigit()) { + return None; + } + rest.strip_suffix(".blob") + .filter(|stem| { + stem.len() == 64 && stem.chars().all(|c| c.is_ascii_hexdigit()) + }) + .map(|s| s.to_string()) + }); + + match is_canonical { + Some(hash) => blobs.push(BackendBlobEntry { hash, mtime }), + None => unknowns.push(BackendUnknownEntry { path: key, mtime }), + } + } + + let next_cursor = if resp.is_truncated.unwrap_or(false) { + resp.next_continuation_token + } else { + None + }; + Ok(BlobListPage { + blobs, + unknowns, + next_cursor, + }) + }) + } }