feat(recoverable-job): add backend_consistency (storage)

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<DateTime<Utc>>,
+}
+
+/// 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<DateTime<Utc>>,
+}
+
+/// 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<BackendBlobEntry>,
+    pub unknowns: Vec<BackendUnknownEntry>,
+    pub next_cursor: Option<String>,
+}
+
 /// Boxed future alias used by [`BlobStorageBackend`] to keep the trait dyn-compatible.
 type BoxFut<'a, T> = Pin<Box<dyn Future<Output = T> + 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<String>,
+        _limit: usize,
+    ) -> BoxFut<'_, Result<BlobListPage, DomainError>> {
+        Box::pin(async {
+            Err(DomainError::operation_not_supported(
+                "list_blob_hashes",
+                "this backend does not implement enumeration",
+            ))
+        })
+    }
 }

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

src/infrastructure/services/azure_blob_backend.rs

@@ -389,4 +389,14 @@ impl BlobStorageBackend for AzureBlobBackend {
     fn local_blob_path(&self, _hash: &str) -> Option<PathBuf> {
         None
     }
+
+    // TODO: implement `list_blob_hashes` via
+    // `container_client.list_blobs()` (`azure_storage_blobs`
+    // paginator). Same filter as local + S3 impls:
+    // `<xx>/<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.
 }

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<PgPool>,
+    backend: Arc<dyn BlobStorageBackend>,
+}
+
+impl BackendConsistencyCheck {
+    pub fn new(pool: Arc<PgPool>, backend: Arc<dyn BlobStorageBackend>) -> Self {
+        Self { pool, backend }
+    }
+
+    pub async fn register_recoverable_job(
+        self: Arc<Self>,
+        registry: &JobRegistry,
+        provider: &Arc<dyn JobStoreProvider>,
+    ) -> Arc<Self> {
+        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<u64> {
+        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<Vec<u8>>,
+    ) -> 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<String> = 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<String> = page.blobs.iter().map(|e| e.hash.clone()).collect();
+            let db_present: HashSet<String> = 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;
+            }
+        }
+    }
+}

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<String>,
+        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) ───────────────────────

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<String>,
+        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<u8>`.

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/<xx>/`. Cursor format:
+    ///
+    /// * `None` — start from the first shard (`00`) at file offset 0
+    /// * `Some("<shard>/<hash>")` — 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<String>,
+        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<String>) = 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<BackendBlobEntry> = Vec::with_capacity(limit);
+            let mut unknowns: Vec<BackendUnknownEntry> = Vec::new();
+            let mut next_cursor: Option<String> = 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<DateTime<Utc>>)> = Vec::new();
+                let mut shard_unknowns: Vec<(String, Option<DateTime<Utc>>)> = 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::<Utc>::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)]

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<String>,
+        _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",
+            ))
+        })
+    }
 }

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;

src/infrastructure/services/retry_blob_backend.rs

@@ -355,4 +355,26 @@ impl BlobStorageBackend for RetryBlobBackend {
     fn local_blob_path(&self, hash: &str) -> Option<PathBuf> {
         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<String>,
+        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)
+    }
 }

src/infrastructure/services/s3_blob_backend.rs

@@ -421,4 +421,97 @@ impl BlobStorageBackend for S3BlobBackend {
     fn local_blob_path(&self, _hash: &str) -> Option<PathBuf> {
         None // Remote backend — no local path
     }
+
+    /// Enumerate blobs via S3 `ListObjectsV2`. Cursor is the S3
+    /// continuation token verbatim (opaque). Filter: keys must
+    /// match `<xx>/<64-hex>.blob` — matches how `blob_key` writes
+    /// them — so any future non-blob namespace living in the same
+    /// bucket (e.g. `thumbnails/<hash>.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<String>,
+        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<BackendBlobEntry> = Vec::with_capacity(objects.len());
+            let mut unknowns: Vec<BackendUnknownEntry> = 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::<chrono::Utc>::from_timestamp(secs, nsecs)
+                });
+
+                // Canonical S3 key shape: `<xx>/<64-hex>.blob`.
+                // Anything else is a sidecar or foreign namespace
+                // (e.g. future `thumbnails/<hash>.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,
+            })
+        })
+    }
 }