Merge origin/main into claude/upbeat-wright-btruto

Resolves the four-file conflict with PR #444, which landed a parallel
implementation of the same batch-fsync design (unsynced chunk writes plus
one sync_blobs durability barrier). Both sides converged on identical
trait signatures; main's version is kept for all overlapping code since
it is already merged and carries unit tests:

- blob_storage_ports.rs: doc-comment-only conflict, main's docs kept
- dedup_service.rs: main's pipeline kept. The naive textual merge would
  have retained BOTH durability-barrier/INSERT blocks, double-counting
  ref_count for every new chunk; the duplicated block from this branch
  is dropped
- encrypted_blob_backend.rs, local_blob_backend.rs: main's
  implementations kept (fsync_paths_parallel + tests); this branch's
  equivalent helpers (fsync_dir, fsync_blob_file, SYNC_BLOBS_CONCURRENCY)
  are referenced nowhere else and are dropped

Everything unique to this branch (streaming PROPFIND, range requests,
batch ID resolution, CalDAV/CardDAV indexed lookups, transcode and
breadcrumb changes) merges cleanly and is unchanged.

https://claude.ai/code/session_01GpprjxjtXFYLfXNkoKnHuL

.dockerignore

@@ -21,11 +21,19 @@ rootless-compose.yml
 storage/
 
 # Documentation
-doc/
+docs/
 *.md
 LICENSE
 CODEOWNERS
 
+# Project assets and tooling not needed by the build
+# (the Dockerfile only COPYs src, static, migrations, templates, build.rs,
+#  Cargo.*, and entrypoint.sh; everything else is dead weight in the context)
+images/
+charts/
+tools/
+tests/
+
 # Miscellaneous
 .github/
 .env

Dockerfile

@@ -1,7 +1,10 @@
 # ─── Stage 1: Shared build base (avoids duplicate apk install) ────────────────
 FROM rust:1.94.1-alpine3.23 AS base
+# sqlx's postgres driver speaks the wire protocol in pure Rust (no pq-sys in
+# Cargo.lock) and TLS goes through rustls, so libpq headers are never needed at
+# build time. perl/make/gcc/musl-dev remain for the C builds of aws-lc-sys.
 RUN apk --no-cache upgrade && \
-    apk add --no-cache musl-dev pkgconfig postgresql-dev gcc perl make
+    apk add --no-cache musl-dev pkgconfig gcc perl make
 
 # ─── Stage 2: Cache dependencies ─────────────────────────────────────────────
 FROM base AS cacher
@@ -49,9 +52,10 @@ LABEL org.opencontainers.image.title="OxiCloud" \
       org.opencontainers.image.licenses="MIT"
 
 # Install only necessary runtime dependencies and update packages
-# su-exec is needed by the entrypoint to drop privileges after fixing volume permissions
+# su-exec is needed by the entrypoint to drop privileges after fixing volume permissions.
+# No libpq: the pure-Rust sqlx postgres driver never links it.
 RUN apk --no-cache upgrade && \
-    apk add --no-cache libgcc ca-certificates libpq tzdata su-exec && \
+    apk add --no-cache libgcc ca-certificates tzdata su-exec && \
     addgroup -g 1001 -S oxicloud && \
     adduser -u 1001 -S oxicloud -G oxicloud
 

entrypoint.sh

@@ -9,14 +9,26 @@ set -e
 STORAGE_DIR="/app/storage"
 STATIC_DIR="/app/static"
 
-# Ensure the storage directory exists and is writable by oxicloud
-if [ -d "$STORAGE_DIR" ] && [ "$(id -u)" -eq 0 ]; then
-    chown -R oxicloud:oxicloud "$STORAGE_DIR"
-fi
+# Recursively chown DIR to the oxicloud user, but only when its top-level
+# entry is not already owned by that user. The storage volume is a
+# content-addressable blob store that can hold millions of objects; a blind
+# "chown -R" on every boot would re-stat and rewrite the inode of every blob,
+# turning startup into minutes of disk I/O. Checking the root entry is the
+# cheap idempotent guard: the first boot fixes a freshly mounted (root-owned)
+# volume, and every later boot is a no-op.
+ensure_owned() {
+    dir="$1"
+    if [ -d "$dir" ] && [ "$(stat -c %u "$dir")" != "$OXI_UID" ]; then
+        chown -R oxicloud:oxicloud "$dir"
+    fi
+}
 
-# Ensure static directory is readable
-if [ -d "$STATIC_DIR" ] && [ "$(id -u)" -eq 0 ]; then
-    chown -R oxicloud:oxicloud "$STATIC_DIR"
+# Only root can chown; when started unprivileged the volume permissions are
+# assumed to be correct already.
+if [ "$(id -u)" -eq 0 ]; then
+    OXI_UID="$(id -u oxicloud)"
+    ensure_owned "$STORAGE_DIR"
+    ensure_owned "$STATIC_DIR"
 fi
 
 # Drop privileges and exec the main binary (or whatever was passed as CMD)

