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
2026-06-09 10:29:43 +02:00
parent 964d66e7f8
commit 2843b3351b
2 changed files with 122 additions and 48 deletions
@@ -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<String> = 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<PathBuf> = 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]
@@ -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)))?;