diff --git a/.dockerignore b/.dockerignore index 3b855bb0..eb838161 100644 --- a/.dockerignore +++ b/.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 diff --git a/Dockerfile b/Dockerfile index 59dc48e2..e50b8107 100644 --- a/Dockerfile +++ b/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 diff --git a/entrypoint.sh b/entrypoint.sh index 1b757c10..f0490dae 100755 --- a/entrypoint.sh +++ b/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) diff --git a/src/application/ports/blob_storage_ports.rs b/src/application/ports/blob_storage_ports.rs index f7f7092c..cf3be3dc 100644 --- a/src/application/ports/blob_storage_ports.rs +++ b/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>; - /// 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(()) }) } diff --git a/src/infrastructure/services/dedup_service.rs b/src/infrastructure/services/dedup_service.rs index 53e09834..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 @@ -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 = new_ops.iter().map(|(h, _, _)| h.clone()).collect(); - let new_sizes: Vec = 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> = 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 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 = 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 = 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", ) diff --git a/src/infrastructure/services/encrypted_blob_backend.rs b/src/infrastructure/services/encrypted_blob_backend.rs index f2620c2e..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,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> + 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) } diff --git a/src/infrastructure/services/local_blob_backend.rs b/src/infrastructure/services/local_blob_backend.rs index 7bf57c85..632701cb 100644 --- a/src/infrastructure/services/local_blob_backend.rs +++ b/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, 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, 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 { - 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> + Send + '_>> { let paths: Vec = 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 = 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> = 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 = 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 { + 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();