src/application/ports/blob_storage_ports.rs

@@ -60,17 +60,18 @@ 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 a durability barrier**.
+    /// Store a blob from in-memory bytes **without forcing durability**.
     ///
-    /// The CDC chunk path writes thousands of small chunks per file; paying
-    /// two fsyncs per chunk (file + parent dir) put ~8 000 fsyncs on the
-    /// critical path of a 1 GB upload. Callers using this MUST issue one
-    /// [`Self::sync_blobs`] barrier over the written hashes before
-    /// persisting any record that references them (the chunk manifest).
+    /// 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 [`Self::put_blob_from_bytes`] — correct for
-    /// remote backends (S3, Azure) where a successful PUT is already
-    /// durable and the "unsynced" notion does not exist.
+    /// 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,
@@ -79,14 +80,15 @@ pub trait BlobStorageBackend: Send + Sync + 'static {
         self.put_blob_from_bytes(hash, data)
     }
 
-    /// Durability barrier for blobs previously written with
-    /// [`Self::put_blob_from_bytes_unsynced`]: when this resolves, the
-    /// listed blobs and their directory entries survive a power loss.
+    /// Make previously written blobs durable in one batched operation.
     ///
-    /// Default: no-op — matches the default `put_blob_from_bytes_unsynced`,
-    /// which is already durable on completion.
-    fn sync_blobs(&self, hashes: &[String]) -> BoxFut<'_, Result<(), DomainError>> {
-        let _ = hashes;
+    /// 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(()) })
     }
 

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
@@ -535,22 +545,16 @@ impl DedupService {
             .filter(|c| !existing_hashes.contains(&c.hash))
             .map(|c| (c.hash.clone(), c.offset as u64, c.length))
             .collect();
-        let new_hashes: Vec<String> = new_ops.iter().map(|(h, _, _)| h.clone()).collect();
-        let new_sizes: Vec<i64> = new_ops.iter().map(|(_, _, l)| *l as i64).collect();
 
         let source = Arc::new(std::fs::File::open(source_path).map_err(|e| {
             DomainError::internal_error("Dedup", format!("Failed to open source file: {}", e))
         })?);
 
-        // Chunks are written WITHOUT a per-chunk durability barrier — the
-        // former two fsyncs per chunk put ~8 000 fsyncs on the critical path
-        // of a 1 GB upload. One `sync_blobs` barrier below makes every chunk
-        // durable before the manifest (the only record referencing them) is
-        // committed, so the durability contract observed by the caller is
-        // unchanged. A crash before the barrier leaves orphan chunk files
-        // with no DB row — the same orphan class the per-chunk scheme had,
-        // just a wider window.
-        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 backend = backend.clone();
@@ -574,30 +578,36 @@ impl DedupService {
                     backend
                         .put_blob_from_bytes_unsynced(&hash, Bytes::from(bytes))
                         .await?;
-                    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_hashes.is_empty() {
-            // Durability barrier: every new chunk (and its dirent) is on
-            // stable storage before any DB row references it.
+        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?;
 
-            // One batched upsert for all new chunks (was one round-trip per
-            // chunk, interleaved with the uploads). ON CONFLICT covers a
-            // concurrent uploader inserting the same brand-new chunk between
-            // the existence check above and this INSERT.
+            // ── 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 t.hash, t.size, 1
-                 FROM unnest($1::text[], $2::bigint[]) AS t(hash, size)
+                 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",
             )

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,19 +126,8 @@ 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
         })
     }
 
@@ -138,21 +140,8 @@ 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);
-
-            // Delegate the relaxed-durability variant so encryption over
-            // the local backend keeps the batched `sync_blobs` barrier.
-            inner
-                .put_blob_from_bytes_unsynced(&hash, Bytes::from(encrypted))
-                .await
+            let encrypted = encrypt_bytes(&cipher, data.as_ref())?;
+            inner.put_blob_from_bytes_unsynced(&hash, encrypted).await
         })
     }
 
@@ -160,7 +149,8 @@ impl BlobStorageBackend for EncryptedBlobBackend {
         &self,
         hashes: &[String],
     ) -> Pin<Box<dyn std::future::Future<Output = Result<(), DomainError>> + Send + '_>> {
-        // Pure passthrough — encryption does not change blob addressing.
+        // Hashes key the *plaintext* content but address the same inner
+        // blobs, so the durability sweep forwards untouched.
         self.inner.sync_blobs(hashes)
     }
 

src/infrastructure/services/local_blob_backend.rs

