From 2843b3351bbc0f215bfd05b2936a8ed27d161f24 Mon Sep 17 00:00:00 2001 From: Edouard Vanbelle Date: Tue, 9 Jun 2026 10:29:43 +0200 Subject: [PATCH] feat(blake digest): ensure digest is done on same passe as the write of chunks purpose: avoid current scheme: 1. write to disk 2. reopen file to read data and digest it now: digest is done while writing data to disk all other implementation than Nextcloud are corrrect --- .../nextcloud_chunked_upload_service.rs | 134 +++++++++++++----- src/interfaces/nextcloud/uploads_handler.rs | 36 +++-- 2 files changed, 122 insertions(+), 48 deletions(-) diff --git a/src/infrastructure/services/nextcloud_chunked_upload_service.rs b/src/infrastructure/services/nextcloud_chunked_upload_service.rs index 6f371143..34e6cb52 100644 --- a/src/infrastructure/services/nextcloud_chunked_upload_service.rs +++ b/src/infrastructure/services/nextcloud_chunked_upload_service.rs @@ -93,11 +93,24 @@ impl NextcloudChunkedUploadService { Ok(()) } - /// Assemble all chunks in numeric order into a temp file. + /// Assemble all chunks in numeric order into a temp file, computing + /// the BLAKE3 of the concatenated stream **during** the same read/ + /// write pass (hash-on-write). /// - /// Returns `(temp_path, total_size)`. The caller is responsible for - /// cleaning up the temp file after use. - pub async fn assemble(&self, user: &str, upload_id: &str) -> Result<(PathBuf, u64)> { + /// Returns `(temp_path, total_size, blake3_hex)`. The caller passes + /// the hash to the upload service as `pre_computed_hash` so the + /// downstream dedup layer never has to re-read the assembled file + /// to compute it — saving one full file-sized read pass per upload. + /// + /// The read/hash/write loop runs inside `spawn_blocking` because + /// BLAKE3 is CPU-bound and would otherwise starve the Tokio worker + /// running other connections; synchronous I/O is used inside the + /// blocking thread because the workload is sequential and the + /// async reactor overhead would only slow it down. For files larger + /// than ~10 MB BLAKE3's Rayon mode parallelises across cores — + /// mirrors what `ChunkedUploadService::complete_upload_inner` does + /// for the REST chunked path. + pub async fn assemble(&self, user: &str, upload_id: &str) -> Result<(PathBuf, u64, String)> { let session_dir = self.safe_session_dir(user, upload_id)?; let mut entries: Vec = Vec::new(); @@ -120,43 +133,74 @@ impl NextcloudChunkedUploadService { // Sort chunks numerically (Nextcloud sends them as "00001", "00002", ...). entries.sort(); - // Stream chunks to a temp file instead of buffering in memory. let temp_path = session_dir.join(".assembled"); - let mut out = fs::File::create(&temp_path) + let chunk_paths: Vec = entries.iter().map(|n| session_dir.join(n)).collect(); + let assembled_for_blocking = temp_path.clone(); + + // Read/hash/write loop runs synchronously on the blocking pool. + // BLAKE3 is computed in the same pass that copies bytes from chunk + // files into the assembled file — no second read after the fact. + let (total_size, hash) = + tokio::task::spawn_blocking(move || -> std::io::Result<(u64, String)> { + use std::io::{BufWriter as StdBufWriter, Read, Write}; + + let raw_output = std::fs::OpenOptions::new() + .create(true) + .write(true) + .truncate(true) + .open(&assembled_for_blocking)?; + + // 512 KB write buffer — 8× fewer syscalls than 64 KB. + let mut output = StdBufWriter::with_capacity(524_288, raw_output); + let mut hasher = blake3::Hasher::new(); + let mut buf = vec![0u8; 524_288]; + let mut total: u64 = 0; + + // Files >10 MB benefit from BLAKE3's multi-threaded mode. + // The threshold matches the REST chunked path's heuristic. + const RAYON_THRESHOLD_PER_FRAME: usize = 128 * 1024; + + for chunk_path in &chunk_paths { + let mut chunk_file = std::fs::File::open(chunk_path)?; + loop { + let n = chunk_file.read(&mut buf)?; + if n == 0 { + break; + } + if n >= RAYON_THRESHOLD_PER_FRAME { + hasher.update_rayon(&buf[..n]); + } else { + hasher.update(&buf[..n]); + } + output.write_all(&buf[..n])?; + total += n as u64; + } + } + + output.flush()?; + // ── Durability boundary ───────────────────────────────── + // sync_all is the actual fsync; without it, a power loss + // before the kernel writeback timer (~5 s) loses + // acknowledged data. Pull the inner File out of the + // BufWriter so we can sync the underlying handle — + // dropping the BufWriter wouldn't trigger fsync. macOS + // caveat: fsync there flushes to the disk controller + // only; true durability needs F_FULLFSYNC, not exposed + // by std. + let raw_output = output + .into_inner() + .map_err(|e| std::io::Error::other(format!("into_inner: {e}")))?; + raw_output.sync_all()?; + + Ok((total, hasher.finalize().to_hex().to_string())) + }) .await + .map_err(|e| { + DomainError::internal_error("ChunkedUpload", format!("assemble task: {e}")) + })? .map_err(|e| DomainError::internal_error("ChunkedUpload", e.to_string()))?; - let mut total_size: u64 = 0; - for chunk_name in &entries { - let mut chunk_file = fs::File::open(session_dir.join(chunk_name)) - .await - .map_err(|e| DomainError::internal_error("ChunkedUpload", e.to_string()))?; - let copied = tokio::io::copy(&mut chunk_file, &mut out) - .await - .map_err(|e| DomainError::internal_error("ChunkedUpload", e.to_string()))?; - total_size += copied; - } - - out.flush() - .await - .map_err(|e| DomainError::internal_error("ChunkedUpload", e.to_string()))?; - // ── Durability boundary ─────────────────────────────────────── - // `flush()` only drains tokio's userspace buffer; the bytes - // still sit in the kernel page cache until the OS writeback - // timer fires (~5 s on default Linux). A power loss in that - // window would lose acknowledged data — the metadata row - // referencing this blob would survive in PG (synchronous_commit - // is on), the blob file would not. `sync_all` issues `fsync(2)` - // and only returns once the bytes are on the storage medium - // (modulo macOS's well-known `F_FULLFSYNC` caveat — `fsync` there - // flushes to the disk controller, not the platter). Cost is - // ~1–5 ms on SSD, ~5–50 ms on HDD — invisible against the - // assembly I/O we already did. - out.sync_all() - .await - .map_err(|e| DomainError::internal_error("ChunkedUpload", e.to_string()))?; - - Ok((temp_path, total_size)) + Ok((temp_path, total_size, hash)) } /// Delete the upload session directory. @@ -299,10 +343,16 @@ mod tests { .await .unwrap(); - let (temp_path, size) = svc.assemble("alice", "upload-002").await.unwrap(); + let (temp_path, size, hash) = svc.assemble("alice", "upload-002").await.unwrap(); let assembled = fs::read(&temp_path).await.unwrap(); assert_eq!(assembled, b"Hello, World!"); assert_eq!(size, 13); + // BLAKE3("Hello, World!") — proves hash-on-write happens during + // the assemble pass, not via a re-read. + assert_eq!( + hash, + "288a86a79f20a3d6dccdca7713beaed178798296bdfa7913fa2a62d9727bf8f8" + ); } #[tokio::test] @@ -321,10 +371,16 @@ mod tests { .await .unwrap(); - let (temp_path, size) = svc.assemble("alice", "upload-003").await.unwrap(); + let (temp_path, size, hash) = svc.assemble("alice", "upload-003").await.unwrap(); let assembled = fs::read(&temp_path).await.unwrap(); assert_eq!(assembled, b"ABC"); assert_eq!(size, 3); + // BLAKE3("ABC") — confirms sort happened (chunks were stored in + // order 3,1,2 but the hash matches "ABC", not "CAB" or "BAC"). + assert_eq!( + hash, + "d1717274597cf0289694f75d96d444b992a096f1afd8e7bbfa6ebb1d360fedfc" + ); } #[tokio::test] diff --git a/src/interfaces/nextcloud/uploads_handler.rs b/src/interfaces/nextcloud/uploads_handler.rs index 56737ada..8bb6ca8f 100644 --- a/src/interfaces/nextcloud/uploads_handler.rs +++ b/src/interfaces/nextcloud/uploads_handler.rs @@ -239,8 +239,12 @@ async fn handle_assemble( let dest_subpath = extract_files_subpath(&destination, &user.username) .ok_or_else(|| AppError::bad_request("Invalid Destination URL"))?; - // Assemble chunks into a temp file (no full-file buffering in RAM). - let (temp_path, size) = nc + // Assemble chunks into a temp file with hash-on-write (BLAKE3 computed + // during the same read/write loop that copies chunks into the + // assembled file). The hash is passed downstream as `pre_computed_hash` + // so the dedup layer never re-reads the assembled file just to compute + // it — saves one full file-sized read pass per upload. + let (temp_path, size, blake3_hash) = nc .chunked_uploads .assemble(&user.username, upload_id) .await @@ -249,6 +253,7 @@ async fn handle_assemble( // Write assembled file to storage via the upload service. let upload_service = &state.applications.file_upload_service; let file_service = &state.applications.file_retrieval_service; + let folder_service = &state.applications.folder_service; let internal_path = format!( "My Folder - {}/{}", @@ -271,7 +276,7 @@ async fn handle_assemble( &temp_path, size, &content_type, - None, + Some(blake3_hash.clone()), oc_mtime, ) .await @@ -279,11 +284,12 @@ async fn handle_assemble( Some(dto.etag) } else { - // For new files we still need to read the temp file since create_file takes &[u8]. - let assembled = tokio::fs::read(&temp_path).await.map_err(|e| { - AppError::internal_error(format!("Failed to read assembled file: {}", e)) - })?; - + // New-file branch: resolve the parent folder by path and pass the + // assembled file's path directly to `upload_file_from_path` so the + // bytes never get read back into RAM. Previously this branch did + // `tokio::fs::read(&temp_path)` — an extra full file-sized read + // pass AND a peak-RAM allocation equal to the upload size, which + // defeated the streaming model on large NC uploads. let (parent_sub, filename) = match dest_subpath.rsplit_once('/') { Some((p, n)) => (p, n), None => ("", dest_subpath.as_str()), @@ -295,8 +301,20 @@ async fn handle_assemble( ); let parent_internal = parent_internal.trim_end_matches('/'); + use crate::application::ports::folder_ports::FolderUseCase; + let parent_folder = folder_service + .get_folder_by_path(parent_internal) + .await + .map_err(|e| AppError::internal_error(format!("Parent folder lookup failed: {}", e)))?; + let dto = upload_service - .create_file(parent_internal, filename, &assembled, &content_type) + .upload_file_from_path( + filename.to_string(), + Some(parent_folder.id), + content_type.to_string(), + &temp_path, + Some(blake3_hash), + ) .await .map_err(|e| AppError::internal_error(format!("Failed to create file: {}", e)))?;