From f70dffeaf5b9a071e6d0228856f2956f025649c2 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 15 Jun 2026 10:56:33 +0000 Subject: [PATCH] perf(dedup): backend-aware chunk read-ahead for CDC reassembly MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 | 14 ++++++++++++++ src/infrastructure/services/azure_blob_backend.rs | 5 +++++ src/infrastructure/services/cached_blob_backend.rs | 6 ++++++ src/infrastructure/services/dedup_service.rs | 13 +++++++++---- .../services/encrypted_blob_backend.rs | 5 +++++ .../services/migration_blob_backend.rs | 6 ++++++ src/infrastructure/services/retry_blob_backend.rs | 5 +++++ src/infrastructure/services/s3_blob_backend.rs | 5 +++++ 8 files changed, 55 insertions(+), 4 deletions(-) diff --git a/src/application/ports/blob_storage_ports.rs b/src/application/ports/blob_storage_ports.rs index 10310626..13a01ff0 100644 --- a/src/application/ports/blob_storage_ports.rs +++ b/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; + + /// 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 + } } diff --git a/src/infrastructure/services/azure_blob_backend.rs b/src/infrastructure/services/azure_blob_backend.rs index 5f949a89..77f3faad 100644 --- a/src/infrastructure/services/azure_blob_backend.rs +++ b/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 { None } diff --git a/src/infrastructure/services/cached_blob_backend.rs b/src/infrastructure/services/cached_blob_backend.rs index 2cb3288f..2457a763 100644 --- a/src/infrastructure/services/cached_blob_backend.rs +++ b/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 { // If the blob is cached locally, return that path let path = self.cached_path(hash); diff --git a/src/infrastructure/services/dedup_service.rs b/src/infrastructure/services/dedup_service.rs index 867c0767..5bb3e3c0 100644 --- a/src/infrastructure/services/dedup_service.rs +++ b/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)) diff --git a/src/infrastructure/services/encrypted_blob_backend.rs b/src/infrastructure/services/encrypted_blob_backend.rs index a16411c7..16f6101d 100644 --- a/src/infrastructure/services/encrypted_blob_backend.rs +++ b/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 { // Encrypted blobs cannot be served directly from disk None diff --git a/src/infrastructure/services/migration_blob_backend.rs b/src/infrastructure/services/migration_blob_backend.rs index 75e0c716..e5a87fd0 100644 --- a/src/infrastructure/services/migration_blob_backend.rs +++ b/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 { // Prefer target, fall back to source. self.target diff --git a/src/infrastructure/services/retry_blob_backend.rs b/src/infrastructure/services/retry_blob_backend.rs index 6467081a..8727a1ed 100644 --- a/src/infrastructure/services/retry_blob_backend.rs +++ b/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 { self.inner.local_blob_path(hash) } diff --git a/src/infrastructure/services/s3_blob_backend.rs b/src/infrastructure/services/s3_blob_backend.rs index abf6630b..98ca5968 100644 --- a/src/infrastructure/services/s3_blob_backend.rs +++ b/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 { None // Remote backend — no local path }