@@ -36,16 +36,11 @@ async fn fsync_parent_dir(child_path: &Path) {
     let Some(parent) = child_path.parent() else {
         return;
     };
-    fsync_dir(parent).await;
-}
-
-/// Fsync a directory (best-effort — see [`fsync_parent_dir`]).
-async fn fsync_dir(dir: &Path) {
-    let dir_owned = dir.to_owned();
+    let parent = parent.to_owned();
     // std::fs (synchronous) opens directories reliably on Linux/macOS;
     // do it on the blocking pool so we don't park the tokio worker.
     let result = tokio::task::spawn_blocking(move || -> std::io::Result<()> {
-        let dir = std::fs::File::open(&dir_owned)?;
+        let dir = std::fs::File::open(&parent)?;
         dir.sync_all()
     })
     .await;
@@ -54,37 +49,96 @@ async fn fsync_dir(dir: &Path) {
         Ok(Err(e)) => {
             tracing::warn!(
                 error = %e,
-                path = %dir.display(),
-                "Blob dir fsync failed (rename durability not guaranteed)"
+                path = %child_path.display(),
+                "Blob parent-dir fsync failed (rename durability not guaranteed)"
             );
         }
         Err(e) => {
             tracing::warn!(
                 error = %e,
-                path = %dir.display(),
-                "Blob dir fsync task join failed"
+                path = %child_path.display(),
+                "Blob parent-dir fsync task join failed"
             );
         }
     }
 }
 
-/// Open + fsync a blob file so its content survives a power loss.
-async fn fsync_blob_file(path: &Path) -> Result<(), DomainError> {
-    let file = fs::File::open(path).await.map_err(|e| {
-        DomainError::internal_error("Blob", format!("Failed to open blob for fsync: {}", e))
-    })?;
-    file.sync_all().await.map_err(|e| {
-        DomainError::internal_error("Blob", format!("Failed to fsync blob file: {}", e))
-    })?;
+/// 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(())
 }
 
-/// Concurrent fsyncs in flight during a [`BlobStorageBackend::sync_blobs`]
-/// barrier — lets the kernel coalesce writeback across many small chunks.
-const SYNC_BLOBS_CONCURRENCY: usize = 16;
-
-/// Chunk size for streaming file reads (256 KB).
-const STREAM_CHUNK_SIZE: usize = 256 * 1024;
+/// 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] = [
@@ -137,28 +191,6 @@ impl LocalBlobBackend {
     pub fn blob_root(&self) -> &Path {
         &self.blob_root
     }
-
-    /// Write `data` to `blob_path` unless it already exists (idempotent
-    /// dedup-skip). Returns `true` when a new file was written. No
-    /// durability barrier here — callers choose between the per-file
-    /// fsync (`put_blob_from_bytes`) and the batched `sync_blobs` barrier
-    /// (`put_blob_from_bytes_unsynced`).
-    async fn write_blob_bytes(&self, blob_path: &Path, data: &Bytes) -> Result<bool, DomainError> {
-        if fs::try_exists(blob_path).await.unwrap_or(false) {
-            return Ok(false);
-        }
-
-        // `fs::write` is `create + write_all + close` — but the close on
-        // tokio::fs::File does NOT fsync, so durability is layered on
-        // explicitly by the caller.
-        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(true)
-    }
 }
 
 impl BlobStorageBackend for LocalBlobBackend {
@@ -262,15 +294,22 @@ impl BlobStorageBackend for LocalBlobBackend {
         let hash = hash.to_owned();
         Box::pin(async move {
             let blob_path = self.blob_path(&hash);
-            if self.write_blob_bytes(&blob_path, &data).await? {
-                // 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.
-                fsync_blob_file(&blob_path).await?;
+            let size = data.len() as u64;
+
+            // 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;
             }
-            Ok(data.len() as u64)
+
+            Ok(size)
         })
     }
 
@@ -282,11 +321,18 @@ impl BlobStorageBackend for LocalBlobBackend {
         let hash = hash.to_owned();
         Box::pin(async move {
             let blob_path = self.blob_path(&hash);
-            // No fsync here — the CDC chunk path issues one `sync_blobs`
-            // barrier over all written chunks before the manifest row that
-            // references them is committed.
-            self.write_blob_bytes(&blob_path, &data).await?;
-            Ok(data.len() as u64)
+            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)
         })
     }
 
@@ -295,31 +341,25 @@ impl BlobStorageBackend for LocalBlobBackend {
         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();
-        // One fsync per DISTINCT parent directory — the hash-sharded layout
-        // spreads chunks over at most 256 dirs, so this replaces the former
-        // one-dir-fsync-per-chunk. Best-effort, same as `put_blob`.
-        let parents: std::collections::HashSet<PathBuf> = paths
-            .iter()
-            .filter_map(|p| p.parent().map(Path::to_owned))
-            .collect();
         Box::pin(async move {
-            // Fsync every blob file concurrently — the kernel coalesces
-            // batched writeback far better than an fsync after every write
-            // on the upload critical path.
-            use futures::stream::{self, StreamExt};
-            let results: Vec<Result<(), DomainError>> = stream::iter(paths)
-                .map(|path| async move { fsync_blob_file(&path).await })
-                .buffer_unordered(SYNC_BLOBS_CONCURRENCY)
-                .collect()
-                .await;
-            for result in results {
-                result?;
+            if paths.is_empty() {
+                return Ok(());
             }
 
-            for parent in &parents {
-                fsync_dir(parent).await;
-            }
+            // 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(())
         })
     }
@@ -453,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();