From 5a1959bf239ec7fdf3f88d698be3e5745c86a4a3 Mon Sep 17 00:00:00 2001 From: Dionisio Date: Wed, 25 Feb 2026 23:41:16 +0100 Subject: [PATCH] =?UTF-8?q?perf:=20eliminate=20Vec=20buffer=20paths=20?= =?UTF-8?q?=E2=80=94=20all=20uploads=20now=20stream=20to=20disk?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Issue #4 (HIGH): save_file(Vec) and update_file_content(Vec) accepted up to 10 MB of contiguous memory per request. While the main upload paths already used streaming, the WebDAV compat methods (create_file, update_file) and the empty-file handler still used the buffered path, creating a .to_vec() copy. Changes: - FileWritePort trait: remove save_file(Vec) and update_file_content(Vec) — only streaming variants remain - FileUploadUseCase trait: remove upload_file(Vec) - file_upload_service.rs: create_file() and update_file() now spool &[u8] to NamedTempFile + Sha256::digest, then delegate to streaming path (save_file_from_temp / update_file_streaming) - file_handler.rs: empty file uploads use upload_file_streaming with - FileBlobWriteRepository: remove save_file and update_file_content impls - StubFileWritePort, StubFileUploadUseCase, MockFileRepository: remove corresponding dead method impls Impact: impossible to accidentally use a buffered upload path. All content goes through streaming with ~256 KB peak RAM. -166 LOC. --- src/application/ports/file_ports.rs | 24 ++--- src/application/ports/storage_ports.rs | 13 --- .../services/file_upload_service.rs | 84 ++++++++-------- .../services/trash_service_test.rs | 43 +-------- src/common/stubs.rs | 28 ------ .../pg/file_blob_write_repository.rs | 95 ------------------- src/interfaces/api/handlers/file_handler.rs | 13 ++- 7 files changed, 67 insertions(+), 233 deletions(-) diff --git a/src/application/ports/file_ports.rs b/src/application/ports/file_ports.rs index 42d1202e..a6951a76 100644 --- a/src/application/ports/file_ports.rs +++ b/src/application/ports/file_ports.rs @@ -15,13 +15,18 @@ use crate::common::errors::DomainError; /// Primary port for file upload operations. /// -/// All upload paths converge on streaming-to-disk: +/// **All upload paths converge on streaming-to-disk** — no method accepts +/// `Vec` for content. Even `create_file` / `update_file` (WebDAV +/// helpers that receive `&[u8]`) spool to a temp file internally so that +/// peak RAM stays at ~256 KB regardless of file size. +/// /// - Normal uploads: handler spools multipart to temp file → `upload_file_streaming` -/// - WebDAV PUT: small in-memory buffer → `upload_file` /// - Chunked uploads: chunks already on disk → `upload_file_from_path` +/// - WebDAV PUT (new): handler streams to temp file → `update_file_streaming` +/// - WebDAV PUT (small/compat): `create_file` / `update_file` spool internally #[async_trait] pub trait FileUploadUseCase: Send + Sync + 'static { - /// Upload from a temp file already on disk (true streaming, ~64 KB RAM). + /// Upload from a temp file already on disk (true streaming, ~256 KB RAM). /// /// When `pre_computed_hash` is `Some`, the blob store skips the hash /// re-read — the handler already computed it during the multipart spool. @@ -35,19 +40,6 @@ pub trait FileUploadUseCase: Send + Sync + 'static { pre_computed_hash: Option, ) -> Result; - /// Upload from in-memory bytes (for small payloads: WebDAV, empty files). - /// - /// Only used for WebDAV PUT and empty files where the content is already - /// buffered by the protocol handler. For normal uploads, prefer - /// `upload_file_streaming`. - async fn upload_file( - &self, - name: String, - folder_id: Option, - content_type: String, - content: Vec, - ) -> Result; - /// Upload from a file already assembled on disk (chunked uploads). /// /// Same as `upload_file_streaming` but with a separate name for clarity. diff --git a/src/application/ports/storage_ports.rs b/src/application/ports/storage_ports.rs index 0153107e..b1115533 100644 --- a/src/application/ports/storage_ports.rs +++ b/src/application/ports/storage_ports.rs @@ -197,15 +197,6 @@ pub struct CopyFolderTreeResult { /// and deferred registration for the write-behind cache. #[async_trait] pub trait FileWritePort: Send + Sync + 'static { - /// Saves a new file from bytes. - async fn save_file( - &self, - name: String, - folder_id: Option, - content_type: String, - content: Vec, - ) -> Result; - /// Streaming upload — saves a file from a temp file already on disk. /// /// When `pre_computed_hash` is provided, the dedup service skips the @@ -233,10 +224,6 @@ pub trait FileWritePort: Send + Sync + 'static { /// Deletes a file. async fn delete_file(&self, id: &str) -> Result<(), DomainError>; - /// Updates the content of an existing file. - async fn update_file_content(&self, file_id: &str, content: Vec) - -> Result<(), DomainError>; - /// Streaming update — replaces file content from a temp file on disk. /// /// When `pre_computed_hash` is provided, the dedup service skips the diff --git a/src/application/services/file_upload_service.rs b/src/application/services/file_upload_service.rs index 146ee358..ce8e5e42 100644 --- a/src/application/services/file_upload_service.rs +++ b/src/application/services/file_upload_service.rs @@ -1,4 +1,5 @@ use async_trait::async_trait; +use sha2::{Digest, Sha256}; use std::path::Path; use std::sync::Arc; @@ -29,12 +30,16 @@ fn extract_username_from_path(path: &str) -> Option { /// Service for file upload operations. /// -/// All upload paths converge on streaming-to-disk: +/// **Every upload path converges on streaming-to-disk** — there is no +/// `Vec` buffer path. +/// /// - **Normal uploads**: handler spools multipart to temp file → `upload_file_streaming` /// - **Chunked uploads**: chunks already on disk → `upload_file_from_path` -/// - **WebDAV PUT / empty files**: small in-memory buffer → `upload_file` +/// - **WebDAV PUT (large)**: handler streams body to temp file → `update_file_streaming` +/// - **WebDAV PUT (small / compat)**: `create_file` / `update_file` spool `&[u8]` +/// to a temp file internally, then call the streaming path. /// -/// Peak RAM usage during upload: ~256 KB (streaming hash) regardless of file size. +/// Peak RAM usage during any upload: ~256 KB (streaming hash) regardless of file size. pub struct FileUploadService { /// Write port — handles save, streaming, deferred registration file_write: Arc, @@ -136,23 +141,6 @@ impl FileUploadUseCase for FileUploadService { Ok(dto) } - /// Simple byte-based upload (for WebDAV and empty files only). - async fn upload_file( - &self, - name: String, - folder_id: Option, - content_type: String, - content: Vec, - ) -> Result { - let file = self - .file_write - .save_file(name, folder_id, content_type, content) - .await?; - let dto = FileDto::from(file); - self.maybe_update_storage_usage(&dto); - Ok(dto) - } - /// Upload from a file already on disk (chunked uploads). async fn upload_file_from_path( &self, @@ -184,6 +172,10 @@ impl FileUploadUseCase for FileUploadService { } /// Creates a file at a specific path (for WebDAV PUT on new resource). + /// + /// Spools the in-memory `&[u8]` to a temp file with hash-on-write, + /// then delegates to the streaming path. Peak RAM: the caller's + /// buffer + ~256 KB for the hasher. async fn create_file( &self, parent_path: &str, @@ -201,13 +193,24 @@ impl FileUploadUseCase for FileUploadService { None }; + // Spool to temp file + hash + let temp = tempfile::NamedTempFile::new().map_err(|e| { + DomainError::internal_error("FileUpload", format!("temp file: {e}")) + })?; + tokio::fs::write(temp.path(), content).await.map_err(|e| { + DomainError::internal_error("FileUpload", format!("write temp: {e}")) + })?; + let hash = hex::encode(Sha256::digest(content)); + let file = self .file_write - .save_file( + .save_file_from_temp( filename.to_string(), parent_id, content_type.to_string(), - content.to_vec(), + temp.path(), + content.len() as u64, + Some(hash), ) .await?; let dto = FileDto::from(file); @@ -216,26 +219,27 @@ impl FileUploadUseCase for FileUploadService { } /// Updates an existing file's content, or creates it if not found (for WebDAV PUT). + /// + /// Spools the in-memory `&[u8]` to a temp file with hash-on-write, + /// then delegates to the streaming update/create path. async fn update_file(&self, path: &str, content: &[u8]) -> Result<(), DomainError> { - // Direct SQL lookup — O(folder_depth) instead of O(total_files) - if let Some(file_read) = &self.file_read - && let Some(file) = file_read.find_file_by_path(path).await? - { - self.file_write - .update_file_content(file.id(), content.to_vec()) - .await?; - return Ok(()); - } + // Spool to temp file + hash + let temp = tempfile::NamedTempFile::new().map_err(|e| { + DomainError::internal_error("FileUpload", format!("temp file: {e}")) + })?; + tokio::fs::write(temp.path(), content).await.map_err(|e| { + DomainError::internal_error("FileUpload", format!("write temp: {e}")) + })?; + let hash = hex::encode(Sha256::digest(content)); - let path_normalized = path.trim_start_matches('/').trim_end_matches('/'); - let (parent_path, filename) = if let Some(idx) = path_normalized.rfind('/') { - (&path_normalized[..idx], &path_normalized[idx + 1..]) - } else { - ("", path_normalized) - }; - self.create_file(parent_path, filename, content, "application/octet-stream") - .await?; - Ok(()) + self.update_file_streaming( + path, + temp.path(), + content.len() as u64, + "application/octet-stream", + Some(hash), + ) + .await } /// Streaming update — replaces file content from a temp file on disk. diff --git a/src/application/services/trash_service_test.rs b/src/application/services/trash_service_test.rs index 424a5f81..cc5c8ea4 100644 --- a/src/application/services/trash_service_test.rs +++ b/src/application/services/trash_service_test.rs @@ -209,16 +209,6 @@ impl FileReadPort for MockFileRepository { #[async_trait] impl FileWritePort for MockFileRepository { - async fn save_file( - &self, - _name: String, - _folder_id: Option, - _content_type: String, - _content: Vec, - ) -> std::result::Result { - unimplemented!() - } - async fn save_file_from_temp( &self, _name: String, @@ -251,14 +241,6 @@ impl FileWritePort for MockFileRepository { Ok(()) } - async fn update_file_content( - &self, - _file_id: &str, - _content: Vec, - ) -> std::result::Result<(), DomainError> { - Ok(()) - } - async fn update_file_content_from_temp( &self, _file_id: &str, @@ -514,10 +496,7 @@ mod tests { // Arrange let trashed_files = Arc::new(Mutex::new(HashMap::new())); let trashed_folders = Arc::new(Mutex::new(HashMap::new())); - let trash_repo = Arc::new(MockTrashRepository::new( - trashed_files.clone(), - trashed_folders.clone(), - )); + let trash_repo = Arc::new(MockTrashRepository::new(trashed_files.clone(), trashed_folders.clone())); let file_repo = Arc::new(MockFileRepository::new(trashed_files)); let folder_repo = Arc::new(MockFolderRepository::new(trashed_folders)); @@ -588,10 +567,7 @@ mod tests { // Arrange let trashed_files = Arc::new(Mutex::new(HashMap::new())); let trashed_folders = Arc::new(Mutex::new(HashMap::new())); - let trash_repo = Arc::new(MockTrashRepository::new( - trashed_files.clone(), - trashed_folders.clone(), - )); + let trash_repo = Arc::new(MockTrashRepository::new(trashed_files.clone(), trashed_folders.clone())); let file_repo = Arc::new(MockFileRepository::new(trashed_files)); let folder_repo = Arc::new(MockFolderRepository::new(trashed_folders)); @@ -653,10 +629,7 @@ mod tests { // Arrange let trashed_files = Arc::new(Mutex::new(HashMap::new())); let trashed_folders = Arc::new(Mutex::new(HashMap::new())); - let trash_repo = Arc::new(MockTrashRepository::new( - trashed_files.clone(), - trashed_folders.clone(), - )); + let trash_repo = Arc::new(MockTrashRepository::new(trashed_files.clone(), trashed_folders.clone())); let file_repo = Arc::new(MockFileRepository::new(trashed_files)); let folder_repo = Arc::new(MockFolderRepository::new(trashed_folders)); @@ -723,10 +696,7 @@ mod tests { // Arrange let trashed_files = Arc::new(Mutex::new(HashMap::new())); let trashed_folders = Arc::new(Mutex::new(HashMap::new())); - let trash_repo = Arc::new(MockTrashRepository::new( - trashed_files.clone(), - trashed_folders.clone(), - )); + let trash_repo = Arc::new(MockTrashRepository::new(trashed_files.clone(), trashed_folders.clone())); let file_repo = Arc::new(MockFileRepository::new(trashed_files)); let folder_repo = Arc::new(MockFolderRepository::new(trashed_folders)); @@ -792,10 +762,7 @@ mod tests { // Arrange let trashed_files = Arc::new(Mutex::new(HashMap::new())); let trashed_folders = Arc::new(Mutex::new(HashMap::new())); - let trash_repo = Arc::new(MockTrashRepository::new( - trashed_files.clone(), - trashed_folders.clone(), - )); + let trash_repo = Arc::new(MockTrashRepository::new(trashed_files.clone(), trashed_folders.clone())); let file_repo = Arc::new(MockFileRepository::new(trashed_files)); let folder_repo = Arc::new(MockFolderRepository::new(trashed_folders)); diff --git a/src/common/stubs.rs b/src/common/stubs.rs index 38852377..6eff7934 100644 --- a/src/common/stubs.rs +++ b/src/common/stubs.rs @@ -131,16 +131,6 @@ pub struct StubFileWritePort; #[async_trait] impl FileWritePort for StubFileWritePort { - async fn save_file( - &self, - _name: String, - _folder_id: Option, - _content_type: String, - _content: Vec, - ) -> Result { - Ok(File::default()) - } - async fn save_file_from_temp( &self, _name: String, @@ -177,14 +167,6 @@ impl FileWritePort for StubFileWritePort { Ok(()) } - async fn update_file_content( - &self, - _file_id: &str, - _content: Vec, - ) -> Result<(), DomainError> { - Ok(()) - } - async fn update_file_content_from_temp( &self, _file_id: &str, @@ -454,16 +436,6 @@ impl FileUploadUseCase for StubFileUploadUseCase { Ok(FileDto::default()) } - async fn upload_file( - &self, - _name: String, - _folder_id: Option, - _content_type: String, - _content: Vec, - ) -> Result { - Ok(FileDto::default()) - } - async fn upload_file_from_path( &self, _name: String, diff --git a/src/infrastructure/repositories/pg/file_blob_write_repository.rs b/src/infrastructure/repositories/pg/file_blob_write_repository.rs index 3f3de58b..212a8c54 100644 --- a/src/infrastructure/repositories/pg/file_blob_write_repository.rs +++ b/src/infrastructure/repositories/pg/file_blob_write_repository.rs @@ -182,88 +182,6 @@ impl FileBlobWriteRepository { #[async_trait] impl FileWritePort for FileBlobWriteRepository { - async fn save_file( - &self, - name: String, - folder_id: Option, - content_type: String, - content: Vec, - ) -> Result { - let user_id = self.resolve_user_id(folder_id.as_deref()).await?; - let size = content.len() as i64; - - // Store content in blob store - let dedup_result = self - .dedup - .store_bytes(&content, Some(content_type.clone())) - .await?; - let blob_hash = dedup_result.hash().to_string(); - - // Insert file metadata — if this fails, compensate by removing the blob ref - let row = match sqlx::query_as::<_, (String, i64, i64)>( - r#" - INSERT INTO storage.files (name, folder_id, user_id, blob_hash, size, mime_type) - VALUES ($1, $2::uuid, $3, $4, $5, $6) - RETURNING id::text, - EXTRACT(EPOCH FROM created_at)::bigint, - EXTRACT(EPOCH FROM updated_at)::bigint - "#, - ) - .bind(&name) - .bind(&folder_id) - .bind(&user_id) - .bind(&blob_hash) - .bind(size) - .bind(&content_type) - .fetch_one(self.pool.as_ref()) - .await - { - Ok(row) => row, - Err(e) => { - // ── Compensation: undo the blob ref so it doesn't become orphaned ── - if let Err(rollback_err) = self.dedup.remove_reference(&blob_hash).await { - tracing::error!( - "Blob orphaned after failed INSERT — hash: {}, err: {}", - &blob_hash[..12], - rollback_err - ); - } - if let sqlx::Error::Database(ref db_err) = e - && db_err.code().as_deref() == Some("23505") - { - return Err(DomainError::already_exists( - "File", - format!("{name} already exists in folder"), - )); - } - return Err(DomainError::internal_error( - "FileBlobWrite", - format!("insert: {e}"), - )); - } - }; - - tracing::info!( - "💾 BLOB WRITE: {} ({} bytes, hash: {})", - name, - size, - &blob_hash[..12] - ); - - let folder_path = self.lookup_folder_path(folder_id.as_deref()).await?; - Self::row_to_file( - row.0, - name, - folder_id, - folder_path, - size, - content_type, - row.1, - row.2, - Some(user_id), - ) - } - async fn save_file_from_temp( &self, name: String, @@ -534,19 +452,6 @@ impl FileWritePort for FileBlobWriteRepository { Ok(()) } - async fn update_file_content( - &self, - file_id: &str, - content: Vec, - ) -> Result<(), DomainError> { - // Store new content first (blob store is idempotent) - let new_size = content.len() as i64; - let dedup_result = self.dedup.store_bytes(&content, None).await?; - let new_hash = dedup_result.hash().to_string(); - - self.swap_blob_hash(file_id, &new_hash, new_size).await - } - async fn update_file_content_from_temp( &self, file_id: &str, diff --git a/src/interfaces/api/handlers/file_handler.rs b/src/interfaces/api/handlers/file_handler.rs index a3f5b243..cef6dbc9 100644 --- a/src/interfaces/api/handlers/file_handler.rs +++ b/src/interfaces/api/handlers/file_handler.rs @@ -167,11 +167,18 @@ impl FileHandler { )); } - // Empty file — use in-memory path + // Empty file — use streaming path with the (empty) temp file if total_size == 0 { - let _ = tokio::fs::remove_file(&temp_path).await; + let hash = hex::encode(hasher.finalize()); return upload_service - .upload_file(filename, folder_id, content_type, vec![]) + .upload_file_streaming( + filename, + folder_id, + content_type, + &temp_path, + 0, + Some(hash), + ) .await .map_err(Self::domain_error_response); }