From 85908311dc125dbc06c98d71b1ebdc17030fd6e5 Mon Sep 17 00:00:00 2001 From: Diocrafts Date: Mon, 23 Feb 2026 00:51:46 +0100 Subject: [PATCH] =?UTF-8?q?perf:=20findings=206.1,=206.2,=202.6=20?= =?UTF-8?q?=E2=80=94=20async=20Argon2,=20moka=20cache,=20full=20streaming?= =?UTF-8?q?=20migration?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 6.1: PasswordHasherPort now async_trait with spawn_blocking for Argon2 - 6.2: OIDC pending maps migrated from std::sync::Mutex to moka::sync::Cache with TTL - 2.6: All file download paths migrated to 64KB streaming (get_file_stream / read_blob_stream) - WOPI, dedup, batch ZIP, file_retrieval_service consumers migrated - WebDAV COPY uses zero-copy dedup (copy_file) - Removed dead code: get_file_content, get_file_mmap, read_blob, read_blob_bytes from traits, impls, stubs, and mocks (18 files touched) --- src/application/dtos/user_dto.rs | 4 +- src/application/ports/auth_ports.rs | 8 +- src/application/ports/dedup_ports.rs | 8 - src/application/ports/file_ports.rs | 3 - src/application/ports/storage_ports.rs | 6 - .../services/auth_application_service.rs | 144 +++++++----------- src/application/services/batch_operations.rs | 38 +++-- .../services/file_retrieval_service.rs | 40 ++--- src/application/services/share_service.rs | 15 +- .../services/trash_service_test.rs | 8 - src/common/stubs.rs | 26 ---- src/domain/repositories/file_repository.rs | 6 - .../pg/file_blob_read_repository.rs | 12 -- src/infrastructure/services/dedup_service.rs | 69 ++++----- .../services/password_hasher.rs | 94 ++++++++---- src/interfaces/api/handlers/dedup_handler.rs | 21 ++- src/interfaces/api/handlers/webdav_handler.rs | 28 ++-- src/interfaces/api/handlers/wopi_handler.rs | 10 +- 18 files changed, 235 insertions(+), 305 deletions(-) diff --git a/src/application/dtos/user_dto.rs b/src/application/dtos/user_dto.rs index 7a84d95d..e1916fa6 100644 --- a/src/application/dtos/user_dto.rs +++ b/src/application/dtos/user_dto.rs @@ -2,7 +2,7 @@ use crate::domain::entities::user::User; use chrono::{DateTime, Utc}; use serde::{Deserialize, Serialize}; -#[derive(Debug, Serialize, Deserialize)] +#[derive(Debug, Clone, Serialize, Deserialize)] pub struct UserDto { pub id: String, pub username: String, @@ -49,7 +49,7 @@ pub struct RegisterDto { pub role: Option, } -#[derive(Debug, Serialize, Deserialize)] +#[derive(Debug, Clone, Serialize, Deserialize)] pub struct AuthResponseDto { pub user: UserDto, pub access_token: String, diff --git a/src/application/ports/auth_ports.rs b/src/application/ports/auth_ports.rs index cc175e08..e419bd3a 100644 --- a/src/application/ports/auth_ports.rs +++ b/src/application/ports/auth_ports.rs @@ -11,12 +11,16 @@ use async_trait::async_trait; /// /// This trait abstracts cryptographic password operations, allowing the domain /// layer to remain independent of specific hashing implementations (argon2, bcrypt, etc.) +/// +/// Methods are async because implementations (e.g. Argon2) are CPU-intensive +/// and must run on a blocking thread pool to avoid starving Tokio workers. +#[async_trait] pub trait PasswordHasherPort: Send + Sync + 'static { /// Hash a plain text password - fn hash_password(&self, password: &str) -> Result; + async fn hash_password(&self, password: &str) -> Result; /// Verify a plain text password against a hash - fn verify_password(&self, password: &str, hash: &str) -> Result; + async fn verify_password(&self, password: &str, hash: &str) -> Result; } /// Claims contained in a JWT token diff --git a/src/application/ports/dedup_ports.rs b/src/application/ports/dedup_ports.rs index 5c70d120..cc86999c 100644 --- a/src/application/ports/dedup_ports.rs +++ b/src/application/ports/dedup_ports.rs @@ -122,15 +122,7 @@ pub trait DedupPort: Send + Sync + 'static { /// Get metadata for a blob. async fn get_blob_metadata(&self, hash: &str) -> Option; - /// Read blob content as raw bytes. - async fn read_blob(&self, hash: &str) -> Result, DomainError>; - - /// Read blob content as `Bytes`. - async fn read_blob_bytes(&self, hash: &str) -> Result; - /// Stream blob content in chunks (64 KB default) — constant memory usage. - /// - /// Unlike `read_blob()`, this never loads the entire file into RAM. async fn read_blob_stream( &self, hash: &str, diff --git a/src/application/ports/file_ports.rs b/src/application/ports/file_ports.rs index e452123f..ec63d76f 100644 --- a/src/application/ports/file_ports.rs +++ b/src/application/ports/file_ports.rs @@ -120,9 +120,6 @@ pub trait FileRetrievalUseCase: Send + Sync + 'static { /// Lists files in a folder async fn list_files(&self, folder_id: Option<&str>) -> Result, DomainError>; - /// Gets file content as bytes (for small files) - async fn get_file_content(&self, id: &str) -> Result, DomainError>; - /// Gets file content as a stream (for large files) async fn get_file_stream( &self, diff --git a/src/application/ports/storage_ports.rs b/src/application/ports/storage_ports.rs index 8c16b46f..d9a59d0c 100644 --- a/src/application/ports/storage_ports.rs +++ b/src/application/ports/storage_ports.rs @@ -32,9 +32,6 @@ pub trait FileReadPort: Send + Sync + 'static { /// Lists files in a folder. async fn list_files(&self, folder_id: Option<&str>) -> Result, DomainError>; - /// Gets the full content as bytes (small/medium files only). - async fn get_file_content(&self, id: &str) -> Result, DomainError>; - /// Gets content as a stream (ideal for large files). async fn get_file_stream( &self, @@ -49,9 +46,6 @@ pub trait FileReadPort: Send + Sync + 'static { end: Option, ) -> Result> + Send>, DomainError>; - /// Memory-map of a file for zero-copy access (10–100 MB). - async fn get_file_mmap(&self, id: &str) -> Result; - /// Gets the logical storage path of a file. async fn get_file_path(&self, id: &str) -> Result; diff --git a/src/application/services/auth_application_service.rs b/src/application/services/auth_application_service.rs index abd5f2da..d4214462 100644 --- a/src/application/services/auth_application_service.rs +++ b/src/application/services/auth_application_service.rs @@ -10,29 +10,23 @@ use crate::common::config::OidcConfig; use crate::common::errors::{DomainError, ErrorKind}; use crate::domain::entities::session::Session; use crate::domain::entities::user::{User, UserRole}; -use std::collections::HashMap; +use moka::sync::Cache; use std::path::PathBuf; use std::sync::Arc; -use std::sync::Mutex; use std::sync::RwLock; -use std::time::Instant; - -/// Maximum age for pending OIDC flows (10 minutes) -const OIDC_FLOW_TTL_SECS: u64 = 600; -/// Maximum age for pending one-time token codes (60 seconds) -const OIDC_TOKEN_TTL_SECS: u64 = 60; +use std::time::Duration; /// Tracks a pending OIDC authorization flow (CSRF + PKCE + nonce) +#[derive(Clone)] struct PendingOidcFlow { - created_at: Instant, pkce_verifier: String, nonce: String, } /// Tracks a pending one-time token exchange after successful OIDC callback +#[derive(Clone)] struct PendingOidcToken { auth_response: AuthResponseDto, - created_at: Instant, } /// Interior state for OIDC — protected by RwLock for hot-reload. @@ -54,10 +48,12 @@ pub struct AuthApplicationService { /// Path to the storage directory, used for disk-space–aware quota calculation storage_path: PathBuf, oidc: RwLock, - /// Pending OIDC authorization flows keyed by state token (CSRF + PKCE + nonce) - pending_oidc_flows: Mutex>, - /// Pending one-time token codes for secure token delivery after OIDC callback - pending_oidc_tokens: Mutex>, + /// Pending OIDC authorization flows keyed by state token (CSRF + PKCE + nonce). + /// Auto-expires after 10 minutes via moka TTL; max 10 000 entries for DoS protection. + pending_oidc_flows: Cache, + /// Pending one-time token codes for secure token delivery after OIDC callback. + /// Auto-expires after 60 seconds via moka TTL; max 10 000 entries for DoS protection. + pending_oidc_tokens: Cache, } impl AuthApplicationService { @@ -79,8 +75,14 @@ impl AuthApplicationService { service: None, config: None, }), - pending_oidc_flows: Mutex::new(HashMap::new()), - pending_oidc_tokens: Mutex::new(HashMap::new()), + pending_oidc_flows: Cache::builder() + .max_capacity(10_000) + .time_to_live(Duration::from_secs(600)) + .build(), + pending_oidc_tokens: Cache::builder() + .max_capacity(10_000) + .time_to_live(Duration::from_secs(60)) + .build(), } } @@ -300,7 +302,7 @@ impl AuthApplicationService { } // Hash the password using the infrastructure service - let password_hash = self.password_hasher.hash_password(&dto.password)?; + let password_hash = self.password_hasher.hash_password(&dto.password).await?; // Create user with the pre-generated hash let user = User::new(dto.username.clone(), dto.email, password_hash, role, quota).map_err( @@ -346,7 +348,8 @@ impl AuthApplicationService { // Verify password using the injected hasher let is_valid = self .password_hasher - .verify_password(&dto.password, user.password_hash())?; + .verify_password(&dto.password, user.password_hash()) + .await?; if !is_valid { return Err(DomainError::new( @@ -502,7 +505,8 @@ impl AuthApplicationService { // Verify current password using the injected hasher let is_valid = self .password_hasher - .verify_password(&dto.current_password, user.password_hash())?; + .verify_password(&dto.current_password, user.password_hash()) + .await?; if !is_valid { return Err(DomainError::new( @@ -522,7 +526,7 @@ impl AuthApplicationService { } // Hash new password and update user - let new_hash = self.password_hasher.hash_password(&dto.new_password)?; + let new_hash = self.password_hasher.hash_password(&dto.new_password).await?; user.update_password_hash(new_hash); // Save updated user @@ -641,13 +645,7 @@ impl AuthApplicationService { let password_hash = self .password_hasher .hash_password(&dto.password) - .map_err(|e| { - DomainError::new( - ErrorKind::InternalError, - "User", - format!("Error hashing password: {}", e), - ) - })?; + .await?; // Create the new admin user let user = User::new( @@ -753,7 +751,7 @@ impl AuthApplicationService { }); // Hash password - let password_hash = self.password_hasher.hash_password(&dto.password)?; + let password_hash = self.password_hasher.hash_password(&dto.password).await?; // Create domain entity let user = @@ -806,7 +804,7 @@ impl AuthApplicationService { "Password must be at least 8 characters long".to_string(), )); } - let hash = self.password_hasher.hash_password(new_password)?; + let hash = self.password_hasher.hash_password(new_password).await?; self.user_storage.change_password(user_id, &hash).await } @@ -917,22 +915,14 @@ impl AuthApplicationService { base64_url_encode(&hash) }; - // Store pending flow - { - let mut flows = self.pending_oidc_flows.lock().unwrap(); - // Cleanup expired entries - let now = Instant::now(); - flows.retain(|_, f| now.duration_since(f.created_at).as_secs() < OIDC_FLOW_TTL_SECS); - - flows.insert( - state_token.clone(), - PendingOidcFlow { - created_at: now, - pkce_verifier, - nonce: nonce.clone(), - }, - ); - } + // Store pending flow (auto-expires after 10 min via moka TTL) + self.pending_oidc_flows.insert( + state_token.clone(), + PendingOidcFlow { + pkce_verifier, + nonce: nonce.clone(), + }, + ); // Build authorization URL with state, nonce, and PKCE challenge let authorize_url = oidc @@ -952,28 +942,15 @@ impl AuthApplicationService { /// issue internal tokens, and return a one-time exchange code. pub async fn oidc_callback(&self, code: &str, state: &str) -> Result { // 0. Validate CSRF state and retrieve PKCE verifier + nonce - let (pkce_verifier, nonce) = { - let mut flows = self.pending_oidc_flows.lock().unwrap(); - let flow = flows.remove(state).ok_or_else(|| { - tracing::warn!("OIDC callback with invalid/expired state token"); - DomainError::new( - ErrorKind::AccessDenied, "OIDC", - "Invalid or expired OIDC state — possible CSRF attack. Please try logging in again.", - ) - })?; - - // Check TTL - if Instant::now().duration_since(flow.created_at).as_secs() >= OIDC_FLOW_TTL_SECS { - tracing::warn!("OIDC callback with expired state token"); - return Err(DomainError::new( - ErrorKind::AccessDenied, - "OIDC", - "OIDC authorization flow expired. Please try logging in again.", - )); - } - - (flow.pkce_verifier, flow.nonce) - }; + // (entry is auto-expired by moka TTL — remove returns None if expired) + let flow = self.pending_oidc_flows.remove(state).ok_or_else(|| { + tracing::warn!("OIDC callback with invalid/expired state token"); + DomainError::new( + ErrorKind::AccessDenied, "OIDC", + "Invalid or expired OIDC state — possible CSRF attack. Please try logging in again.", + ) + })?; + let (pkce_verifier, nonce) = (flow.pkce_verifier, flow.nonce); // Clone the Arc and config out of the RwLock so we don't hold the lock across await points let (oidc, oidc_config) = { @@ -1178,20 +1155,11 @@ impl AuthApplicationService { OsRng.fill_bytes(&mut code_bytes); let exchange_code = hex::encode(code_bytes); - { - let mut tokens = self.pending_oidc_tokens.lock().unwrap(); - // Cleanup expired entries - let now = Instant::now(); - tokens.retain(|_, t| now.duration_since(t.created_at).as_secs() < OIDC_TOKEN_TTL_SECS); - - tokens.insert( - exchange_code.clone(), - PendingOidcToken { - auth_response, - created_at: now, - }, - ); - } + // Store auth response (auto-expires after 60 s via moka TTL) + self.pending_oidc_tokens.insert( + exchange_code.clone(), + PendingOidcToken { auth_response }, + ); tracing::info!("OIDC login successful, one-time exchange code generated"); @@ -1199,10 +1167,9 @@ impl AuthApplicationService { } /// Exchange a one-time code for the authentication tokens. - /// The code is single-use and expires after 60 seconds. + /// The code is single-use and expires after 60 seconds (moka TTL). pub fn exchange_oidc_token(&self, one_time_code: &str) -> Result { - let mut tokens = self.pending_oidc_tokens.lock().unwrap(); - let pending = tokens.remove(one_time_code).ok_or_else(|| { + let pending = self.pending_oidc_tokens.remove(one_time_code).ok_or_else(|| { DomainError::new( ErrorKind::AccessDenied, "OIDC", @@ -1210,15 +1177,6 @@ impl AuthApplicationService { ) })?; - // Check TTL - if Instant::now().duration_since(pending.created_at).as_secs() >= OIDC_TOKEN_TTL_SECS { - return Err(DomainError::new( - ErrorKind::AccessDenied, - "OIDC", - "Exchange code expired. Please try logging in again.", - )); - } - Ok(pending.auth_response) } diff --git a/src/application/services/batch_operations.rs b/src/application/services/batch_operations.rs index 5631e284..ff894297 100644 --- a/src/application/services/batch_operations.rs +++ b/src/application/services/batch_operations.rs @@ -1,4 +1,4 @@ -use futures::{Future, future::join_all}; +use futures::{Future, StreamExt, future::join_all}; use std::sync::Arc; use thiserror::Error; use tokio::sync::Semaphore; @@ -702,18 +702,30 @@ impl BatchOperationService { // Add individual files at the root of the ZIP for file_id in &file_ids { match self.file_retrieval.get_file(file_id).await { - Ok(file_dto) => match self.file_retrieval.get_file_content(file_id).await { - Ok(content) => { + Ok(file_dto) => match self.file_retrieval.get_file_stream(file_id).await { + Ok(stream) => { + let mut stream = std::pin::Pin::from(stream); if let Err(e) = zip.start_file(&file_dto.name, options) { info!("Could not start zip entry for {}: {}", file_dto.name, e); continue; } - if let Err(e) = zip.write_all(&content) { - info!("Could not write zip entry for {}: {}", file_dto.name, e); + while let Some(chunk) = stream.next().await { + match chunk { + Ok(bytes) => { + if let Err(e) = zip.write_all(&bytes) { + info!("Could not write zip chunk for {}: {}", file_dto.name, e); + break; + } + } + Err(e) => { + info!("Stream error for {}: {}", file_dto.name, e); + break; + } + } } } Err(e) => { - info!("Could not read file content {}: {}", file_id, e); + info!("Could not stream file content {}: {}", file_id, e); } }, Err(e) => { @@ -805,13 +817,21 @@ impl BatchOperationService { let dir_path = format!("{}/", current.path); let _ = zip.add_directory(&dir_path, *options); - // Add files + // Add files via streaming (constant ~64 KB memory per file) if let Ok(files) = self.file_retrieval.list_files(Some(¤t.id)).await { for file in files { let file_path = format!("{}{}", dir_path, file.name); - if let Ok(content) = self.file_retrieval.get_file_content(&file.id).await { + if let Ok(stream) = self.file_retrieval.get_file_stream(&file.id).await { + let mut stream = std::pin::Pin::from(stream); if zip.start_file(&file_path, *options).is_ok() { - let _ = zip.write_all(&content); + while let Some(chunk) = stream.next().await { + match chunk { + Ok(bytes) => { + let _ = zip.write_all(&bytes); + } + Err(_) => break, + } + } } } } diff --git a/src/application/services/file_retrieval_service.rs b/src/application/services/file_retrieval_service.rs index d14902dc..45e1f1b8 100644 --- a/src/application/services/file_retrieval_service.rs +++ b/src/application/services/file_retrieval_service.rs @@ -1,6 +1,6 @@ use async_trait::async_trait; -use bytes::Bytes; -use futures::Stream; +use bytes::{Bytes, BytesMut}; +use futures::{Stream, StreamExt}; use std::sync::Arc; use crate::application::dtos::file_dto::FileDto; @@ -9,7 +9,7 @@ use crate::application::ports::file_ports::{FileRetrievalUseCase, OptimizedFileC use crate::application::ports::storage_ports::FileReadPort; use crate::application::ports::transcode_ports::{ImageTranscodePort, OutputFormat}; use crate::common::errors::DomainError; -use tracing::{debug, info, warn}; +use tracing::{debug, info}; /// Threshold below which files are served from RAM cache (10 MB). const CACHE_THRESHOLD: u64 = 10 * 1024 * 1024; @@ -183,10 +183,17 @@ impl FileRetrievalService { )); } - // Cache miss – load from disk + // Cache miss – load from disk via streaming (constant 64 KB memory) debug!("💾 TIER 1 Cache MISS: {} – loading from disk", file_name); - let content = self.file_read.get_file_content(id).await?; - let content_bytes = Bytes::from(content); + let stream = self.file_read.get_file_stream(id).await?; + let mut stream = std::pin::Pin::from(stream); + let mut buf = BytesMut::with_capacity(file_size as usize); + while let Some(chunk) = stream.next().await { + buf.extend_from_slice(&chunk.map_err(|e| { + DomainError::internal_error("File", format!("Stream read error: {}", e)) + })?); + } + let content_bytes = buf.freeze(); // Store in cache if let Some(cache) = &self.content_cache { @@ -231,21 +238,8 @@ impl FileRetrievalService { file_name, file_size / (1024 * 1024) ); - match self.file_read.get_file_stream(id).await { - Ok(stream) => Ok((dto, OptimizedFileContent::Stream(Box::into_pin(stream)))), - Err(e) => { - warn!("Streaming failed, last-resort content load: {}", e); - let content = self.file_read.get_file_content(id).await?; - Ok(( - dto, - OptimizedFileContent::Bytes { - data: Bytes::from(content), - mime_type: mime_type.clone(), - was_transcoded: false, - }, - )) - } - } + let stream = self.file_read.get_file_stream(id).await?; + Ok((dto, OptimizedFileContent::Stream(Box::into_pin(stream)))) } } @@ -273,10 +267,6 @@ impl FileRetrievalUseCase for FileRetrievalService { Ok(files.into_iter().map(FileDto::from).collect()) } - async fn get_file_content(&self, id: &str) -> Result, DomainError> { - self.file_read.get_file_content(id).await - } - async fn get_file_stream( &self, id: &str, diff --git a/src/application/services/share_service.rs b/src/application/services/share_service.rs index d233ee0e..2a34e843 100644 --- a/src/application/services/share_service.rs +++ b/src/application/services/share_service.rs @@ -348,7 +348,7 @@ impl ShareUseCase for ShareService { // Verify the password using the infrastructure port match share.password_hash() { - Some(hash) => self.password_hasher.verify_password(password, hash), + Some(hash) => self.password_hasher.verify_password(password, hash).await, None => Ok(true), // No password required } } @@ -395,12 +395,13 @@ mod tests { struct MockPasswordHasher; + #[async_trait] impl PasswordHasherPort for MockPasswordHasher { - fn hash_password(&self, password: &str) -> Result { + async fn hash_password(&self, password: &str) -> Result { Ok(format!("hashed_{}", password)) } - fn verify_password(&self, _password: &str, _hash: &str) -> Result { + async fn verify_password(&self, _password: &str, _hash: &str) -> Result { Ok(true) } } @@ -439,10 +440,6 @@ mod tests { unimplemented!() } - async fn get_file_content(&self, _id: &str) -> Result, DomainError> { - unimplemented!() - } - async fn get_file_stream( &self, _id: &str, @@ -465,10 +462,6 @@ mod tests { unimplemented!() } - async fn get_file_mmap(&self, _id: &str) -> Result { - unimplemented!() - } - async fn get_file_path( &self, _id: &str, diff --git a/src/application/services/trash_service_test.rs b/src/application/services/trash_service_test.rs index 49408d83..0192c9b5 100644 --- a/src/application/services/trash_service_test.rs +++ b/src/application/services/trash_service_test.rs @@ -142,10 +142,6 @@ impl FileReadPort for MockFileRepository { Ok(vec![]) } - async fn get_file_content(&self, _id: &str) -> std::result::Result, DomainError> { - Ok(vec![]) - } - async fn get_file_stream( &self, _id: &str, @@ -168,10 +164,6 @@ impl FileReadPort for MockFileRepository { unimplemented!() } - async fn get_file_mmap(&self, _id: &str) -> std::result::Result { - unimplemented!() - } - async fn get_file_path(&self, _id: &str) -> std::result::Result { unimplemented!() } diff --git a/src/common/stubs.rs b/src/common/stubs.rs index 6cd37d31..aab72b0f 100644 --- a/src/common/stubs.rs +++ b/src/common/stubs.rs @@ -74,10 +74,6 @@ impl FileReadPort for StubFileReadPort { Ok(Vec::new()) } - async fn get_file_content(&self, _id: &str) -> Result, DomainError> { - Ok(Vec::new()) - } - async fn get_file_stream( &self, _id: &str, @@ -96,10 +92,6 @@ impl FileReadPort for StubFileReadPort { Ok(Box::new(empty_stream)) } - async fn get_file_mmap(&self, _id: &str) -> Result { - Ok(Bytes::new()) - } - async fn get_file_path(&self, _id: &str) -> Result { Ok(StoragePath::from_string("/")) } @@ -525,10 +517,6 @@ impl FileRetrievalUseCase for StubFileRetrievalUseCase { Ok(Vec::new()) } - async fn get_file_content(&self, _id: &str) -> Result, DomainError> { - Ok(Vec::new()) - } - async fn get_file_stream( &self, _id: &str, @@ -697,20 +685,6 @@ impl DedupPort for StubDedupPort { None } - async fn read_blob(&self, _hash: &str) -> Result, DomainError> { - Err(DomainError::internal_error( - "DedupService", - "DedupService not initialized", - )) - } - - async fn read_blob_bytes(&self, _hash: &str) -> Result { - Err(DomainError::internal_error( - "DedupService", - "DedupService not initialized", - )) - } - async fn read_blob_stream( &self, _hash: &str, diff --git a/src/domain/repositories/file_repository.rs b/src/domain/repositories/file_repository.rs index f0dd7495..b5fe889d 100644 --- a/src/domain/repositories/file_repository.rs +++ b/src/domain/repositories/file_repository.rs @@ -34,9 +34,6 @@ pub trait FileReadRepository: Send + Sync + 'static { /// Lists files in a folder. async fn list_files(&self, folder_id: Option<&str>) -> Result, DomainError>; - /// Gets full content as bytes (only for small/medium files). - async fn get_file_content(&self, id: &str) -> Result, DomainError>; - /// Gets content as a stream (ideal for large files). async fn get_file_stream( &self, @@ -51,9 +48,6 @@ pub trait FileReadRepository: Send + Sync + 'static { end: Option, ) -> Result> + Send>, DomainError>; - /// Memory-mapped file for zero-copy access (10–100 MB). - async fn get_file_mmap(&self, id: &str) -> Result; - /// Gets the logical storage path of a file. async fn get_file_path(&self, id: &str) -> Result; diff --git a/src/infrastructure/repositories/pg/file_blob_read_repository.rs b/src/infrastructure/repositories/pg/file_blob_read_repository.rs index b54711dc..a0cab69b 100644 --- a/src/infrastructure/repositories/pg/file_blob_read_repository.rs +++ b/src/infrastructure/repositories/pg/file_blob_read_repository.rs @@ -207,11 +207,6 @@ impl FileReadPort for FileBlobReadRepository { self.resolve_blob_hash(file_id).await } - async fn get_file_content(&self, id: &str) -> Result, DomainError> { - let blob_hash = self.resolve_blob_hash(id).await?; - self.dedup.read_blob(&blob_hash).await - } - async fn get_file_stream( &self, id: &str, @@ -239,13 +234,6 @@ impl FileReadPort for FileBlobReadRepository { Ok(Box::new(stream)) } - async fn get_file_mmap(&self, id: &str) -> Result { - // For RPi targets, mmap is less beneficial than streaming. - // Keep as a fallback that loads content for small/medium files. - let blob_hash = self.resolve_blob_hash(id).await?; - self.dedup.read_blob_bytes(&blob_hash).await - } - async fn get_file_path(&self, id: &str) -> Result { let row = sqlx::query_as::<_, (String, Option)>( r#" diff --git a/src/infrastructure/services/dedup_service.rs b/src/infrastructure/services/dedup_service.rs index 49300c12..48adae61 100644 --- a/src/infrastructure/services/dedup_service.rs +++ b/src/infrastructure/services/dedup_service.rs @@ -31,7 +31,7 @@ use std::path::{Path, PathBuf}; use std::pin::Pin; use std::sync::Arc; use tokio::fs::{self, File}; -use tokio::io::{AsyncReadExt, AsyncSeekExt, BufReader}; +use tokio::io::{AsyncReadExt, AsyncSeekExt}; use tokio_util::io::ReaderStream; use crate::application::ports::dedup_ports::{ @@ -39,10 +39,10 @@ use crate::application::ports::dedup_ports::{ }; use crate::domain::errors::{DomainError, ErrorKind}; -/// Chunk size for streaming hash calculation (256KB) -const HASH_CHUNK_SIZE: usize = 256 * 1024; +/// Block size for SHA-256 file hashing (1MB — optimal syscall/throughput ratio). +const HASH_BLOCK_SIZE: usize = 1024 * 1024; -/// Chunk size for streaming file reads (256 KB — 4x fewer iterations) +/// Chunk size for streaming file reads (256 KB) const STREAM_CHUNK_SIZE: usize = 256 * 1024; /// Content-Addressable Storage Service (PostgreSQL-backed) @@ -124,22 +124,32 @@ impl DedupService { hex::encode(hasher.finalize()) } - /// Calculate SHA-256 hash of a file (streaming). + /// Calculate SHA-256 hash of a file. + /// + /// 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 (~3.8 GB/s on NVMe). pub async fn hash_file(path: &Path) -> std::io::Result { - let file = File::open(path).await?; - let mut reader = BufReader::with_capacity(HASH_CHUNK_SIZE, file); - let mut hasher = Sha256::new(); - let mut buffer = vec![0u8; HASH_CHUNK_SIZE]; + let path = path.to_path_buf(); + tokio::task::spawn_blocking(move || { + use std::io::Read; - loop { - let bytes_read = reader.read(&mut buffer).await?; - if bytes_read == 0 { - break; + let mut file = std::fs::File::open(&path)?; + let mut hasher = Sha256::new(); + 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[..bytes_read]); - } - Ok(hex::encode(hasher.finalize())) + Ok(hex::encode(hasher.finalize())) + }) + .await + .expect("hash_file: spawn_blocking task panicked") } // ── Core store operations ──────────────────────────────────── @@ -515,27 +525,8 @@ impl DedupService { // ── Read operations ────────────────────────────────────────── - /// Read blob content from the filesystem. - pub async fn read_blob(&self, hash: &str) -> Result, DomainError> { - let blob_path = self.blob_path(hash); - - fs::read(&blob_path).await.map_err(|e| { - DomainError::new( - ErrorKind::NotFound, - "Blob", - format!("Failed to read blob {}: {}", hash, e), - ) - }) - } - - /// Read blob content as Bytes. - pub async fn read_blob_bytes(&self, hash: &str) -> Result { - self.read_blob(hash).await.map(Bytes::from) - } - /// Stream blob content in 64 KB chunks — constant memory (~64 KB per stream). /// - /// Unlike `read_blob()`, this never loads the entire file into RAM. /// A 1 GB file uses the same ~64 KB as a 1 KB file. pub async fn read_blob_stream( &self, @@ -772,14 +763,6 @@ impl DedupPort for DedupService { self.get_blob_metadata(hash).await } - async fn read_blob(&self, hash: &str) -> Result, DomainError> { - self.read_blob(hash).await - } - - async fn read_blob_bytes(&self, hash: &str) -> Result { - self.read_blob_bytes(hash).await - } - async fn read_blob_stream( &self, hash: &str, diff --git a/src/infrastructure/services/password_hasher.rs b/src/infrastructure/services/password_hasher.rs index c4f829f0..4bef4ca2 100644 --- a/src/infrastructure/services/password_hasher.rs +++ b/src/infrastructure/services/password_hasher.rs @@ -2,9 +2,14 @@ //! //! This module provides a secure password hashing implementation using the Argon2id //! algorithm, which is the recommended choice for password hashing as of 2023+. +//! +//! Both `hash_password` and `verify_password` are CPU-intensive (~300-500 ms with +//! default parameters) so they run inside `spawn_blocking` to avoid blocking Tokio +//! worker threads. use argon2::password_hash::SaltString; use argon2::{Argon2, PasswordHash, PasswordHasher, PasswordVerifier}; +use async_trait::async_trait; use rand_core::OsRng; use crate::application::ports::auth_ports::PasswordHasherPort; @@ -14,9 +19,11 @@ use crate::common::errors::{DomainError, ErrorKind}; /// /// Uses Argon2id algorithm which provides resistance against both side-channel /// and GPU-based attacks. This is the recommended algorithm for password hashing. +/// +/// The struct is stateless — `Argon2::default()` is constructed per call inside +/// `spawn_blocking` so it is `Send` without extra synchronisation. #[derive(Debug, Clone)] pub struct Argon2PasswordHasher { - /// Argon2 hasher instance - uses default secure parameters _private: (), } @@ -33,35 +40,57 @@ impl Default for Argon2PasswordHasher { } } +#[async_trait] impl PasswordHasherPort for Argon2PasswordHasher { - fn hash_password(&self, password: &str) -> Result { - let salt = SaltString::generate(&mut OsRng); - let argon2 = Argon2::default(); - - argon2 - .hash_password(password.as_bytes(), &salt) - .map(|hash| hash.to_string()) - .map_err(|e| { - DomainError::new( - ErrorKind::InternalError, - "PasswordHasher", - format!("Error generating password hash: {}", e), - ) - }) - } - - fn verify_password(&self, password: &str, hash: &str) -> Result { - let parsed_hash = PasswordHash::new(hash).map_err(|e| { + async fn hash_password(&self, password: &str) -> Result { + let pwd = password.to_owned(); + tokio::task::spawn_blocking(move || { + let salt = SaltString::generate(&mut OsRng); + Argon2::default() + .hash_password(pwd.as_bytes(), &salt) + .map(|hash| hash.to_string()) + .map_err(|e| { + DomainError::new( + ErrorKind::InternalError, + "PasswordHasher", + format!("Error generating password hash: {}", e), + ) + }) + }) + .await + .map_err(|e| { DomainError::new( ErrorKind::InternalError, "PasswordHasher", - format!("Error processing password hash: {}", e), + format!("Task join error: {}", e), ) - })?; + })? + } - Ok(Argon2::default() - .verify_password(password.as_bytes(), &parsed_hash) - .is_ok()) + async fn verify_password(&self, password: &str, hash: &str) -> Result { + let pwd = password.to_owned(); + let hash = hash.to_owned(); + tokio::task::spawn_blocking(move || { + let parsed_hash = PasswordHash::new(&hash).map_err(|e| { + DomainError::new( + ErrorKind::InternalError, + "PasswordHasher", + format!("Error processing password hash: {}", e), + ) + })?; + + Ok(Argon2::default() + .verify_password(pwd.as_bytes(), &parsed_hash) + .is_ok()) + }) + .await + .map_err(|e| { + DomainError::new( + ErrorKind::InternalError, + "PasswordHasher", + format!("Task join error: {}", e), + ) + })? } } @@ -69,33 +98,36 @@ impl PasswordHasherPort for Argon2PasswordHasher { mod tests { use super::*; - #[test] - fn test_hash_and_verify_password() { + #[tokio::test] + async fn test_hash_and_verify_password() { let hasher = Argon2PasswordHasher::new(); let password = "test_password_123"; let hash = hasher .hash_password(password) + .await .expect("Should hash password"); assert!( hasher .verify_password(password, &hash) + .await .expect("Should verify") ); assert!( !hasher .verify_password("wrong_password", &hash) + .await .expect("Should verify") ); } - #[test] - fn test_different_hashes_for_same_password() { + #[tokio::test] + async fn test_different_hashes_for_same_password() { let hasher = Argon2PasswordHasher::new(); let password = "same_password"; - let hash1 = hasher.hash_password(password).expect("Should hash"); - let hash2 = hasher.hash_password(password).expect("Should hash"); + let hash1 = hasher.hash_password(password).await.expect("Should hash"); + let hash2 = hasher.hash_password(password).await.expect("Should hash"); // Hashes should be different due to random salt assert_ne!(hash1, hash2); @@ -104,11 +136,13 @@ mod tests { assert!( hasher .verify_password(password, &hash1) + .await .expect("Should verify") ); assert!( hasher .verify_password(password, &hash2) + .await .expect("Should verify") ); } diff --git a/src/interfaces/api/handlers/dedup_handler.rs b/src/interfaces/api/handlers/dedup_handler.rs index 5515213c..e4aa3d1d 100644 --- a/src/interfaces/api/handlers/dedup_handler.rs +++ b/src/interfaces/api/handlers/dedup_handler.rs @@ -313,13 +313,26 @@ impl DedupHandler { .and_then(|m| m.content_type.clone()) .unwrap_or_else(|| "application/octet-stream".to_string()); - match dedup.read_blob_bytes(&hash).await { - Ok(content) => Response::builder() + // Stream blob in 64 KB chunks — constant memory regardless of size + let size = match dedup.blob_size(&hash).await { + Ok(s) => s, + Err(_) => { + return Response::builder() + .status(StatusCode::NOT_FOUND) + .header(header::CONTENT_TYPE, "application/json") + .body(Body::from(r#"{"error": "Blob not found"}"#)) + .unwrap() + .into_response(); + } + }; + + match dedup.read_blob_stream(&hash).await { + Ok(stream) => Response::builder() .status(StatusCode::OK) .header(header::CONTENT_TYPE, content_type) - .header(header::CONTENT_LENGTH, content.len().to_string()) + .header(header::CONTENT_LENGTH, size.to_string()) .header("X-Dedup-Hash", &hash) - .body(Body::from(content)) + .body(Body::from_stream(stream)) .unwrap() .into_response(), Err(_) => Response::builder() diff --git a/src/interfaces/api/handlers/webdav_handler.rs b/src/interfaces/api/handlers/webdav_handler.rs index 1720bd63..7dca6b7a 100644 --- a/src/interfaces/api/handlers/webdav_handler.rs +++ b/src/interfaces/api/handlers/webdav_handler.rs @@ -936,7 +936,7 @@ async fn handle_copy( // Get services from state let file_retrieval_service = &state.applications.file_retrieval_service; - let file_upload_service = &state.applications.file_upload_service; + let _file_upload_service = &state.applications.file_upload_service; let folder_service = &state.applications.folder_service; // Check if destination already exists (for Overwrite header compliance) @@ -996,26 +996,24 @@ async fn handle_copy( })?; if recursive { - // Copy subfolders and files (simplified implementation) + // Copy files via zero-copy dedup (only increments blob ref_count) let files = file_retrieval_service .list_files(Some(&folder.id)) .await .map_err(|e| AppError::internal_error(format!("Failed to list files: {}", e)))?; + let file_management_service = &state.applications.file_management_service; + let new_folder_id = Some(_new_folder.id.clone()); for file in files { - // Get file content - if let Ok(content) = file_retrieval_service.get_file_content(&file.id).await { - // Create new file in destination - file_upload_service - .create_file(&destination_path, &file.name, &content, &file.mime_type) - .await - .map_err(|e| { - AppError::internal_error(format!( - "Failed to copy file {}: {}", - file.name, e - )) - })?; - } + file_management_service + .copy_file(&file.id, new_folder_id.clone()) + .await + .map_err(|e| { + AppError::internal_error(format!( + "Failed to copy file {}: {}", + file.name, e + )) + })?; } } } else { diff --git a/src/interfaces/api/handlers/wopi_handler.rs b/src/interfaces/api/handlers/wopi_handler.rs index 6224a5a2..f4ab0086 100644 --- a/src/interfaces/api/handlers/wopi_handler.rs +++ b/src/interfaces/api/handlers/wopi_handler.rs @@ -117,6 +117,9 @@ async fn check_file_info( } /// GET /wopi/files/{file_id}/contents — GetFile +/// +/// Streams the file content to Collabora/OnlyOffice in 64 KB chunks. +/// Memory usage is constant (~64 KB) regardless of file size. async fn get_file( Path(file_id): Path, Query(token_query): Query, @@ -138,10 +141,13 @@ async fn get_file( .app_state .applications .file_retrieval_service - .get_file_content(&file_id) + .get_file_stream(&file_id) .await { - Ok(content) => (StatusCode::OK, content).into_response(), + Ok(stream) => { + let body = axum::body::Body::from_stream(std::pin::Pin::from(stream)); + (StatusCode::OK, body).into_response() + } Err(_) => StatusCode::NOT_FOUND.into_response(), } }