From 9a181053bdcc329b09f316f9d527d0f09bf6666c Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 10 Jun 2026 09:55:02 +0000 Subject: [PATCH] perf(blobs): batch chunk fsyncs into one durability sweep per upload Storing a new file through CDC dedup issued sync_all + a parent-dir fsync for every ~256 KB chunk (~8,200 fsyncs for a 1 GB upload), plus one PG INSERT round-trip per chunk. The actual durability boundary is the manifest INSERT: chunks only need to be durable before any PG row references them, not one by one. - BlobStorageBackend grows put_blob_from_bytes_unsynced + sync_blobs with conservative defaults (unsynced delegates to the synced write, sync_blobs is a no-op) so backends that don't opt in keep the per-write durability semantics. Remote stores are durable on PUT. - LocalBlobBackend writes chunks without fsync and implements sync_blobs as a parallel sweep: every listed blob file (hard requirement) plus each distinct prefix directory exactly once (best-effort, same tier as fsync_parent_dir). - DedupService::store_chunks writes new chunks unsynced, runs one sync_blobs sweep, then registers all new chunks in ONE batched UNNEST INSERT - durability before visibility, and the per-chunk PG round-trips collapse into one. - Encrypted/Migration decorators forward both methods so the optimization survives encrypted-local and live-migration stacks. https://claude.ai/code/session_013Bk4BMQEvR9QxCU7QXLRwv --- src/application/ports/blob_storage_ports.rs | 32 +++ src/infrastructure/services/dedup_service.rs | 89 ++++--- .../services/encrypted_blob_backend.rs | 51 +++- .../services/local_blob_backend.rs | 249 ++++++++++++++++-- .../services/migration_blob_backend.rs | 15 ++ 5 files changed, 367 insertions(+), 69 deletions(-) diff --git a/src/application/ports/blob_storage_ports.rs b/src/application/ports/blob_storage_ports.rs index 391fe164..cf3be3dc 100644 --- a/src/application/ports/blob_storage_ports.rs +++ b/src/application/ports/blob_storage_ports.rs @@ -60,6 +60,38 @@ pub trait BlobStorageBackend: Send + Sync + 'static { /// without overwriting. Returns the number of bytes stored. fn put_blob_from_bytes(&self, hash: &str, data: Bytes) -> BoxFut<'_, Result>; + /// Store a blob from in-memory bytes **without forcing durability**. + /// + /// Same idempotency contract as [`Self::put_blob_from_bytes`], but the + /// bytes may still sit in volatile caches (e.g. the OS page cache) when + /// the future resolves. Durability is only guaranteed after a subsequent + /// [`Self::sync_blobs`] covering this hash returns `Ok`. Callers MUST NOT + /// record a durable reference to the blob (e.g. a PostgreSQL row) before + /// that sync completes. + /// + /// Default: delegates to `put_blob_from_bytes` (immediately durable), + /// pairing with the no-op `sync_blobs` default so backends that don't + /// opt in keep today's per-write durability semantics. + fn put_blob_from_bytes_unsynced( + &self, + hash: &str, + data: Bytes, + ) -> BoxFut<'_, Result> { + self.put_blob_from_bytes(hash, data) + } + + /// Make previously written blobs durable in one batched operation. + /// + /// Durability barrier for blobs written via `put_blob_from_bytes_unsynced`: + /// when this returns `Ok`, every listed blob is crash-safe. Local + /// filesystem backends fsync each listed blob file plus each distinct + /// parent directory once — one sweep per upload instead of two fsyncs + /// per chunk. Remote object stores are durable on PUT, so the default + /// is a no-op. + fn sync_blobs(&self, _hashes: &[String]) -> BoxFut<'_, Result<(), DomainError>> { + Box::pin(async { Ok(()) }) + } + /// Stream the full blob content in chunks. fn get_blob_stream(&self, hash: &str) -> BoxFut<'_, Result>; diff --git a/src/infrastructure/services/dedup_service.rs b/src/infrastructure/services/dedup_service.rs index a916d60d..8b27dec7 100644 --- a/src/infrastructure/services/dedup_service.rs +++ b/src/infrastructure/services/dedup_service.rs @@ -21,10 +21,13 @@ //! **Write-first strategy** (store_from_file): //! 1. CDC-analyse the file (mmap → FastCDC boundaries + per-chunk BLAKE3). //! 2. Batch-check which chunk hashes already exist in PG (dedup skip). -//! 3. Read + upload only *new* chunks to the blob backend (idempotent). -//! 4. Bump ref_count for existing chunks (no disk I/O). -//! 5. Single manifest INSERT (~few ms total). -//! 6. PG connection is never held during disk I/O. +//! 3. Bump ref_count for existing chunks (no disk I/O). +//! 4. Read + write only *new* chunks to the blob backend (idempotent, +//! no per-chunk fsync). +//! 5. One batched fsync sweep makes the new chunks durable, then ONE +//! batched INSERT registers them — durability before visibility. +//! 6. Single manifest INSERT (~few ms total). +//! 7. PG connection is never held during disk I/O. //! //! Benefits: //! - Sub-file dedup: edited files share unchanged chunks @@ -457,10 +460,17 @@ impl DedupService { /// /// Phase 0: Batch-queries PG to discover which chunk hashes already /// exist in `storage.blobs`. - /// Phase 1: Reads only *new* chunks from the source file (the biggest - /// I/O saving for versioned files where most chunks are unchanged). - /// Uploads each new chunk and bumps `ref_count` for chunks that already - /// exist, with up to [`CHUNK_UPLOAD_CONCURRENCY`] uploads in flight. + /// Phase 1: Bumps `ref_count` for chunks that already exist (one + /// batched UPDATE, no disk I/O — the biggest saving for versioned + /// files where most chunks are unchanged). + /// Phase 2: Reads + writes only *new* chunks, with up to + /// [`CHUNK_UPLOAD_CONCURRENCY`] writes in flight and **no per-chunk + /// fsync**. + /// Phase 3: One batched `sync_blobs` sweep makes every new chunk + /// durable (no-op for remote backends, which are durable on PUT). + /// Phase 4: ONE batched INSERT registers the new chunks in PG. The + /// sweep runs first so a crash can never leave a `storage.blobs` row + /// pointing at bytes that were still in the page cache. /// /// `ref_count` is incremented once per *distinct* chunk (one reference per /// manifest), staying symmetric with `remove_manifest_reference` so a file @@ -540,10 +550,13 @@ impl DedupService { DomainError::internal_error("Dedup", format!("Failed to open source file: {}", e)) })?); - let results: Vec> = stream::iter(new_ops) + // Writes are *unsynced*: no per-chunk fsync. Durability comes from + // the single batched sweep below, BEFORE any PG row references the + // new chunks — so a crash can never leave storage.blobs claiming a + // chunk whose bytes didn't reach the platter. + let results: Vec> = stream::iter(new_ops) .map(|(hash, offset, length)| { let source = source.clone(); - let pool = pool.clone(); let backend = backend.clone(); async move { // Positioned read of just this chunk (≤ CDC_MAX_CHUNK) off @@ -563,36 +576,48 @@ impl DedupService { })?; backend - .put_blob_from_bytes(&hash, Bytes::from(bytes)) + .put_blob_from_bytes_unsynced(&hash, Bytes::from(bytes)) .await?; - // ON CONFLICT covers a concurrent uploader inserting the - // same brand-new chunk between the existence check above and - // this INSERT. - sqlx::query( - "INSERT INTO storage.blobs (hash, size, ref_count) - VALUES ($1, $2, 1) - ON CONFLICT (hash) DO UPDATE - SET ref_count = storage.blobs.ref_count + 1", - ) - .bind(&hash) - .bind(length as i64) - .execute(pool.as_ref()) - .await - .map_err(|e| { - DomainError::internal_error( - "Dedup", - format!("Failed to upsert chunk: {}", e), - ) - })?; - Ok(()) + Ok((hash, length as i64)) } }) .buffer_unordered(Self::CHUNK_UPLOAD_CONCURRENCY) .collect() .await; + let mut new_rows: Vec<(String, i64)> = Vec::with_capacity(results.len()); for result in results { - result?; + new_rows.push(result?); + } + + if !new_rows.is_empty() { + // ── Phase 3: durability barrier — one batched fsync sweep ────── + // (was 2 fsyncs per chunk: ~8 200 for a 1 GB upload; now one + // parallel sweep over the new files + ≤256 prefix dirs). + // Remote backends are durable on PUT — sync_blobs is a no-op. + let new_hashes: Vec = new_rows.iter().map(|(h, _)| h.clone()).collect(); + backend.sync_blobs(&new_hashes).await?; + + // ── Phase 4: register all new chunks in ONE batched INSERT ───── + // (was one round-trip per chunk). `new_rows` is built from + // `unique_chunks`, so no hash repeats within the batch — safe for + // ON CONFLICT DO UPDATE, which covers a concurrent uploader + // inserting the same brand-new chunk between the existence check + // in Phase 0 and this INSERT. + let new_sizes: Vec = new_rows.iter().map(|(_, s)| *s).collect(); + sqlx::query( + "INSERT INTO storage.blobs (hash, size, ref_count) + SELECT h, s, 1 FROM UNNEST($1::text[], $2::bigint[]) AS t(h, s) + ON CONFLICT (hash) DO UPDATE + SET ref_count = storage.blobs.ref_count + 1", + ) + .bind(&new_hashes) + .bind(&new_sizes) + .execute(pool.as_ref()) + .await + .map_err(|e| { + DomainError::internal_error("Dedup", format!("Failed to upsert chunks: {}", e)) + })?; } // chunk_hashes/chunk_sizes keep the full per-occurrence CDC sequence — diff --git a/src/infrastructure/services/encrypted_blob_backend.rs b/src/infrastructure/services/encrypted_blob_backend.rs index 47a3997a..e690d3b0 100644 --- a/src/infrastructure/services/encrypted_blob_backend.rs +++ b/src/infrastructure/services/encrypted_blob_backend.rs @@ -53,6 +53,19 @@ impl EncryptedBlobBackend { } } +/// Encrypt `data` into the on-disk layout: `[12-byte nonce][ciphertext + tag]`. +fn encrypt_bytes(cipher: &Aes256Gcm, data: &[u8]) -> Result { + let nonce = Aes256Gcm::generate_nonce(&mut OsRng); + let ciphertext = cipher + .encrypt(&nonce, data) + .map_err(|e| DomainError::internal_error("Encryption", format!("encrypt failed: {e}")))?; + + let mut encrypted = Vec::with_capacity(NONCE_SIZE + ciphertext.len()); + encrypted.extend_from_slice(nonce.as_slice()); + encrypted.extend_from_slice(&ciphertext); + Ok(Bytes::from(encrypted)) +} + impl BlobStorageBackend for EncryptedBlobBackend { fn initialize( &self, @@ -113,22 +126,34 @@ impl BlobStorageBackend for EncryptedBlobBackend { let hash = hash.to_string(); let cipher = self.cipher.clone(); Box::pin(async move { - // Encrypt in memory: nonce || ciphertext (includes GCM tag) - let nonce = Aes256Gcm::generate_nonce(&mut OsRng); - let ciphertext = cipher.encrypt(&nonce, data.as_ref()).map_err(|e| { - DomainError::internal_error("Encryption", format!("encrypt failed: {e}")) - })?; - - let mut encrypted = Vec::with_capacity(NONCE_SIZE + ciphertext.len()); - encrypted.extend_from_slice(nonce.as_slice()); - encrypted.extend_from_slice(&ciphertext); - - inner - .put_blob_from_bytes(&hash, Bytes::from(encrypted)) - .await + let encrypted = encrypt_bytes(&cipher, data.as_ref())?; + inner.put_blob_from_bytes(&hash, encrypted).await }) } + fn put_blob_from_bytes_unsynced( + &self, + hash: &str, + data: Bytes, + ) -> Pin> + Send + '_>> { + let inner = self.inner.clone(); + let hash = hash.to_string(); + let cipher = self.cipher.clone(); + Box::pin(async move { + let encrypted = encrypt_bytes(&cipher, data.as_ref())?; + inner.put_blob_from_bytes_unsynced(&hash, encrypted).await + }) + } + + fn sync_blobs( + &self, + hashes: &[String], + ) -> Pin> + Send + '_>> { + // Hashes key the *plaintext* content but address the same inner + // blobs, so the durability sweep forwards untouched. + self.inner.sync_blobs(hashes) + } + fn get_blob_stream( &self, hash: &str, diff --git a/src/infrastructure/services/local_blob_backend.rs b/src/infrastructure/services/local_blob_backend.rs index 926bbbc8..632701cb 100644 --- a/src/infrastructure/services/local_blob_backend.rs +++ b/src/infrastructure/services/local_blob_backend.rs @@ -66,6 +66,80 @@ async fn fsync_parent_dir(child_path: &Path) { /// Chunk size for streaming file reads (256 KB). const STREAM_CHUNK_SIZE: usize = 256 * 1024; +/// Max parallel blocking tasks for the [`fsync_paths_parallel`] sweep. +/// +/// Concurrent fsyncs let journaling filesystems coalesce barriers (ext4 +/// merges parallel fsyncs into shared journal commits), so a sweep over +/// thousands of chunk files costs a small fraction of issuing the same +/// fsyncs sequentially. +const SYNC_SWEEP_CONCURRENCY: usize = 16; + +/// Fsync every path in `paths`, spread over up to +/// [`SYNC_SWEEP_CONCURRENCY`] blocking-pool tasks. +/// +/// `strict` mirrors the two durability tiers already present in this +/// module: blob *files* must be durable (hard error on failure, like +/// `put_blob_from_bytes`), while *directory* fsyncs are best-effort +/// (logged warning, like [`fsync_parent_dir`]) — directories can't be +/// opened for fsync on every platform. +async fn fsync_paths_parallel(paths: Vec, strict: bool) -> Result<(), DomainError> { + if paths.is_empty() { + return Ok(()); + } + let group_size = paths.len().div_ceil(SYNC_SWEEP_CONCURRENCY); + let mut tasks = Vec::with_capacity(SYNC_SWEEP_CONCURRENCY); + for group in paths.chunks(group_size) { + let group = group.to_vec(); + tasks.push(tokio::task::spawn_blocking( + move || -> Result<(), (PathBuf, std::io::Error)> { + for path in &group { + let result = std::fs::File::open(path).and_then(|f| f.sync_all()); + if let Err(e) = result { + if strict { + return Err((path.clone(), e)); + } + tracing::warn!( + error = %e, + path = %path.display(), + "Blob sync sweep: best-effort fsync failed" + ); + } + } + Ok(()) + }, + )); + } + for task in tasks { + task.await + .map_err(|e| DomainError::internal_error("Blob", format!("sync sweep join: {e}")))? + .map_err(|(path, e)| { + DomainError::internal_error( + "Blob", + format!("sync sweep fsync of {} failed: {e}", path.display()), + ) + })?; + } + Ok(()) +} + +/// Create `blob_path` and write `data` into it. +/// +/// Returns the open file handle so the caller decides the durability tier +/// (fsync now vs. deferred batch sync), or `None` when the blob already +/// existed (idempotent skip — content-addressed, so identical by definition). +async fn write_blob_bytes(blob_path: &Path, data: &Bytes) -> Result, DomainError> { + if fs::try_exists(blob_path).await.unwrap_or(false) { + return Ok(None); + } + let mut file = fs::File::create(blob_path).await.map_err(|e| { + DomainError::internal_error("Blob", format!("Failed to create blob file: {}", e)) + })?; + file.write_all(data).await.map_err(|e| { + DomainError::internal_error("Blob", format!("Failed to write blob from bytes: {}", e)) + })?; + Ok(Some(file)) +} + /// Compile-time lookup table for the 256 two-digit lowercase hex prefixes ("00"…"ff"). static HEX_PREFIXES: [&str; 256] = [ "00", "01", "02", "03", "04", "05", "06", "07", "08", "09", "0a", "0b", "0c", "0d", "0e", "0f", @@ -222,37 +296,74 @@ impl BlobStorageBackend for LocalBlobBackend { let blob_path = self.blob_path(&hash); let size = data.len() as u64; - // Idempotent: if blob already exists, skip - if fs::try_exists(&blob_path).await.unwrap_or(false) { - return Ok(size); + // Same durability story as `put_blob`: the blob file is + // fsync'd before the parent directory is, so both the content + // and the dirent creation survive a power loss in the same + // step. (tokio's `sync_all` flushes its internal buffer + // before issuing the fsync.) + if let Some(file) = write_blob_bytes(&blob_path, &data).await? { + file.sync_all().await.map_err(|e| { + DomainError::internal_error("Blob", format!("Failed to fsync blob file: {}", e)) + })?; + drop(file); + fsync_parent_dir(&blob_path).await; } - // Write directly to blob path. `fs::write` is `create + - // write_all + close` — but the close on tokio::fs::File - // does NOT fsync, so we open explicitly to keep the - // `sync_all` call site obvious. Same durability story as - // `put_blob`: the blob file is fsync'd before the parent - // directory is, so both the content and the dirent - // creation survive a power loss in the same step. - let mut file = fs::File::create(&blob_path).await.map_err(|e| { - DomainError::internal_error("Blob", format!("Failed to create blob file: {}", e)) - })?; - file.write_all(&data).await.map_err(|e| { - DomainError::internal_error( - "Blob", - format!("Failed to write blob from bytes: {}", e), - ) - })?; - file.sync_all().await.map_err(|e| { - DomainError::internal_error("Blob", format!("Failed to fsync blob file: {}", e)) - })?; - drop(file); - fsync_parent_dir(&blob_path).await; + Ok(size) + }) + } + + fn put_blob_from_bytes_unsynced( + &self, + hash: &str, + data: Bytes, + ) -> Pin> + Send + '_>> { + let hash = hash.to_owned(); + Box::pin(async move { + let blob_path = self.blob_path(&hash); + let size = data.len() as u64; + + if let Some(mut file) = write_blob_bytes(&blob_path, &data).await? { + // flush surfaces write errors (e.g. ENOSPC) that tokio + // would otherwise swallow on drop. It does NOT fsync — + // durability comes from the caller's later `sync_blobs`. + file.flush().await.map_err(|e| { + DomainError::internal_error("Blob", format!("Failed to flush blob file: {}", e)) + })?; + } Ok(size) }) } + fn sync_blobs( + &self, + hashes: &[String], + ) -> Pin> + Send + '_>> { + let paths: Vec = hashes.iter().map(|h| self.blob_path(h)).collect(); + Box::pin(async move { + if paths.is_empty() { + return Ok(()); + } + + // Each distinct prefix directory is fsync'd exactly once — + // chunks of one upload land in at most 256 prefix dirs, so + // this replaces one dir fsync *per chunk* with ≤256 total. + let mut dirs: Vec = paths + .iter() + .filter_map(|p| p.parent().map(Path::to_path_buf)) + .collect(); + dirs.sort_unstable(); + dirs.dedup(); + + // Files first (hard requirement), then dirents (best-effort, + // same tier as fsync_parent_dir). + fsync_paths_parallel(paths, true).await?; + fsync_paths_parallel(dirs, false).await?; + Ok(()) + }) + } + fn get_blob_stream( &self, hash: &str, @@ -382,3 +493,93 @@ impl BlobStorageBackend for LocalBlobBackend { Some(self.blob_path(hash)) } } + +#[cfg(test)] +mod tests { + use super::*; + use futures::StreamExt; + use tempfile::TempDir; + + /// 64-char fake hash with the given 2-char prefix (selects the prefix dir). + fn fake_hash(prefix: &str) -> String { + format!("{prefix}{}", "0".repeat(62)) + } + + async fn read_blob(backend: &LocalBlobBackend, hash: &str) -> Vec { + let mut stream = backend.get_blob_stream(hash).await.unwrap(); + let mut data = Vec::new(); + while let Some(chunk) = stream.next().await { + data.extend_from_slice(&chunk.unwrap()); + } + data + } + + #[tokio::test] + async fn unsynced_write_then_sync_blobs_roundtrip() { + let tmp = TempDir::new().unwrap(); + let backend = LocalBlobBackend::new(tmp.path()); + backend.initialize().await.unwrap(); + + // Two different prefixes → exercises the distinct-parent-dir dedup. + let h1 = fake_hash("aa"); + let h2 = fake_hash("bb"); + backend + .put_blob_from_bytes_unsynced(&h1, Bytes::from_static(b"chunk one")) + .await + .unwrap(); + backend + .put_blob_from_bytes_unsynced(&h2, Bytes::from_static(b"chunk two")) + .await + .unwrap(); + + backend.sync_blobs(&[h1.clone(), h2.clone()]).await.unwrap(); + + assert!(backend.blob_exists(&h1).await.unwrap()); + assert!(backend.blob_exists(&h2).await.unwrap()); + assert_eq!(read_blob(&backend, &h1).await, b"chunk one"); + assert_eq!(read_blob(&backend, &h2).await, b"chunk two"); + } + + #[tokio::test] + async fn unsynced_write_is_idempotent() { + let tmp = TempDir::new().unwrap(); + let backend = LocalBlobBackend::new(tmp.path()); + backend.initialize().await.unwrap(); + + let hash = fake_hash("cc"); + let size1 = backend + .put_blob_from_bytes_unsynced(&hash, Bytes::from_static(b"same content")) + .await + .unwrap(); + let size2 = backend + .put_blob_from_bytes_unsynced(&hash, Bytes::from_static(b"same content")) + .await + .unwrap(); + + assert_eq!(size1, size2); + assert_eq!(read_blob(&backend, &hash).await, b"same content"); + } + + #[tokio::test] + async fn sync_blobs_fails_on_missing_blob() { + let tmp = TempDir::new().unwrap(); + let backend = LocalBlobBackend::new(tmp.path()); + backend.initialize().await.unwrap(); + + let missing = fake_hash("dd"); + assert!( + backend.sync_blobs(&[missing]).await.is_err(), + "sweeping a never-written blob must fail — the caller would \ + otherwise insert a PG row for a chunk that doesn't exist" + ); + } + + #[tokio::test] + async fn sync_blobs_empty_is_noop() { + let tmp = TempDir::new().unwrap(); + let backend = LocalBlobBackend::new(tmp.path()); + backend.initialize().await.unwrap(); + + backend.sync_blobs(&[]).await.unwrap(); + } +} diff --git a/src/infrastructure/services/migration_blob_backend.rs b/src/infrastructure/services/migration_blob_backend.rs index 40858fb8..75e0c716 100644 --- a/src/infrastructure/services/migration_blob_backend.rs +++ b/src/infrastructure/services/migration_blob_backend.rs @@ -121,6 +121,21 @@ impl BlobStorageBackend for MigrationBlobBackend { Box::pin(async move { self.target.put_blob_from_bytes(&hash, data).await }) } + /// Unsynced writes go to **target** only (same as the synced variant). + fn put_blob_from_bytes_unsynced( + &self, + hash: &str, + data: Bytes, + ) -> BoxFut<'_, Result> { + let hash = hash.to_string(); + Box::pin(async move { self.target.put_blob_from_bytes_unsynced(&hash, data).await }) + } + + /// Durability sweep goes to **target**, where unsynced writes land. + fn sync_blobs(&self, hashes: &[String]) -> BoxFut<'_, Result<(), DomainError>> { + self.target.sync_blobs(hashes) + } + /// Read from target first; fall back to source. fn get_blob_stream(&self, hash: &str) -> BoxFut<'_, Result> { let hash = hash.to_string();