diff --git a/Cargo.lock b/Cargo.lock index f5c28102..9b32a950 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -542,6 +542,20 @@ dependencies = [ "syn", ] +[[package]] +name = "dashmap" +version = "6.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5041cc499144891f3790297212f32a74fb938e5136a14943f338ef9e0ae276cf" +dependencies = [ + "cfg-if", + "crossbeam-utils", + "hashbrown 0.14.5", + "lock_api", + "once_cell", + "parking_lot_core", +] + [[package]] name = "der" version = "0.7.10" @@ -1014,6 +1028,12 @@ dependencies = [ "tracing", ] +[[package]] +name = "hashbrown" +version = "0.14.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e5274423e17b7c9fc20b6e7e208532f9b19825d82dfd615708b70edd83df41f1" + [[package]] name = "hashbrown" version = "0.15.5" @@ -1748,6 +1768,7 @@ dependencies = [ "base64", "bytes", "chrono", + "dashmap", "dotenvy", "flate2", "fs2", diff --git a/Cargo.toml b/Cargo.toml index 28063a8e..74eea17a 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -51,6 +51,7 @@ rayon = "1.10" infer = "0.19" async-compression = { version = "0.4", features = ["tokio", "gzip"] } async_zip = { version = "0.0.18", features = ["tokio", "deflate"] } +dashmap = "6" [features] default = [] diff --git a/src/infrastructure/services/chunked_upload_service.rs b/src/infrastructure/services/chunked_upload_service.rs index cc8c5e57..288fa3c7 100644 --- a/src/infrastructure/services/chunked_upload_service.rs +++ b/src/infrastructure/services/chunked_upload_service.rs @@ -17,6 +17,7 @@ use async_trait::async_trait; use chrono::{DateTime, Utc}; +use dashmap::DashMap; use serde::{Deserialize, Serialize}; use sha2::{Digest, Sha256}; use std::collections::HashMap; @@ -25,7 +26,6 @@ use std::sync::Arc; use std::time::Duration; use tokio::fs::{self, File}; use tokio::io::AsyncWriteExt; -use tokio::sync::RwLock; use uuid::Uuid; use crate::application::ports::chunked_upload_ports::{ @@ -157,8 +157,8 @@ impl UploadSession { /// Persist the full session metadata once (on create). async fn persist_metadata(&self) -> Result<(), String> { let path = self.temp_dir.join(SESSION_META_FILE); - let json = - serde_json::to_vec(self).map_err(|e| format!("Failed to serialise session: {e}"))?; + let json = serde_json::to_vec(self) + .map_err(|e| format!("Failed to serialise session: {e}"))?; // Atomic write: write to .tmp then rename let tmp = self.temp_dir.join("session.json.tmp"); fs::write(&tmp, &json) @@ -186,8 +186,13 @@ impl UploadSession { // ─── Service ───────────────────────────────────────────────────────────────── /// Chunked Upload Service +/// +/// Uses `DashMap` (sharded concurrent map) instead of a global `RwLock` +/// so that operations on independent upload sessions never contend with each +/// other. Disk I/O (temp-dir cleanup) is always performed **outside** any +/// map lock to avoid blocking concurrent uploads. pub struct ChunkedUploadService { - sessions: Arc>>, + sessions: Arc>, temp_base_dir: PathBuf, } @@ -203,12 +208,14 @@ impl ChunkedUploadService { let recovered_count = recovered.len(); let service = Self { - sessions: Arc::new(RwLock::new(recovered)), + sessions: Arc::new(DashMap::from_iter(recovered)), temp_base_dir, }; if recovered_count > 0 { - tracing::info!("♻️ Recovered {recovered_count} chunked-upload session(s) from disk"); + tracing::info!( + "♻️ Recovered {recovered_count} chunked-upload session(s) from disk" + ); } // Start cleanup task @@ -225,7 +232,7 @@ impl ChunkedUploadService { /// Used only by `AppState::default()` (stub wiring). pub fn new_stub(temp_base_dir: PathBuf) -> Self { Self { - sessions: Arc::new(RwLock::new(HashMap::new())), + sessions: Arc::new(DashMap::new()), temp_base_dir, } } @@ -319,7 +326,7 @@ impl ChunkedUploadService { /// Background task to clean expired sessions async fn cleanup_loop( - sessions: Arc>>, + sessions: Arc>, temp_base_dir: PathBuf, ) { let mut interval = tokio::time::interval(Duration::from_secs(3600)); // Every hour @@ -327,23 +334,20 @@ impl ChunkedUploadService { loop { interval.tick().await; - let expired: Vec = { - let sessions = sessions.read().await; - sessions - .iter() - .filter(|(_, s)| s.is_expired()) - .map(|(id, _)| id.clone()) - .collect() - }; + // Collect expired session ids + temp dirs (lock-free iteration) + let expired: Vec<(String, PathBuf)> = sessions + .iter() + .filter(|entry| entry.value().is_expired()) + .map(|entry| (entry.key().clone(), entry.value().temp_dir.clone())) + .collect(); - for id in expired { - let mut sessions = sessions.write().await; - if let Some(session) = sessions.remove(&id) { - if let Err(e) = fs::remove_dir_all(&session.temp_dir).await { - tracing::warn!("Failed to cleanup expired upload {}: {}", id, e); - } else { - tracing::info!("🧹 Cleaned expired upload session: {}", id); - } + // Remove from map (microseconds per entry) then clean disk OUTSIDE lock + for (id, temp_dir) in expired { + sessions.remove(&id); + if let Err(e) = fs::remove_dir_all(&temp_dir).await { + tracing::warn!("Failed to cleanup expired upload {}: {}", id, e); + } else { + tracing::info!("🧹 Cleaned expired upload session: {}", id); } } @@ -354,7 +358,6 @@ impl ChunkedUploadService { if path.is_dir() { let dir_name = path.file_name().and_then(|n| n.to_str()).unwrap_or(""); - let sessions = sessions.read().await; if !sessions.contains_key(dir_name) && let Ok(metadata) = fs::metadata(&path).await && let Ok(modified) = metadata.modified() @@ -433,10 +436,7 @@ impl ChunkedUploadService { session.persist_metadata().await?; session.persist_progress().await?; - { - let mut sessions = self.sessions.write().await; - sessions.insert(upload_id.clone(), session); - } + self.sessions.insert(upload_id.clone(), session); tracing::info!( "📤 Created chunked upload session: {} ({} chunks, {} bytes each)", @@ -463,8 +463,7 @@ impl ChunkedUploadService { ) -> Result { // Validate session exists and chunk index is valid let (chunk_path, expected_size) = { - let sessions = self.sessions.read().await; - let session = sessions + let session = self.sessions .get(upload_id) .ok_or_else(|| format!("Upload session not found: {}", upload_id))?; @@ -501,10 +500,11 @@ impl ChunkedUploadService { // worker free for other connections. if let Some(ref expected_checksum) = checksum { let data_clone = data.clone(); // Bytes::clone is O(1) — just an Arc increment - let actual_checksum = - tokio::task::spawn_blocking(move || format!("{:x}", md5::compute(&data_clone))) - .await - .map_err(|e| format!("MD5 checksum task failed: {e}"))?; + let actual_checksum = tokio::task::spawn_blocking(move || { + format!("{:x}", md5::compute(&data_clone)) + }) + .await + .map_err(|e| format!("MD5 checksum task failed: {e}"))?; if actual_checksum != *expected_checksum { return Err(format!( @@ -523,12 +523,11 @@ impl ChunkedUploadService { .await .map_err(|e| format!("Failed to write chunk: {e}"))?; - // Update session state — keep write lock as short as possible (RAM only). - // Disk I/O (persist_progress) is done AFTER releasing the lock so - // concurrent uploads across all sessions are never blocked by I/O. + // Update session state — DashMap shard lock held only for RAM updates (~µs). + // Disk I/O (persist_progress) is done AFTER the ref is dropped so + // concurrent uploads to other sessions are never blocked. let (bytes_received, progress, is_complete, persist_path, persist_bitmask) = { - let mut sessions = self.sessions.write().await; - let session = sessions + let mut session = self.sessions .get_mut(upload_id) .ok_or_else(|| "Session disappeared".to_string())?; @@ -548,7 +547,7 @@ impl ChunkedUploadService { path, bitmask, ) - }; // Write lock released here — held only for RAM updates (~microseconds) + }; // DashMap shard ref dropped here — held only for RAM updates (~µs) // Persist bitmask to disk OUTSIDE the lock — no longer blocks other uploads if let Err(e) = fs::write(&persist_path, &persist_bitmask).await { @@ -572,9 +571,11 @@ impl ChunkedUploadService { } /// Get upload status - async fn get_status_inner(&self, upload_id: &str) -> Result { - let sessions = self.sessions.read().await; - let session = sessions + async fn get_status_inner( + &self, + upload_id: &str, + ) -> Result { + let session = self.sessions .get(upload_id) .ok_or_else(|| format!("Upload session not found: {}", upload_id))?; @@ -608,22 +609,23 @@ impl ChunkedUploadService { &self, upload_id: &str, ) -> Result<(PathBuf, String, Option, String, u64, String), String> { - // Get session and validate completion + // Get session and validate completion. + // Clone the session data and drop the DashMap ref immediately + // so the shard is not held during the expensive assembly step. let session = { - let sessions = self.sessions.read().await; - let session = sessions + let entry = self.sessions .get(upload_id) .ok_or_else(|| format!("Upload session not found: {}", upload_id))?; - if !session.is_complete() { - let pending = session.pending_chunks(); + if !entry.is_complete() { + let pending = entry.pending_chunks(); return Err(format!( "Upload not complete. Missing chunks: {:?}", pending )); } - session.clone() + entry.clone() }; // Assemble file with hash-on-write. @@ -638,17 +640,12 @@ impl ChunkedUploadService { let chunks_meta: Vec<(usize, PathBuf)> = session .chunks .iter() - .map(|c| { - ( - c.index, - session.temp_dir.join(format!("chunk_{:06}", c.index)), - ) - }) + .map(|c| (c.index, session.temp_dir.join(format!("chunk_{:06}", c.index)))) .collect(); let total_size = session.total_size; let hash = tokio::task::spawn_blocking(move || -> Result { - use std::io::{BufWriter as StdBufWriter, Read, Write}; + use std::io::{Read, Write, BufWriter as StdBufWriter}; let raw_output = std::fs::OpenOptions::new() .create(true) @@ -716,21 +713,27 @@ impl ChunkedUploadService { )) } - /// Finalize upload: remove session from RAM and clean up temp directory + /// Finalize upload: remove session from RAM, then clean disk OUTSIDE lock. async fn finalize_upload_inner(&self, upload_id: &str) -> Result<(), String> { - let mut sessions = self.sessions.write().await; - if let Some(session) = sessions.remove(upload_id) - && let Err(e) = fs::remove_dir_all(&session.temp_dir).await - { - tracing::warn!("Failed to cleanup upload {}: {}", upload_id, e); + // Remove from map (~µs) — releases shard immediately + let removed = self.sessions.remove(upload_id).map(|(_, s)| s); + + // Disk I/O happens with NO lock held + if let Some(session) = removed { + if let Err(e) = fs::remove_dir_all(&session.temp_dir).await { + tracing::warn!("Failed to cleanup upload {}: {}", upload_id, e); + } } Ok(()) } - /// Cancel an upload and cleanup + /// Cancel an upload and cleanup — disk I/O outside lock. async fn cancel_upload_inner(&self, upload_id: &str) -> Result<(), String> { - let mut sessions = self.sessions.write().await; - if let Some(session) = sessions.remove(upload_id) { + // Remove from map (~µs) + let removed = self.sessions.remove(upload_id).map(|(_, s)| s); + + // Disk I/O with NO lock held + if let Some(session) = removed { if let Err(e) = fs::remove_dir_all(&session.temp_dir).await { tracing::warn!("Failed to cleanup cancelled upload {}: {}", upload_id, e); } @@ -746,7 +749,7 @@ impl ChunkedUploadService { /// Get active session count (for monitoring) pub async fn active_sessions(&self) -> usize { - self.sessions.read().await.len() + self.sessions.len() } } @@ -917,7 +920,8 @@ mod tests { }; let json = serde_json::to_vec(&session).expect("serialise"); - let restored: UploadSession = serde_json::from_slice(&json).expect("deserialise"); + let restored: UploadSession = + serde_json::from_slice(&json).expect("deserialise"); assert_eq!(restored.id, session.id); assert_eq!(restored.filename, session.filename); @@ -992,9 +996,7 @@ mod tests { let recovered = ChunkedUploadService::recover_sessions(&base).await; assert_eq!(recovered.len(), 1); - let session = recovered - .get(&upload_id) - .expect("session must be recovered"); + let session = recovered.get(&upload_id).expect("session must be recovered"); assert_eq!(session.filename, "bigfile.bin"); assert_eq!(session.folder_id, Some("folder-x".into())); assert_eq!(session.chunks[0].status, ChunkStatus::Complete); @@ -1049,8 +1051,10 @@ mod tests { assert!(status.pending_chunks.is_empty()); // 4. Complete (assemble) - let (path, filename, _folder, _ct, size, hash) = - service.complete_upload_inner(&id).await.expect("complete"); + let (path, filename, _folder, _ct, size, hash) = service + .complete_upload_inner(&id) + .await + .expect("complete"); assert_eq!(filename, "test.txt"); assert_eq!(size, 1024); assert!(!hash.is_empty()); @@ -1074,13 +1078,7 @@ mod tests { let service = ChunkedUploadService::new(base.clone()).await; let resp = service - .create_session_inner( - "x.bin".into(), - None, - "application/octet-stream".into(), - 512, - Some(512), - ) + .create_session_inner("x.bin".into(), None, "application/octet-stream".into(), 512, Some(512)) .await .expect("create"); @@ -1158,18 +1156,12 @@ mod tests { chunk_size: 512, chunks: vec![ ChunkInfo { - index: 0, - offset: 0, - size: 512, - status: ChunkStatus::Pending, - checksum: None, + index: 0, offset: 0, size: 512, + status: ChunkStatus::Pending, checksum: None, }, ChunkInfo { - index: 1, - offset: 512, - size: 512, - status: ChunkStatus::Pending, - checksum: None, + index: 1, offset: 512, size: 512, + status: ChunkStatus::Pending, checksum: None, }, ], created_at: Utc::now(), @@ -1180,20 +1172,14 @@ mod tests { // Write metadata let json = serde_json::to_vec(&session).unwrap(); - fs::write(session_dir.join(SESSION_META_FILE), &json) - .await - .unwrap(); + fs::write(session_dir.join(SESSION_META_FILE), &json).await.unwrap(); // Write progress marking both chunks complete let bitmask = vec![0b00000011u8]; // bits 0 and 1 - fs::write(session_dir.join(PROGRESS_FILE), &bitmask) - .await - .unwrap(); + fs::write(session_dir.join(PROGRESS_FILE), &bitmask).await.unwrap(); // But only create chunk_000000 on disk — chunk_000001 is "missing" - fs::write(session_dir.join("chunk_000000"), &[0u8; 512]) - .await - .unwrap(); + fs::write(session_dir.join("chunk_000000"), &[0u8; 512]).await.unwrap(); let recovered = ChunkedUploadService::recover_sessions(&base).await; let s = recovered.get("partial-session").expect("must be recovered"); diff --git a/src/infrastructure/services/dedup_service.rs b/src/infrastructure/services/dedup_service.rs index 5f2bf34a..df1c71ea 100644 --- a/src/infrastructure/services/dedup_service.rs +++ b/src/infrastructure/services/dedup_service.rs @@ -13,12 +13,21 @@ //! ``` //! //! The dedup index lives in PostgreSQL (`storage.blobs`) — no in-memory -//! HashMap, no JSON file, no WAL. All concurrency is handled by -//! `SELECT … FOR UPDATE` and PostgreSQL transactions. +//! HashMap, no JSON file, no WAL. +//! +//! **Write-first strategy** (store_bytes / store_from_file): +//! 1. Write/move the blob file to disk *before* touching PostgreSQL. +//! 2. Single `INSERT … ON CONFLICT … RETURNING ref_count` upsert +//! (~2-4 ms) — no explicit transaction, no `SELECT FOR UPDATE`. +//! 3. PG connection is never held during disk I/O. +//! +//! `remove_reference` retains `SELECT … FOR UPDATE` inside a short +//! transaction because it must atomically decide whether to delete the +//! row *and* the blob file. //! //! Benefits: //! - ACID durability — crash-safe, zero orphaned index entries -//! - TOCTOU-free — `SELECT … FOR UPDATE` serialises concurrent mutations +//! - PG connections never blocked by disk I/O (write-first) //! - 30-50% storage reduction typical //! - Faster uploads for existing content (instant dedup) @@ -169,8 +178,10 @@ impl DedupService { /// Store content with deduplication (from bytes). /// - /// Uses `SELECT … FOR UPDATE` + `INSERT … ON CONFLICT` for atomic - /// upsert — completely TOCTOU-free. + /// **Write-first strategy**: the blob file is written to disk *before* + /// touching PostgreSQL, so the PG connection is never held during I/O. + /// The database operation is a single `INSERT … ON CONFLICT` upsert + /// (~2-4 ms) instead of `SELECT FOR UPDATE` + write + commit. /// /// **Guard**: rejects payloads >10 MB. Large content must go through /// `store_from_file` which streams from disk with constant RAM. @@ -192,110 +203,79 @@ impl DedupService { let size = content.len() as u64; let hash = Self::hash_bytes(content); - - // Begin transaction — all index mutations happen atomically - let mut tx = self.pool.begin().await.map_err(|e| { - DomainError::internal_error("Dedup", format!("Failed to begin transaction: {}", e)) - })?; - - // SELECT FOR UPDATE: locks the row if it exists, preventing - // concurrent remove_reference from deleting it mid-operation - let existing = sqlx::query_scalar::<_, i32>( - "SELECT ref_count FROM storage.blobs WHERE hash = $1 FOR UPDATE", - ) - .bind(&hash) - .fetch_optional(&mut *tx) - .await - .map_err(|e| { - DomainError::internal_error("Dedup", format!("Failed to check blob: {}", e)) - })?; - - if existing.is_some() { - // Blob exists — just increment ref_count (still under row lock) - sqlx::query("UPDATE storage.blobs SET ref_count = ref_count + 1 WHERE hash = $1") - .bind(&hash) - .execute(&mut *tx) - .await - .map_err(|e| { - DomainError::internal_error( - "Dedup", - format!("Failed to increment ref_count: {}", e), - ) - })?; - - tx.commit().await.map_err(|e| { - DomainError::internal_error("Dedup", format!("Failed to commit: {}", e)) - })?; - - let blob_path = self.blob_path(&hash); - - tracing::info!("DEDUP HIT: {} ({} bytes saved)", &hash[..12], size); - - return Ok(DedupResultDto::ExistingBlob { - hash, - size, - blob_path, - saved_bytes: size, - }); - } - - // Blob is new — write file to disk, then register in PG let blob_path = self.blob_path(&hash); - if let Some(parent) = blob_path.parent() { - fs::create_dir_all(parent).await.map_err(|e| { - DomainError::internal_error( - "Dedup", - format!("Failed to create blob directory: {}", e), - ) + // ── Phase 1: Write blob to disk (NO PG connection held) ───── + // + // Content-addressable: if two writers race for the same hash, + // both produce identical files. The rename is atomic on the + // same filesystem; if it fails because the other writer won, + // we just discard our temp file — the blob is already there. + if !blob_path.exists() { + if let Some(parent) = blob_path.parent() { + fs::create_dir_all(parent).await.map_err(|e| { + DomainError::internal_error( + "Dedup", + format!("Failed to create blob directory: {}", e), + ) + })?; + } + + let temp_path = self.temp_root.join(format!("{}.tmp", uuid::Uuid::new_v4())); + fs::write(&temp_path, content).await.map_err(|e| { + DomainError::internal_error("Dedup", format!("Failed to write temp blob: {}", e)) })?; + + if let Err(e) = fs::rename(&temp_path, &blob_path).await { + // Another writer already placed the blob — discard ours + let _ = fs::remove_file(&temp_path).await; + tracing::debug!("Blob file already placed by concurrent writer: {}", e); + } } - // Atomic write: temp file → rename - let temp_path = self.temp_root.join(format!("{}.tmp", uuid::Uuid::new_v4())); - fs::write(&temp_path, content).await.map_err(|e| { - DomainError::internal_error("Dedup", format!("Failed to write temp blob: {}", e)) - })?; - - if let Err(e) = fs::rename(&temp_path, &blob_path).await { - // Clean up temp file asynchronously (never block the Tokio worker) - let _ = fs::remove_file(&temp_path).await; - return Err(DomainError::internal_error( - "Dedup", - format!("Failed to move blob: {}", e), - )); - } - - // Register in PostgreSQL (ON CONFLICT handles rare race with another writer) - sqlx::query( + // ── Phase 2: Single atomic upsert (~2-4 ms, no explicit TX) ─ + // + // `INSERT … ON CONFLICT` is executed as a single implicit + // transaction by PostgreSQL. RETURNING ref_count tells us + // whether this was a new blob (ref_count = 1) or a dedup hit. + let ref_count: i32 = sqlx::query_scalar( "INSERT INTO storage.blobs (hash, size, ref_count, content_type) VALUES ($1, $2, 1, $3) - ON CONFLICT (hash) DO UPDATE SET ref_count = storage.blobs.ref_count + 1", + ON CONFLICT (hash) DO UPDATE SET ref_count = storage.blobs.ref_count + 1 + RETURNING ref_count", ) .bind(&hash) .bind(size as i64) .bind(&content_type) - .execute(&mut *tx) + .fetch_one(self.pool.as_ref()) .await .map_err(|e| { - DomainError::internal_error("Dedup", format!("Failed to register blob: {}", e)) + DomainError::internal_error("Dedup", format!("Failed to upsert blob: {}", e)) })?; - tx.commit().await.map_err(|e| { - DomainError::internal_error("Dedup", format!("Failed to commit: {}", e)) - })?; - - tracing::info!("NEW BLOB: {} ({} bytes)", &hash[..12], size); - - Ok(DedupResultDto::NewBlob { - hash, - size, - blob_path, - }) + if ref_count > 1 { + tracing::info!("DEDUP HIT: {} ({} bytes saved)", &hash[..12], size); + Ok(DedupResultDto::ExistingBlob { + hash, + size, + blob_path, + saved_bytes: size, + }) + } else { + tracing::info!("NEW BLOB: {} ({} bytes)", &hash[..12], size); + Ok(DedupResultDto::NewBlob { + hash, + size, + blob_path, + }) + } } /// Store content with deduplication (streaming from file). - /// Store content with deduplication (streaming from file). + /// + /// **Write-first strategy**: the source file is moved/copied to the + /// blob store *before* touching PostgreSQL, so the PG connection is + /// never held during disk I/O. /// /// If `pre_computed_hash` is `Some`, the file will NOT be re-read for /// SHA-256 — saving one full sequential read (the biggest I/O win). @@ -320,103 +300,78 @@ impl DedupService { .map_err(DomainError::from)?, }; - // Begin transaction - let mut tx = self.pool.begin().await.map_err(|e| { - DomainError::internal_error("Dedup", format!("Failed to begin transaction: {}", e)) - })?; + let blob_path = self.blob_path(&hash); - // SELECT FOR UPDATE - let existing = sqlx::query_scalar::<_, i32>( - "SELECT ref_count FROM storage.blobs WHERE hash = $1 FOR UPDATE", - ) - .bind(&hash) - .fetch_optional(&mut *tx) - .await - .map_err(|e| { - DomainError::internal_error("Dedup", format!("Failed to check blob: {}", e)) - })?; - - if existing.is_some() { - // Blob already exists — increment and delete source file - sqlx::query("UPDATE storage.blobs SET ref_count = ref_count + 1 WHERE hash = $1") - .bind(&hash) - .execute(&mut *tx) - .await - .map_err(|e| { + // ── Phase 1: Move/place blob on disk (NO PG connection held) ─ + // + // If the blob file already exists on disk, the source is simply + // deleted — the file content is identical by definition. + if blob_path.exists() { + // Blob already on disk — discard the source file + let _ = fs::remove_file(source_path).await; + } else { + if let Some(parent) = blob_path.parent() { + fs::create_dir_all(parent).await.map_err(|e| { DomainError::internal_error( "Dedup", - format!("Failed to increment ref_count: {}", e), + format!("Failed to create blob directory: {}", e), ) })?; + } - tx.commit().await.map_err(|e| { - DomainError::internal_error("Dedup", format!("Failed to commit: {}", e)) - })?; + // rename is atomic on the same filesystem. If source and blob + // dirs live on different filesystems (rare), this falls back to + // copy+delete which is slower but still correct. + if let Err(e) = fs::rename(source_path, &blob_path).await { + // Another writer may have placed the blob concurrently + if blob_path.exists() { + let _ = fs::remove_file(source_path).await; + tracing::debug!("Blob file placed by concurrent writer: {}", e); + } else { + return Err(DomainError::internal_error( + "Dedup", + format!("Failed to move file to blob store: {}", e), + )); + } + } + } - // Delete source file — we don't need it - let _ = fs::remove_file(source_path).await; - - let blob_path = self.blob_path(&hash); + // ── Phase 2: Single atomic upsert (~2-4 ms, no explicit TX) ─ + let ref_count: i32 = sqlx::query_scalar( + "INSERT INTO storage.blobs (hash, size, ref_count, content_type) + VALUES ($1, $2, 1, $3) + ON CONFLICT (hash) DO UPDATE SET ref_count = storage.blobs.ref_count + 1 + RETURNING ref_count", + ) + .bind(&hash) + .bind(file_size as i64) + .bind(&content_type) + .fetch_one(self.pool.as_ref()) + .await + .map_err(|e| { + DomainError::internal_error("Dedup", format!("Failed to upsert blob: {}", e)) + })?; + if ref_count > 1 { tracing::info!( "DEDUP HIT (file): {} ({} bytes saved)", &hash[..12], file_size ); - - return Ok(DedupResultDto::ExistingBlob { + Ok(DedupResultDto::ExistingBlob { hash, size: file_size, blob_path, saved_bytes: file_size, - }); + }) + } else { + tracing::info!("NEW BLOB (file): {} ({} bytes)", &hash[..12], file_size); + Ok(DedupResultDto::NewBlob { + hash, + size: file_size, + blob_path, + }) } - - // Move source file to blob store - let blob_path = self.blob_path(&hash); - - if let Some(parent) = blob_path.parent() { - fs::create_dir_all(parent).await.map_err(|e| { - DomainError::internal_error( - "Dedup", - format!("Failed to create blob directory: {}", e), - ) - })?; - } - - fs::rename(source_path, &blob_path).await.map_err(|e| { - DomainError::internal_error( - "Dedup", - format!("Failed to move file to blob store: {}", e), - ) - })?; - - // Register in PostgreSQL - sqlx::query( - "INSERT INTO storage.blobs (hash, size, ref_count, content_type) - VALUES ($1, $2, 1, $3) - ON CONFLICT (hash) DO UPDATE SET ref_count = storage.blobs.ref_count + 1", - ) - .bind(&hash) - .bind(file_size as i64) - .bind(&content_type) - .execute(&mut *tx) - .await - .map_err(|e| { - DomainError::internal_error("Dedup", format!("Failed to register blob: {}", e)) - })?; - - tx.commit().await.map_err(|e| { - DomainError::internal_error("Dedup", format!("Failed to commit: {}", e)) - })?; - - tracing::info!("NEW BLOB (file): {} ({} bytes)", &hash[..12], file_size); - - Ok(DedupResultDto::NewBlob { - hash, - size: file_size, - blob_path, - }) } // ── Reference counting ───────────────────────────────────────