From 0cd544ab962af271935fe6b7075f59167fd307a5 Mon Sep 17 00:00:00 2001
From: Dionisio <dpozo@futurespace.es>
Date: Tue, 3 Mar 2026 11:18:40 +0100
Subject: [PATCH] perf: enable BLAKE3 multithreaded hashing (update_rayon) for
 files >10MB

---
 Cargo.lock                                    |  1 +
 Cargo.toml                                    |  2 +-
 .../services/chunked_upload_service.rs        | 10 +++-
 src/infrastructure/services/dedup_service.rs  | 47 ++++++++++++++-----
 4 files changed, 46 insertions(+), 14 deletions(-)

diff --git a/Cargo.lock b/Cargo.lock
index 9c068ca5..9a67059c 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -289,6 +289,7 @@ dependencies = [
  "cfg-if",
  "constant_time_eq",
  "cpufeatures",
+ "rayon-core",
 ]
 
 [[package]]
diff --git a/Cargo.toml b/Cargo.toml
index 9b228692..a9eb3bc3 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -41,7 +41,7 @@ http-range-header = "0.4"
 image = { version = "0.25", default-features = false, features = ["jpeg", "png", "gif", "webp"] }
 md5 = "0.8.0"
 sha2 = "0.10.9"
-blake3 = "1.8.3"
+blake3 = { version = "1.8.3", features = ["rayon"] }
 hex = "0.4.3"
 http-body-util = "0.1.3"
 percent-encoding = "2.3"
diff --git a/src/infrastructure/services/chunked_upload_service.rs b/src/infrastructure/services/chunked_upload_service.rs
index a999265b..d1a6e33d 100644
--- a/src/infrastructure/services/chunked_upload_service.rs
+++ b/src/infrastructure/services/chunked_upload_service.rs
@@ -661,6 +661,10 @@ impl ChunkedUploadService {
             let mut output = StdBufWriter::with_capacity(524_288, raw_output);
             let mut hasher = blake3::Hasher::new();
 
+            // For files >10 MB, use multithreaded BLAKE3 hashing (all cores)
+            const RAYON_THRESHOLD: u64 = 10 * 1024 * 1024;
+            let use_rayon = total_size > RAYON_THRESHOLD;
+
             // Single 512 KB read buffer reused across all chunks (avoids N allocations)
             let mut buf = vec![0u8; 524_288];
             for (index, chunk_path) in &chunks_meta {
@@ -673,7 +677,11 @@ impl ChunkedUploadService {
                     if n == 0 {
                         break;
                     }
-                    hasher.update(&buf[..n]);
+                    if use_rayon {
+                        hasher.update_rayon(&buf[..n]);
+                    } else {
+                        hasher.update(&buf[..n]);
+                    }
                     output.write_all(&buf[..n]).map_err(|e| {
                         format!("Failed to write chunk {index} to assembled file: {e}")
                     })?;
diff --git a/src/infrastructure/services/dedup_service.rs b/src/infrastructure/services/dedup_service.rs
index 583d5578..2b8ac446 100644
--- a/src/infrastructure/services/dedup_service.rs
+++ b/src/infrastructure/services/dedup_service.rs
@@ -52,6 +52,10 @@ use crate::domain::errors::{DomainError, ErrorKind};
 /// Block size for BLAKE3 file hashing (1MB — optimal syscall/throughput ratio).
 const HASH_BLOCK_SIZE: usize = 1024 * 1024;
 
+/// Files larger than this threshold use multithreaded BLAKE3 hashing via
+/// `update_rayon()`, which splits the work across all available cores.
+const RAYON_HASH_THRESHOLD: u64 = 10 * 1024 * 1024; // 10 MB
+
 /// Chunk size for streaming file reads (256 KB)
 const STREAM_CHUNK_SIZE: usize = 256 * 1024;
 
@@ -156,30 +160,49 @@ impl DedupService {
     // ── Hash helpers ─────────────────────────────────────────────
 
     /// Calculate BLAKE3 hash of content (~5× faster than SHA-256).
+    ///
+    /// For buffers larger than 10 MB the computation is parallelised across
+    /// all available cores via `update_rayon()`.
     pub fn hash_bytes(content: &[u8]) -> String {
-        blake3::hash(content).to_hex().to_string()
+        if content.len() as u64 > RAYON_HASH_THRESHOLD {
+            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
-    /// worker threads are never blocked by CPU-bound hashing.  Uses 1 MB
-    /// reads for optimal syscall-to-throughput ratio.
+    /// worker threads are never blocked by CPU-bound hashing.
+    ///
+    /// For files larger than 10 MB the hash is computed with `update_rayon()`,
+    /// which splits the work across all available cores.  Smaller files use
+    /// sequential 1 MB reads for optimal syscall-to-throughput ratio.
     pub async fn hash_file(path: &Path) -> std::io::Result<String> {
         let path = path.to_path_buf();
         tokio::task::spawn_blocking(move || {
-            use std::io::Read;
-
-            let mut file = std::fs::File::open(&path)?;
+            let file_size = std::fs::metadata(&path)?.len();
             let mut hasher = blake3::Hasher::new();
-            let mut buffer = vec![0u8; HASH_BLOCK_SIZE];
 
-            loop {
-                let n = file.read(&mut buffer)?;
-                if n == 0 {
-                    break;
+            if file_size > RAYON_HASH_THRESHOLD {
+                // Large file: read into memory and hash with all cores
+                let content = std::fs::read(&path)?;
+                hasher.update_rayon(&content);
+            } else {
+                // Small file: sequential streaming with 1 MB reads
+                use std::io::Read;
+                let mut file = std::fs::File::open(&path)?;
+                let mut buffer = vec![0u8; HASH_BLOCK_SIZE];
+                loop {
+                    let n = file.read(&mut buffer)?;
+                    if n == 0 {
+                        break;
+                    }
+                    hasher.update(&buffer[..n]);
                 }
-                hasher.update(&buffer[..n]);
             }
 
             Ok(hasher.finalize().to_hex().to_string())