perf(dedup): backend-aware chunk read-ahead for CDC reassembly

read_blob_stream / read_blob_range_stream reassembled a CDC file by fetching
its chunks with `buffered(1)` — strictly sequential, so the next chunk's
backend fetch (a file `open` locally; a full request round-trip on S3/Azure)
only started after the current chunk was fully drained.

A benchmark of the exact pipeline (stream::iter(chunks).map(get).buffered(K)
.try_flatten()) showed a blind `buffered(4)` is the WRONG fix: on a local
disk it is neutral on a warm page cache and ~37% SLOWER cold, because
concurrent opens turn one sequential read into several competing random-I/O
streams over content-addressed (scattered) chunk files. The win is entirely
on remote backends, where per-chunk request latency dominates and overlapping
fetches hide it (≈ linear in K).

So the read-ahead depth is now a backend hint, not a constant:
- BlobStorageBackend::read_prefetch() default 1 (sequential; safe for local).
- S3 / Azure override to 8 (overlap GETs to hide TTFB).
- cached / encrypted / retry / migration delegate to the backend that serves
  the bytes.
- Both CDC read paths use `self.backend.read_prefetch().max(1)`.

Net: local backend unchanged (no regression); remote reassembly ~4-8x faster.
Ordered `buffered` (not buffer_unordered) keeps chunks in sequence.

Bench (per-chunk fetch-latency model): buffered(1)->(4)/(8) = x3.9 / x7.8
@1ms, x4.0 / x8.1 @5ms, x4.0 / x8.0 @20ms. Local warm: 230ms@1 vs 227ms@4
(noise); local cold: 425ms@1 vs 585ms@4 (why local stays at 1).

https://claude.ai/code/session_01DCszkkU11LYxMEUWr4setK

src/application/ports/blob_storage_ports.rs

@@ -126,4 +126,18 @@ pub trait BlobStorageBackend: Send + Sync + 'static {
     /// Only meaningful for local-filesystem backends.  Remote backends
     /// return `None`; callers that need a local file must stream + spool.
     fn local_blob_path(&self, hash: &str) -> Option<PathBuf>;
+
+    /// How many chunk fetches the CDC reader may run concurrently when
+    /// reassembling a file (`read_blob_stream`'s `buffered(N)` read-ahead).
+    ///
+    /// The default is **1** — sequential, because for a local disk concurrent
+    /// opens turn one sequential read into several competing random-I/O streams
+    /// over content-addressed (scattered) chunk files, which is neutral on a
+    /// warm page cache and *slower* cold. Remote backends (S3/Azure) override
+    /// this with a higher value: there the dominant cost is per-chunk request
+    /// latency, and overlapping fetches hides it (≈ N× faster reassembly).
+    /// Wrapping backends delegate to the backend that actually serves the bytes.
+    fn read_prefetch(&self) -> usize {
+        1
+    }
 }

src/infrastructure/services/azure_blob_backend.rs

@@ -319,6 +319,11 @@ impl BlobStorageBackend for AzureBlobBackend {
         "azure"
     }
 
+    /// Remote object store: overlap chunk GETs to hide per-request latency.
+    fn read_prefetch(&self) -> usize {
+        8
+    }
+
     fn local_blob_path(&self, _hash: &str) -> Option<PathBuf> {
         None
     }

src/infrastructure/services/cached_blob_backend.rs

@@ -385,6 +385,12 @@ impl BlobStorageBackend for CachedBlobBackend {
         "cached"
     }
 
+    /// A cache miss fetches from the inner backend, so adopt its read-ahead
+    /// (high for remote, where prefetch pays off; hits read local cache files).
+    fn read_prefetch(&self) -> usize {
+        self.inner.read_prefetch()
+    }
+
     fn local_blob_path(&self, hash: &str) -> Option<PathBuf> {
         // If the blob is cached locally, return that path
         let path = self.cached_path(hash);

src/infrastructure/services/dedup_service.rs

@@ -1438,7 +1438,10 @@ impl DedupService {
         .map_err(|e| DomainError::internal_error("Dedup", format!("Manifest lookup: {}", e)))?;
 
         if let Some(chunk_hashes) = manifest {
-            // CDC file: stream chunks in order
+            // CDC file: stream chunks in order. Read-ahead depth is the
+            // backend's hint (1 for local disk, higher for remote object
+            // stores where overlapping fetches hide per-chunk latency).
+            let prefetch = self.backend.read_prefetch().max(1);
             let backend = self.backend.clone();
             let chunk_stream = stream::iter(chunk_hashes)
                 .map(move |chunk_hash| {
@@ -1450,7 +1453,7 @@ impl DedupService {
                             .map_err(|e| std::io::Error::other(e.to_string()))
                     }
                 })
-                .buffered(1)
+                .buffered(prefetch)
                 .try_flatten();
 
             Ok(Box::pin(chunk_stream))
@@ -1529,7 +1532,9 @@ impl DedupService {
                 }
             }
 
-            // Stream selected chunks with ranges
+            // Stream selected chunks with ranges. Read-ahead depth from the
+            // backend hint (local=1; remote overlaps fetches — see read_blob_stream).
+            let prefetch = self.backend.read_prefetch().max(1);
             let backend = self.backend.clone();
             let chunk_stream = stream::iter(selected)
                 .map(move |(chunk_hash, range_start, range_end)| {
@@ -1541,7 +1546,7 @@ impl DedupService {
                             .map_err(|e| std::io::Error::other(e.to_string()))
                     }
                 })
-                .buffered(1)
+                .buffered(prefetch)
                 .try_flatten();
 
             Ok(Box::pin(chunk_stream))

src/infrastructure/services/encrypted_blob_backend.rs

@@ -306,6 +306,11 @@ impl BlobStorageBackend for EncryptedBlobBackend {
         "encrypted"
     }
 
+    /// Transparent wrapper: the inner backend serves the bytes.
+    fn read_prefetch(&self) -> usize {
+        self.inner.read_prefetch()
+    }
+
     fn local_blob_path(&self, _hash: &str) -> Option<PathBuf> {
         // Encrypted blobs cannot be served directly from disk
         None

src/infrastructure/services/migration_blob_backend.rs

@@ -217,6 +217,12 @@ impl BlobStorageBackend for MigrationBlobBackend {
         "migration"
     }
 
+    /// Reads are served target-first (see `get_blob_stream`), so adopt the
+    /// target's read-ahead.
+    fn read_prefetch(&self) -> usize {
+        self.target.read_prefetch()
+    }
+
     fn local_blob_path(&self, hash: &str) -> Option<PathBuf> {
         // Prefer target, fall back to source.
         self.target

src/infrastructure/services/retry_blob_backend.rs

@@ -268,6 +268,11 @@ impl BlobStorageBackend for RetryBlobBackend {
         "retry"
     }
 
+    /// Transparent wrapper: the inner backend serves the bytes.
+    fn read_prefetch(&self) -> usize {
+        self.inner.read_prefetch()
+    }
+
     fn local_blob_path(&self, hash: &str) -> Option<PathBuf> {
         self.inner.local_blob_path(hash)
     }

src/infrastructure/services/s3_blob_backend.rs

@@ -381,6 +381,11 @@ impl BlobStorageBackend for S3BlobBackend {
         "s3"
     }
 
+    /// Remote object store: overlap chunk GETs to hide per-request latency.
+    fn read_prefetch(&self) -> usize {
+        8
+    }
+
     fn local_blob_path(&self, _hash: &str) -> Option<PathBuf> {
         None // Remote backend — no local path
     }