Merge pull request #444 from AtalayaLabs/claude/vigilant-goodall-oismlc

Batch fsync for blob uploads: unsynced writes + parallel durability sweep

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<u64, DomainError>>;
 
+    /// 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<u64, DomainError>> {
+        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<BlobStream, DomainError>>;
 

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).
+//!   3. Bump ref_count for existing chunks (no disk I/O).
-//!   4. Bump ref_count for existing chunks (no disk I/O).
+//!   4. Read + write only *new* chunks to the blob backend (idempotent,
-//!   5. Single manifest INSERT (~few ms total).
+//!      no per-chunk fsync).
-//!   6. PG connection is never held during disk I/O.
+//!   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
+    /// Phase 1: Bumps `ref_count` for chunks that already exist (one
-    /// I/O saving for versioned files where most chunks are unchanged).
+    /// batched UPDATE, no disk I/O — the biggest saving for versioned
-    /// Uploads each new chunk and bumps `ref_count` for chunks that already
+    /// files where most chunks are unchanged).
-    /// exist, with up to [`CHUNK_UPLOAD_CONCURRENCY`] uploads in flight.
+    /// 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<Result<(), DomainError>> = 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<Result<(String, i64), DomainError>> = 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
+                    Ok((hash, length as i64))
-                    // 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(())
                 }
             })
             .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<String> = 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<i64> = 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 —

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<Bytes, DomainError> {
+    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 encrypted = encrypt_bytes(&cipher, data.as_ref())?;
-            let nonce = Aes256Gcm::generate_nonce(&mut OsRng);
+            inner.put_blob_from_bytes(&hash, encrypted).await
-            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
         })
     }
 
+    fn put_blob_from_bytes_unsynced(
+        &self,
+        hash: &str,
+        data: Bytes,
+    ) -> Pin<Box<dyn std::future::Future<Output = Result<u64, DomainError>> + 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<Box<dyn std::future::Future<Output = Result<(), DomainError>> + 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,

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<PathBuf>, 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<Option<File>, 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
+            // Same durability story as `put_blob`: the blob file is
-            if fs::try_exists(&blob_path).await.unwrap_or(false) {
+            // fsync'd before the parent directory is, so both the content
-                return Ok(size);
+            // 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 +
+            Ok(size)
-            // 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
+    fn put_blob_from_bytes_unsynced(
-            // directory is, so both the content and the dirent
+        &self,
-            // creation survive a power loss in the same step.
+        hash: &str,
-            let mut file = fs::File::create(&blob_path).await.map_err(|e| {
+        data: Bytes,
-                DomainError::internal_error("Blob", format!("Failed to create blob file: {}", e))
+    ) -> Pin<Box<dyn std::future::Future<Output = Result<u64, DomainError>> + Send + '_>> {
-            })?;
+        let hash = hash.to_owned();
-            file.write_all(&data).await.map_err(|e| {
+        Box::pin(async move {
-                DomainError::internal_error(
+            let blob_path = self.blob_path(&hash);
-                    "Blob",
+            let size = data.len() as u64;
-                    format!("Failed to write blob from bytes: {}", e),
+
-                )
+            if let Some(mut file) = write_blob_bytes(&blob_path, &data).await? {
-            })?;
+                // flush surfaces write errors (e.g. ENOSPC) that tokio
-            file.sync_all().await.map_err(|e| {
+                // would otherwise swallow on drop. It does NOT fsync —
-                DomainError::internal_error("Blob", format!("Failed to fsync blob file: {}", e))
+                // durability comes from the caller's later `sync_blobs`.
-            })?;
+                file.flush().await.map_err(|e| {
-            drop(file);
+                    DomainError::internal_error("Blob", format!("Failed to flush blob file: {}", e))
-            fsync_parent_dir(&blob_path).await;
+                })?;
+            }
 
             Ok(size)
         })
     }
 
+    fn sync_blobs(
+        &self,
+        hashes: &[String],
+    ) -> Pin<Box<dyn std::future::Future<Output = Result<(), DomainError>> + Send + '_>> {
+        let paths: Vec<PathBuf> = 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<PathBuf> = 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<u8> {
+        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();
+    }
+}

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<u64, DomainError>> {
+        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<BlobStream, DomainError>> {
         let hash = hash.to_string();