From 81f33458e0c782bb49dbda5e766e8859f154e6e4 Mon Sep 17 00:00:00 2001 From: Diocrafts Date: Sat, 11 Apr 2026 16:17:47 +0200 Subject: [PATCH] perf: streaming hash-on-write dedup upload, remove dead code --- Cargo.lock | 2 +- src/application/ports/dedup_ports.rs | 10 - src/common/stubs.rs | 11 - src/infrastructure/services/dedup_service.rs | 132 +------- src/interfaces/api/handlers/dedup_handler.rs | 317 ++++++++++++++++--- 5 files changed, 277 insertions(+), 195 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index ffe5d72f..0ed3ad3d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2547,7 +2547,7 @@ dependencies = [ [[package]] name = "oxicloud" -version = "0.5.4" +version = "0.5.5" dependencies = [ "argon2", "async-compression", diff --git a/src/application/ports/dedup_ports.rs b/src/application/ports/dedup_ports.rs index c8db13bd..6d6f493e 100644 --- a/src/application/ports/dedup_ports.rs +++ b/src/application/ports/dedup_ports.rs @@ -92,16 +92,6 @@ pub struct DedupStatsDto { /// duplicate storage automatically. Multiple file references can /// point to the same physical blob. pub trait DedupPort: Send + Sync + 'static { - /// Store content with deduplication (from bytes). - /// - /// If content with the same hash already exists, a reference is added - /// instead of storing a duplicate. - async fn store_bytes( - &self, - content: &[u8], - content_type: Option, - ) -> Result; - /// Store content with deduplication (streaming from file). /// /// If `pre_computed_hash` is provided (e.g. hash-on-write from the handler), diff --git a/src/common/stubs.rs b/src/common/stubs.rs index e41a95fc..156563a5 100644 --- a/src/common/stubs.rs +++ b/src/common/stubs.rs @@ -716,17 +716,6 @@ use crate::application::ports::dedup_ports::{ pub struct StubDedupPort; impl DedupPort for StubDedupPort { - async fn store_bytes( - &self, - _content: &[u8], - _content_type: Option, - ) -> Result { - Err(DomainError::internal_error( - "DedupService", - "DedupService not initialized", - )) - } - async fn store_from_file( &self, _source_path: &Path, diff --git a/src/infrastructure/services/dedup_service.rs b/src/infrastructure/services/dedup_service.rs index 5229a029..587b985c 100644 --- a/src/infrastructure/services/dedup_service.rs +++ b/src/infrastructure/services/dedup_service.rs @@ -15,7 +15,7 @@ //! The dedup index lives in PostgreSQL (`storage.blobs`) — no in-memory //! HashMap, no JSON file, no WAL. //! -//! **Write-first strategy** (store_bytes / store_from_file): +//! **Write-first strategy** (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`. @@ -58,7 +58,7 @@ pub struct DedupService { /// Root directory for temporary files during upload temp_root: PathBuf, /// PostgreSQL connection pool (dedup index in `storage.blobs`) — primary, - /// used by request-path operations (store_bytes, store_from_file, etc.). + /// used by request-path operations (store_from_file, etc.). pool: Arc, /// Isolated maintenance pool for long-running operations /// (verify_integrity, garbage_collect) that must never starve the primary. @@ -168,20 +168,6 @@ impl DedupService { // ── Hash helpers ───────────────────────────────────────────── - /// Calculate BLAKE3 hash of in-memory content (~5× faster than SHA-256). - /// - /// For buffers larger than 128 KB the computation is parallelised across - /// all available cores via `update_rayon()`. - pub fn hash_bytes(content: &[u8]) -> String { - if content.len() > 128 * 1024 { - let mut hasher = blake3::Hasher::new(); - hasher.update_rayon(content); - hasher.finalize().to_hex().to_string() - } else { - blake3::hash(content).to_hex().to_string() - } - } - /// Calculate BLAKE3 hash of a file (~5× faster than SHA-256). /// /// Runs entirely on `spawn_blocking` with synchronous I/O so the Tokio @@ -205,108 +191,6 @@ impl DedupService { // ── Core store operations ──────────────────────────────────── - /// Maximum payload accepted by `store_bytes`. Anything larger - /// should use `store_from_file` (streaming — constant RAM). - const MAX_STORE_BYTES: usize = 10 * 1024 * 1024; // 10 MB - - /// Store content with deduplication (from bytes). - /// - /// **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. - pub async fn store_bytes( - &self, - content: &[u8], - content_type: Option, - ) -> Result { - if content.len() > Self::MAX_STORE_BYTES { - return Err(DomainError::internal_error( - "Dedup", - format!( - "store_bytes called with {} bytes (max {}). Use store_from_file for large content.", - content.len(), - Self::MAX_STORE_BYTES - ), - )); - } - - let size = content.len() as u64; - let hash = Self::hash_bytes(content); - let blob_path = self.blob_path(&hash); - - // ── 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 !fs::try_exists(&blob_path).await.unwrap_or(false) { - // Parent directory (xx/) guaranteed to exist — created by initialize() - 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 { - if e.raw_os_error() == Some(18) { - // EXDEV: cross-device link — fall back to copy+delete - fs::copy(&temp_path, &blob_path).await.map_err(|ce| { - DomainError::internal_error( - "Dedup", - format!("Failed to copy temp blob cross-device: {}", ce), - ) - })?; - let _ = fs::remove_file(&temp_path).await; - } else { - // 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); - } - } - } - - // ── 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 - RETURNING ref_count", - ) - .bind(&hash) - .bind(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: {} ({} 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). /// /// **Write-first strategy**: the source file is moved/copied to the @@ -494,7 +378,7 @@ impl DedupService { DomainError::internal_error("Dedup", format!("Failed to begin transaction: {}", e)) })?; - // Lock the row exclusively — prevents concurrent store_bytes from + // Lock the row exclusively — prevents concurrent store_from_file from // incrementing ref_count while we might be deleting let row = sqlx::query_as::<_, (i32, i64)>( "SELECT ref_count, size FROM storage.blobs WHERE hash = $1 FOR UPDATE", @@ -532,7 +416,7 @@ impl DedupService { })?; // Delete blob file AFTER committing PG — the row is gone, so no - // concurrent store_bytes can resurrect a reference to this hash. + // concurrent store_from_file can resurrect a reference to this hash. let blob_path = self.blob_path(hash); if let Err(e) = fs::remove_file(&blob_path).await { tracing::warn!("Failed to delete blob file {}: {}", hash, e); @@ -854,14 +738,6 @@ impl DedupService { // ─── Port implementation ───────────────────────────────────────────────────── impl DedupPort for DedupService { - async fn store_bytes( - &self, - content: &[u8], - content_type: Option, - ) -> Result { - self.store_bytes(content, content_type).await - } - async fn store_from_file( &self, source_path: &Path, diff --git a/src/interfaces/api/handlers/dedup_handler.rs b/src/interfaces/api/handlers/dedup_handler.rs index d7def5b3..ba0a84f6 100644 --- a/src/interfaces/api/handlers/dedup_handler.rs +++ b/src/interfaces/api/handlers/dedup_handler.rs @@ -4,8 +4,8 @@ use axum::{ http::{Response, StatusCode, header}, response::IntoResponse, }; -use bytes::Bytes; use serde::Serialize; +use tokio::io::AsyncWriteExt; use crate::application::ports::dedup_ports::DedupResultDto; use crate::common::di::AppState; @@ -134,15 +134,14 @@ impl DedupHandler { } } - /// Upload content with automatic deduplication + /// Upload content with automatic deduplication (streaming). /// - /// This endpoint calculates the SHA-256 hash of the uploaded content - /// and either creates a new blob or increments the reference count - /// of an existing blob. + /// Spools the upload to a temp file while computing the BLAKE3 hash + /// incrementally (hash-on-write). Memory usage is constant (~512 KB) + /// regardless of file size. Then delegates to `store_from_file` with + /// the pre-computed hash so the file is never re-read for hashing. /// /// POST /api/dedup/upload - /// - /// Returns information about whether the content was new or deduplicated. pub async fn upload_with_dedup( State(state): State, _auth_user: AuthUser, @@ -160,38 +159,65 @@ impl DedupHandler { .unwrap_or("application/octet-stream") .to_string(); - // Collect all chunks — explicit match to detect client disconnection - let mut chunks: Vec = Vec::new(); - let mut total_size: usize = 0; + // ── Spool to temp file + BLAKE3 hash-on-write ──────── + let temp_dir = state.core.path_service.get_root_path().join(".dedup_temp"); + let temp_path = temp_dir.join(format!("dedup-{}", uuid::Uuid::new_v4())); + + let mut total_size: u64 = 0; + let mut hasher = blake3::Hasher::new(); let mut field = field; - loop { - match field.chunk().await { - Ok(Some(chunk)) => { - total_size += chunk.len(); - chunks.push(chunk); - } - Ok(None) => break, - Err(e) => { - tracing::warn!( - "Connection lost during dedup upload (received {} bytes): {}", - total_size, - e - ); - return Response::builder() - .status(StatusCode::BAD_REQUEST) - .header(header::CONTENT_TYPE, "application/json") - .body(Body::from(format!( - r#"{{"error": "Connection lost during upload: {}"}}"#, - e - ))) - .unwrap() - .into_response(); + let spool_result: Result<(), String> = async { + let file = tokio::fs::File::create(&temp_path) + .await + .map_err(|e| format!("Failed to create temp file: {}", e))?; + + // 512 KB buffer — reduces write syscalls + let mut writer = tokio::io::BufWriter::with_capacity(524_288, file); + + loop { + match field.chunk().await { + Ok(Some(chunk)) => { + total_size += chunk.len() as u64; + hasher.update(&chunk); + writer.write_all(&chunk).await.map_err(|e| { + format!("Failed to write chunk: {}", e) + })?; + } + Ok(None) => break, + Err(e) => { + return Err(format!( + "Connection lost during upload (received {} bytes): {}", + total_size, e + )); + } } } + + writer + .flush() + .await + .map_err(|e| format!("Failed to flush temp file: {}", e))?; + Ok(()) + } + .await; + + if let Err(msg) = spool_result { + let _ = tokio::fs::remove_file(&temp_path).await; + tracing::warn!("Dedup upload spool failed: {}", msg); + return Response::builder() + .status(StatusCode::BAD_REQUEST) + .header(header::CONTENT_TYPE, "application/json") + .body(Body::from(format!( + r#"{{"error": "{}"}}"#, + msg + ))) + .unwrap() + .into_response(); } - if chunks.is_empty() { + if total_size == 0 { + let _ = tokio::fs::remove_file(&temp_path).await; return Response::builder() .status(StatusCode::BAD_REQUEST) .header(header::CONTENT_TYPE, "application/json") @@ -200,19 +226,13 @@ impl DedupHandler { .into_response(); } - // Combine chunks - let data: Vec = if chunks.len() == 1 { - chunks.into_iter().next().unwrap().to_vec() - } else { - let mut combined = Vec::with_capacity(total_size); - for chunk in chunks { - combined.extend_from_slice(&chunk); - } - combined - }; + let hash = hasher.finalize().to_hex().to_string(); - // Store with deduplication - match dedup.store_bytes(&data, Some(content_type)).await { + // ── Store with deduplication (pre-computed hash) ────── + match dedup + .store_from_file(&temp_path, Some(content_type), Some(hash)) + .await + { Ok(result) => { let (is_new, bytes_saved) = match &result { DedupResultDto::NewBlob { .. } => (true, 0), @@ -250,6 +270,7 @@ impl DedupHandler { .into_response(); } Err(e) => { + let _ = tokio::fs::remove_file(&temp_path).await; tracing::error!("Dedup upload failed: {}", e); return Response::builder() .status(StatusCode::INTERNAL_SERVER_ERROR) @@ -465,3 +486,209 @@ impl DedupHandler { .into_response() } } + +#[cfg(test)] +mod tests { + use super::*; + use tokio::io::AsyncWriteExt; + + /// Verify that the hash-on-write pattern produces the same BLAKE3 hash + /// as hashing the entire content at once. + #[tokio::test] + async fn hash_on_write_matches_full_hash() { + let content = b"Hello, OxiCloud dedup streaming upload!"; + + // 1. Full-content hash (reference) + let full_hash = blake3::hash(content).to_hex().to_string(); + + // 2. Incremental hash-on-write (what the handler does) + let mut hasher = blake3::Hasher::new(); + // Simulate multiple chunks + hasher.update(&content[..10]); + hasher.update(&content[10..25]); + hasher.update(&content[25..]); + let incremental_hash = hasher.finalize().to_hex().to_string(); + + assert_eq!(full_hash, incremental_hash); + } + + /// Verify BLAKE3 incremental hashing produces a valid 64-char hex hash. + #[tokio::test] + async fn incremental_hash_format_is_valid() { + let content = vec![0xABu8; 1024 * 1024]; // 1 MB of data + + let mut hasher = blake3::Hasher::new(); + // Feed in 64 KB chunks like real uploads + for chunk in content.chunks(65_536) { + hasher.update(chunk); + } + let hash = hasher.finalize().to_hex().to_string(); + + assert_eq!(hash.len(), 64, "BLAKE3 hash should be 64 hex characters"); + assert!( + hash.chars().all(|c| c.is_ascii_hexdigit()), + "Hash should only contain hex characters" + ); + } + + /// Verify spool-to-disk + BLAKE3 hash-on-write writes correct content + /// and produces the correct hash. + #[tokio::test] + async fn spool_to_temp_file_preserves_content_and_hash() { + let temp_dir = tempfile::tempdir().unwrap(); + let temp_path = temp_dir.path().join("test-upload.tmp"); + + let content = b"The quick brown fox jumps over the lazy dog"; + let expected_hash = blake3::hash(content).to_hex().to_string(); + + // Simulate the handler's hash-on-write spool loop + let mut hasher = blake3::Hasher::new(); + let mut total_size: u64 = 0; + { + let file = tokio::fs::File::create(&temp_path).await.unwrap(); + let mut writer = tokio::io::BufWriter::with_capacity(524_288, file); + + // Simulate 3 incoming chunks + let chunks: &[&[u8]] = &[&content[..10], &content[10..30], &content[30..]]; + for chunk in chunks { + total_size += chunk.len() as u64; + hasher.update(chunk); + writer.write_all(chunk).await.unwrap(); + } + writer.flush().await.unwrap(); + } + + let hash = hasher.finalize().to_hex().to_string(); + + // Verify hash matches + assert_eq!(hash, expected_hash); + + // Verify total size + assert_eq!(total_size, content.len() as u64); + + // Verify file content on disk is identical + let disk_content = tokio::fs::read(&temp_path).await.unwrap(); + assert_eq!(disk_content, content); + } + + /// Verify that an empty upload produces total_size == 0. + #[tokio::test] + async fn empty_upload_detected_before_store() { + let hasher = blake3::Hasher::new(); + let total_size: u64 = 0; + + // No chunks fed — simulates empty file + let _hash = hasher.finalize().to_hex().to_string(); + + // The handler checks total_size == 0 and returns 400 + assert_eq!(total_size, 0); + } + + /// Verify large payload streaming produces consistent hash + /// without buffering all content in memory. + #[tokio::test] + async fn large_payload_streaming_hash_consistency() { + let temp_dir = tempfile::tempdir().unwrap(); + let temp_path = temp_dir.path().join("large-upload.tmp"); + + // 5 MB of patterned data + let chunk_size = 65_536usize; // 64 KB chunks + let total_chunks = 80; // 80 × 64 KB = 5 MB + let mut reference_data = Vec::with_capacity(chunk_size * total_chunks); + + let mut hasher = blake3::Hasher::new(); + let mut total_size: u64 = 0; + { + let file = tokio::fs::File::create(&temp_path).await.unwrap(); + let mut writer = tokio::io::BufWriter::with_capacity(524_288, file); + + for i in 0..total_chunks { + // Patterned data: each chunk filled with its index byte + let chunk = vec![(i % 256) as u8; chunk_size]; + reference_data.extend_from_slice(&chunk); + total_size += chunk.len() as u64; + hasher.update(&chunk); + writer.write_all(&chunk).await.unwrap(); + } + writer.flush().await.unwrap(); + } + + let streaming_hash = hasher.finalize().to_hex().to_string(); + let reference_hash = blake3::hash(&reference_data).to_hex().to_string(); + + // Hashes match + assert_eq!(streaming_hash, reference_hash); + + // File on disk matches + let file_size = tokio::fs::metadata(&temp_path).await.unwrap().len(); + assert_eq!(file_size, total_size); + assert_eq!(total_size, (chunk_size * total_chunks) as u64); + + // Verify file content matches (read back) + let disk_data = tokio::fs::read(&temp_path).await.unwrap(); + assert_eq!(disk_data, reference_data); + } + + /// Verify temp file is cleaned up when spool fails partway through. + #[tokio::test] + async fn temp_file_cleanup_on_partial_write() { + let temp_dir = tempfile::tempdir().unwrap(); + let temp_path = temp_dir.path().join("partial-upload.tmp"); + + // Create the file and write some data + { + let file = tokio::fs::File::create(&temp_path).await.unwrap(); + let mut writer = tokio::io::BufWriter::with_capacity(524_288, file); + writer.write_all(b"partial data").await.unwrap(); + writer.flush().await.unwrap(); + } + + // File exists before cleanup + assert!(tokio::fs::try_exists(&temp_path).await.unwrap()); + + // Simulate the handler's error cleanup path + let _ = tokio::fs::remove_file(&temp_path).await; + + // File is gone after cleanup + assert!(!tokio::fs::try_exists(&temp_path).await.unwrap_or(true)); + } + + /// Verify the DedupUploadResponse serializes correctly for new blobs. + #[test] + fn dedup_upload_response_serialization_new_blob() { + let response = DedupUploadResponse { + is_new: true, + hash: "a".repeat(64), + size: 1024, + bytes_saved: 0, + ref_count: 1, + }; + + let json = serde_json::to_string(&response).unwrap(); + let parsed: serde_json::Value = serde_json::from_str(&json).unwrap(); + + assert_eq!(parsed["is_new"], true); + assert_eq!(parsed["size"], 1024); + assert_eq!(parsed["bytes_saved"], 0); + assert_eq!(parsed["ref_count"], 1); + } + + /// Verify the DedupUploadResponse serializes correctly for dedup hits. + #[test] + fn dedup_upload_response_serialization_dedup_hit() { + let response = DedupUploadResponse { + is_new: false, + hash: "b".repeat(64), + size: 2048, + bytes_saved: 2048, + ref_count: 3, + }; + + let json = serde_json::to_string(&response).unwrap(); + let parsed: serde_json::Value = serde_json::from_str(&json).unwrap(); + + assert_eq!(parsed["is_new"], false); + assert_eq!(parsed["bytes_saved"], 2048); + assert_eq!(parsed["ref_count"], 3); + } +}