From b06206207f3fcc8b9198523d54d0ed219f9e67c1 Mon Sep 17 00:00:00 2001 From: Dionisio Date: Mon, 2 Mar 2026 01:30:34 +0100 Subject: [PATCH] =?UTF-8?q?perf:=20optimize=20hot=20paths=20=E2=80=94=20ba?= =?UTF-8?q?tch=20concurrency,=20pagination,=20sorting,=20transcoding,=20fo?= =?UTF-8?q?lder=20ops?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - batch_operations.rs: replace join_all with buffer_unordered, Arc for shared IDs, remove redundant clones and dead Semaphore - folder_db_repository.rs: use COUNT(*) OVER() for single-query pagination; UPDATE RETURNING for rename/move (eliminates extra SELECTs) - folder_service.rs: remove StorageTransaction wrapper from rename/move — direct repo call (4→2 and 5→3 queries) - search_service.rs: replace sort_by(to_lowercase) with sort_by_cached_key (N vs 2·N·log₂N allocations) - image_transcode_service.rs: dynamic rayon pool sizing via available_parallelism() instead of hardcoded 2 threads - Remove dead transactions module (zero consumers after folder_service refactor) --- PERFORMANCE_AUDIT.md | 810 ------------------ src/application/mod.rs | 1 - src/application/services/batch_operations.rs | 278 ++---- src/application/services/folder_service.rs | 99 +-- src/application/services/search_service.rs | 15 +- src/application/transactions/mod.rs | 1 - .../transactions/storage_transaction.rs | 150 ---- .../repositories/pg/folder_db_repository.rs | 106 +-- .../services/image_transcode_service.rs | 20 +- 9 files changed, 169 insertions(+), 1311 deletions(-) delete mode 100644 PERFORMANCE_AUDIT.md delete mode 100644 src/application/transactions/mod.rs delete mode 100644 src/application/transactions/storage_transaction.rs diff --git a/PERFORMANCE_AUDIT.md b/PERFORMANCE_AUDIT.md deleted file mode 100644 index d486f8c0..00000000 --- a/PERFORMANCE_AUDIT.md +++ /dev/null @@ -1,810 +0,0 @@ -# OxiCloud — Comprehensive Architecture & Performance Audit - -> **Scope**: Full source-level analysis of all layers (domain → infrastructure → application → interfaces). -> **Methodology**: Static analysis of every critical `.rs` file. No code changes made. - ---- - -## Table of Contents - -1. [CRITICAL — `std::sync::Mutex` Blocking the Tokio Runtime](#1-critical--stdsyncmutex-blocking-the-tokio-runtime) -2. [CRITICAL — ZIP Service Loads Entire Files into Memory](#2-critical--zip-service-loads-entire-files-into-memory) -3. [CRITICAL — Share Repository: JSON File I/O per Operation](#3-critical--share-repository-json-file-io-per-operation) -4. [HIGH — Blocking Filesystem Calls in Async Context](#4-high--blocking-filesystem-calls-in-async-context) -5. [HIGH — Unbounded Task Spawning in Recursive Search](#5-high--unbounded-task-spawning-in-recursive-search) -6. [HIGH — Thumbnail Cache Write-Lock Contention on Reads](#6-high--thumbnail-cache-write-lock-contention-on-reads) -7. [HIGH — HTTP Cache Middleware Buffers Entire Response Bodies](#7-high--http-cache-middleware-buffers-entire-response-bodies) -8. [MEDIUM — N+1 Queries / Extra Database Round Trips](#8-medium--n1-queries--extra-database-round-trips) -9. [MEDIUM — Unnecessary String Allocations in Error Paths](#9-medium--unnecessary-string-allocations-in-error-paths) -10. [MEDIUM — Redundant Path String in Domain Entities](#10-medium--redundant-path-string-in-domain-entities) -11. [MEDIUM — Unbounded Parallel Tasks in Storage Usage Update](#11-medium--unbounded-parallel-tasks-in-storage-usage-update) -12. [MEDIUM — Upload Handler Re-parses Its Own HTTP Response](#12-medium--upload-handler-re-parses-its-own-http-response) -13. [LOW — One-Shot Cache Pattern Defeats Caching Purpose](#13-low--one-shot-cache-pattern-defeats-caching-purpose) -14. [LOW — Duplicated SQL in Paginated Search](#14-low--duplicated-sql-in-paginated-search) -15. [LOW — Sequential Trash Cleanup Without Batching](#15-low--sequential-trash-cleanup-without-batching) -16. [LOW — Search Cache Key Serializes Entire DTO to JSON](#16-low--search-cache-key-serializes-entire-dto-to-json) -17. [Positive Patterns — What's Done Well](#17-positive-patterns--whats-done-well) -18. [Summary Matrix](#18-summary-matrix) - ---- - -## 1. CRITICAL — `std::sync::Mutex` Blocking the Tokio Runtime - -### Location - -| File | Line(s) | Symbol | -|------|---------|--------| -| `src/application/services/search_service.rs` | 4, 56 | `search_cache: Arc>>` | -| `src/interfaces/middleware/cache.rs` | 14, 51 | `cache: Arc>>` | -| `src/application/services/auth_application_service.rs` | 62–63 | `pending_oidc_flows`, `pending_oidc_tokens` | - -### Problematic Pattern - -```rust -// search_service.rs:4 -use std::sync::Mutex; - -// search_service.rs:56 -search_cache: Arc>>, -``` - -Every call to `.lock()` on a `std::sync::Mutex` across an `.await` boundary **blocks the entire Tokio worker thread**. If all Tokio workers are blocked on the Mutex simultaneously, the runtime deadlocks. - -In `search_service.rs`, `get_from_cache()` and `store_in_cache()` both call `.lock()`, and `store_in_cache()` does eviction work (iteration + removal) while holding the lock. The cleanup task (`start_cache_cleanup_task`) also locks the Mutex inside a `tokio::spawn` future. - -In `cache.rs`, every HTTP GET request passes through `.get()` or `.set()`, each calling `self.cache.lock().unwrap()`. The `evict_oldest()` method sorts all entries by timestamp while the parent lock is held. - -### Impact - -- **Severity**: CRITICAL under concurrent load. -- Under 50+ concurrent requests, Tokio worker threads park on the Mutex, causing tail-latency spikes (p99 > 100ms) and potential deadlock. -- The cleanup tasks also lock, creating periodic contention peaks. - -### Fix Sketch - -**Option A — Replace with `tokio::sync::RwLock`** (minimal change): -```rust -use tokio::sync::RwLock; -search_cache: Arc>>, -``` - -**Option B — Replace with `moka` (lock-free, recommended)**: -```rust -use moka::future::Cache; - -// In SearchService -search_cache: Cache, - -// Construction -let search_cache = Cache::builder() - .max_capacity(max_cache_size as u64) - .time_to_live(Duration::from_secs(cache_ttl)) - .build(); -``` -This eliminates all manual eviction logic and the cleanup task entirely. Already used successfully in `image_transcode_service.rs` and `file_content_cache.rs`. - -For the HTTP cache middleware: consider replacing with `moka::future::Cache`. - -For the auth service: the OIDC maps are short-lived and low-contention, so `tokio::sync::Mutex` would suffice, or use `dashmap::DashMap`. - ---- - -## 2. CRITICAL — ZIP Service Loads Entire Files into Memory - -### Location - -| File | Line(s) | Symbol | -|------|---------|--------| -| `src/infrastructure/services/zip_service.rs` | 209–230 | `add_file_to_zip()` | - -### Problematic Pattern - -```rust -// zip_service.rs: add_file_to_zip() -async fn add_file_to_zip( - &self, - zip: &mut ZipWriter>>, - // ... -) -> Result<()> { - // Loads ENTIRE file content into memory - let content = self.file_service.get_file_content(&file_id).await?; - zip.write_all(&content)?; -} -``` - -For a folder download containing N files of size S, memory usage is `O(N × S)` **plus** the ZIP buffer itself. A folder with 100 × 100MB files = 10GB in RAM. - -### Impact - -- **Severity**: CRITICAL for large folders. OOM-kill risk in production. -- The ZIP buffer (`Cursor>`) also holds the entire compressed output in memory. - -### Fix Sketch - -Use `tokio::io::AsyncRead` + streaming ZIP writer (e.g., `async_zip` crate): -```rust -// Stream-based approach: -let blob_stream = self.dedup_service.read_blob_stream(&blob_hash).await?; -// Pipe directly to zip writer without buffering the entire file -zip.write_entry_stream(file_name, blob_stream).await?; -``` -Alternatively, use `read_blob_stream()` (which already exists in `DedupService` with 256KB chunks) and write chunks incrementally to the ZIP. - ---- - -## 3. CRITICAL — Share Repository: JSON File I/O per Operation - -### Location - -| File | Line(s) | Symbol | -|------|---------|--------| -| `src/infrastructure/repositories/share_fs_repository.rs` | 1–286 | `ShareFsRepository` | - -### Problematic Pattern - -```rust -// Every read operation: -async fn get_share(&self, id: &str) -> Result<…> { - let shares = self.read_shares().await?; // Read ENTIRE file - shares.into_iter().find(|s| s.id == id) // Linear scan -} - -// Every write operation: -async fn create_share(&self, share: Share) -> Result<…> { - let mut shares = self.read_shares().await?; // Read ENTIRE file - shares.push(share); - self.write_shares(&shares).await?; // Write ENTIRE file -} -``` - -### Impact - -- **Severity**: CRITICAL for concurrent users. -- **Race condition**: Two concurrent `create_share()` calls read the same file, each appends its share, and the second write loses the first share. -- **O(n)** per operation — every read scans all shares. -- **Blocking I/O**: `tokio::fs::read` / `tokio::fs::write` are async but the entire file is serialized/deserialized on every call. - -### Fix Sketch - -**Option A — Migrate to PostgreSQL** (recommended, consistent with other repos): -```sql -CREATE TABLE storage.shares ( - id UUID PRIMARY KEY DEFAULT gen_random_uuid(), - resource_id UUID NOT NULL, - resource_type TEXT NOT NULL, - token TEXT UNIQUE NOT NULL, - password_hash TEXT, - expires_at TIMESTAMPTZ, - created_at TIMESTAMPTZ DEFAULT now() -); -CREATE INDEX idx_shares_token ON storage.shares(token); -``` - -**Option B — Add file locking + in-memory index** (minimal change): -```rust -struct ShareFsRepository { - shares: Arc>>, // In-memory index - path: PathBuf, - file_lock: tokio::sync::Mutex<()>, // Serialize writes -} -``` - ---- - -## 4. HIGH — Blocking Filesystem Calls in Async Context - -### Location - -| File | Line(s) | Symbol | -|------|---------|--------| -| `src/infrastructure/services/path_service.rs` | 137, 148, 165, 172 | `physical_path.exists()`, `.is_file()`, `.is_dir()` | -| `src/main.rs` | 60, 64 | `std::fs::create_dir_all()` | -| `src/infrastructure/services/dedup_service.rs` | 224 | `std::fs::remove_file()` | - -### Problematic Pattern - -```rust -// path_service.rs — inside async fn file_exists() -async fn file_exists(&self, storage_path: &StoragePath) -> Result { - let physical_path = self.resolve_path(storage_path); - let exists = physical_path.exists() && physical_path.is_file(); // BLOCKING - Ok(exists) -} - -// Also in directory_exists() and ensure_directory() -``` - -`Path::exists()`, `.is_file()`, and `.is_dir()` perform synchronous `stat()` syscalls. On network-attached storage (NFS, CIFS) or slow disks, these can take 10–100ms, blocking a Tokio worker. - -### Impact - -- **Severity**: HIGH on NFS/CIFS storage; moderate on local SSD. -- `path_service.rs` is called by the StoragePort trait used throughout the application. - -### Fix Sketch - -```rust -async fn file_exists(&self, storage_path: &StoragePath) -> Result { - let physical_path = self.resolve_path(storage_path); - match tokio::fs::metadata(&physical_path).await { - Ok(meta) => Ok(meta.is_file()), - Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(false), - Err(e) => Err(DomainError::from(e)), - } -} -``` - -For `main.rs` (startup-only), the blocking calls are acceptable but could use `tokio::fs::create_dir_all()` for consistency. - ---- - -## 5. HIGH — Unbounded Task Spawning in Recursive Search - -### Location - -| File | Line(s) | Symbol | -|------|---------|--------| -| `src/application/services/search_service.rs` | 310–360 | `search_parallel()` | - -### Problematic Pattern - -```rust -fn search_parallel(…) -> Pin + Send>> { - Box::pin(async move { - let folders = folder_repo.list_folders(current_folder_id.as_deref()).await?; - - // Spawns one task PER subfolder — NO concurrency limit - let mut handles = Vec::with_capacity(folder_dtos.len()); - for subfolder in &folder_dtos { - handles.push(tokio::spawn(async move { - Self::search_parallel(fr, fdr, Some(folder_id), crit).await - })); - } - // ...joins all - }) -} -``` - -For a directory tree of depth D with branching factor B, this spawns `B^D` tasks. A user with 1000 folders in a flat structure spawns 1000 concurrent tasks, each making DB queries. - -### Impact - -- **Severity**: HIGH — DB connection pool exhaustion (max 20 connections), Tokio task backlog. -- Contrast with `batch_operations.rs` which correctly uses `Semaphore::new(10)`. - -### Fix Sketch - -```rust -use tokio::sync::Semaphore; - -fn search_parallel( - semaphore: Arc, - // ... other args -) { - Box::pin(async move { - let _permit = semaphore.acquire().await.unwrap(); - // ... existing logic, pass semaphore to recursive calls - }) -} -``` - -Or better: for non-recursive search, the DB-level pagination path is already used. For recursive search, consider a single recursive SQL CTE instead of application-level recursion. - ---- - -## 6. HIGH — Thumbnail Cache Write-Lock Contention on Reads - -### Location - -| File | Line(s) | Symbol | -|------|---------|--------| -| `src/infrastructure/services/thumbnail_service.rs` | ~25, 200–280 | `cache: Arc>>` | - -### Problematic Pattern - -```rust -// LruCache requires write access on EVERY read (LRU promotion) -pub async fn get_thumbnail(&self, …) -> Result { - // Read from cache — but LRU needs write lock! - let cache = self.cache.read().await; // Can't actually use read lock for LRU - // ... -} - -pub async fn add_to_cache(&self, key: ThumbnailCacheKey, data: Bytes) { - let mut current_size = self.current_cache_bytes.write().await; // Lock #1 - // ... eviction loop also acquires: - let mut cache = self.cache.write().await; // Lock #2 - // TWO write locks held simultaneously -} -``` - -Every cache hit and miss requires a write lock. Under concurrent image requests, this creates a bottleneck. - -### Impact - -- **Severity**: HIGH for image-heavy workloads. -- Two separate `RwLock` acquisitions in `add_to_cache()` — potential for deadlock if ordering is inconsistent. - -### Fix Sketch - -Replace with `moka::future::Cache` (already used in `image_transcode_service.rs`): - -```rust -use moka::future::Cache; - -pub struct ThumbnailService { - cache: Cache, // Lock-free reads, weight-based eviction - // Remove current_cache_bytes — moka tracks weight internally -} - -// Construction -let cache = Cache::builder() - .max_capacity(max_cache_bytes as u64) - .weigher(|_k: &ThumbnailCacheKey, v: &Bytes| v.len() as u32) - .time_to_idle(Duration::from_secs(300)) - .build(); -``` - ---- - -## 7. HIGH — HTTP Cache Middleware Buffers Entire Response Bodies - -### Location - -| File | Line(s) | Symbol | -|------|---------|--------| -| `src/interfaces/middleware/cache.rs` | 247–259, 450–470 | `cache_middleware()`, `HttpCacheService::call()` | - -### Problematic Pattern - -```rust -// cache.rs: cache_middleware() -let bytes = axum::body::to_bytes(_body, 1024 * 1024 * 10) // Buffer up to 10MB - .await - .unwrap_or_default(); - -let etag = cache.calculate_etag_for_bytes(&bytes); // Hash all bytes -cache.set(cache_key, etag, Some(bytes.clone()), …); // Clone + store -``` - -Every non-cached GET response is fully buffered to calculate an ETag, even for responses that shouldn't be cached (large file listings, etc.). - -### Impact - -- **Severity**: HIGH — 10MB max buffer per concurrent request × N concurrent requests. -- The `bytes.clone()` doubles peak memory per response. - -### Fix Sketch - -1. Only cache small responses (check `Content-Length` first). -2. Use streaming hash (SHA-256) to compute ETag without buffering. -3. Skip caching for responses > 1MB. -4. Replace `std::sync::Mutex` backing the cache (see Issue #1). - ---- - -## 8. MEDIUM — N+1 Queries / Extra Database Round Trips - -### Location - -| File | Line(s) | Symbol | Issue | -|------|---------|--------|-------| -| `src/infrastructure/repositories/pg/folder_db_repository.rs` | ~rename_folder | `rename_folder()` | UPDATE + separate SELECT | -| `src/infrastructure/repositories/pg/folder_db_repository.rs` | ~move_folder | `move_folder()` | UPDATE + separate SELECT | -| `src/infrastructure/repositories/pg/file_blob_write_repository.rs` | ~lookup_folder_path | `lookup_folder_path()` | Extra query per file write | -| `src/infrastructure/repositories/pg/trash_db_repository.rs` | ~clear_trash | `clear_trash()` | 2 separate DELETEs | - -### Problematic Pattern - -```rust -// folder_db_repository.rs: rename_folder() -async fn rename_folder(&self, id: &str, new_name: &str) -> Result { - // Query 1: UPDATE - sqlx::query("UPDATE storage.folders SET name = $1 WHERE id = $2::uuid") - .execute(self.pool.as_ref()).await?; - - // Query 2: SELECT (separate round trip) - self.get_folder(id).await -} -``` - -### Impact - -- **Severity**: MEDIUM — adds 1–5ms per extra round trip, compounded in batch operations. -- `lookup_folder_path()` is called per file write; in batch uploads of N files to the same folder, it makes N identical queries. - -### Fix Sketch - -```sql --- Use RETURNING to get the updated row in a single query -UPDATE storage.folders SET name = $1 -WHERE id = $2::uuid -RETURNING id::text, name, parent_id::text, path, … -``` - -For `lookup_folder_path()` in batch operations, cache the folder path for the duration of the batch. - ---- - -## 9. MEDIUM — Unnecessary String Allocations in Error Paths - -### Location - -| File | Line(s) | Symbol | -|------|---------|--------| -| `src/domain/errors.rs` | throughout | `DomainError` factory methods | -| `src/infrastructure/repositories/pg/*.rs` | throughout | `.map_err(|e| DomainError::internal_error(…, format!("…: {e}")))` | - -### Problematic Pattern - -```rust -// domain/errors.rs -pub fn not_found(entity_type: &'static str, id: impl Into) -> Self { - let entity_id = id.into(); - Self { - message: format!("{} not found: {}", entity_type, entity_id), // ALLOCATION - entity_id: Some(entity_id), // ALLOCATION - // ... - } -} -``` - -Every error — even `NotFound` which may be a normal control flow path (e.g., checking if a file exists) — allocates 2 strings via `format!()` and `into()`. - -### Impact - -- **Severity**: MEDIUM — hot error paths (404 checks, duplicate detection) trigger allocations. -- In batch operations checking 1000 files, this creates thousands of unnecessary allocations. - -### Fix Sketch - -Use `Cow<'static, str>` for common messages: -```rust -pub fn not_found(entity_type: &'static str, id: impl Into) -> Self { - Self { - message: Cow::Borrowed(""), // Defer formatting to Display impl - entity_id: Some(id.into()), - kind: ErrorKind::NotFound, - entity_type, - source: None, - } -} - -impl fmt::Display for DomainError { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - // Format lazily only when displayed - write!(f, "{} {}: {}", self.entity_type, self.kind, - self.entity_id.as_deref().unwrap_or("unknown")) - } -} -``` - ---- - -## 10. MEDIUM — Redundant Path String in Domain Entities - -### Location - -| File | Line(s) | Symbol | -|------|---------|--------| -| `src/domain/entities/file.rs` | ~30–50 | `storage_path: StoragePath` + `path_string: String` | -| `src/domain/entities/folder.rs` | ~30–50 | Same pattern | - -### Problematic Pattern - -```rust -pub struct File { - storage_path: StoragePath, - path_string: String, // Redundant: same data as storage_path.to_string() - // ... -} -``` - -Every `File` and `Folder` entity carries both a `StoragePath` (which internally holds `Vec`) **and** a pre-rendered `String` copy. This doubles the path memory per entity. - -### Impact - -- **Severity**: MEDIUM — when listing 10,000 files, each path is stored twice. - -### Fix Sketch - -Remove `path_string` and derive it on demand: -```rust -impl File { - pub fn path_string(&self) -> String { - self.storage_path.to_string() - } -} -``` - -Or cache it lazily with `OnceCell` if `.to_string()` is called frequently. - ---- - -## 11. MEDIUM — Unbounded Parallel Tasks in Storage Usage Update - -### Location - -| File | Line(s) | Symbol | -|------|---------|--------| -| `src/application/services/storage_usage_service.rs` | 138–165 | `update_all_users_storage_usage()` | - -### Problematic Pattern - -```rust -async fn update_all_users_storage_usage(&self) -> Result<(), DomainError> { - let users = self.user_repository.list_users(1000, 0).await?; - - let mut update_tasks = Vec::new(); - for user in users { - let service_clone = self.clone(); - // Spawn one task per user — NO concurrency limit - let task = task::spawn(async move { - service_clone.update_user_storage_usage(&user_id).await - }); - update_tasks.push(task); - } - // joins all -} -``` - -### Impact - -- **Severity**: MEDIUM — 1000 users = 1000 concurrent DB queries. DB pool has max 20 connections, so 980 tasks queue, but Tokio task overhead + connection wait time is wasteful. - -### Fix Sketch - -```rust -use futures::stream::{self, StreamExt}; - -stream::iter(users) - .map(|user| { - let svc = self.clone(); - async move { svc.update_user_storage_usage(&user.id).await } - }) - .buffer_unordered(10) // Max 10 concurrent - .collect::>() - .await; -``` - ---- - -## 12. MEDIUM — Upload Handler Re-parses Its Own HTTP Response - -### Location - -| File | Line(s) | Symbol | -|------|---------|--------| -| `src/interfaces/api/handlers/file_handler.rs` | ~upload_file_with_thumbnails | `upload_file_with_thumbnails()` | - -### Problematic Pattern - -The `upload_file_with_thumbnails` handler calls the upload logic, gets back an HTTP response, then reads the response body back to extract the file ID for thumbnail generation. This means: - -1. Serialize file info → JSON response body -2. Read response body → bytes -3. Deserialize bytes → file info -4. Use file info for thumbnail generation - -### Impact - -- **Severity**: MEDIUM — unnecessary serialize → deserialize round trip per upload. - -### Fix Sketch - -Call the upload service directly and pass the result to thumbnail generation, instead of going through HTTP serialization: - -```rust -let file = upload_service.upload_file(…).await?; -thumbnail_service.generate_all_sizes_background(file.id.clone(), path); -Ok(Json(FileDto::from(file))) -``` - ---- - -## 13. LOW — One-Shot Cache Pattern Defeats Caching Purpose - -### Location - -| File | Line(s) | Symbol | -|------|---------|--------| -| `src/infrastructure/repositories/pg/file_blob_read_repository.rs` | ~95–115 | `resolve_blob_hash()` | - -### Problematic Pattern - -```rust -async fn resolve_blob_hash(&self, file_id: &str) -> Result { - // Check moka cache - if let Some(hash) = self.hash_cache.get(file_id) { - self.hash_cache.invalidate(file_id); // Immediately invalidate! - return Ok(hash); - } - // ... DB query -} -``` - -The hash is cached then immediately invalidated after first use. This means repeated reads of the same file always hit the database. - -### Impact - -- **Severity**: LOW — the pattern only provides "write-behind" benefit (avoiding a DB query between upload and first download). Repeated downloads bypass cache. - -### Fix Sketch - -Remove the invalidation and let moka's TTI (30s) handle expiry: -```rust -if let Some(hash) = self.hash_cache.get(file_id) { - return Ok(hash); // Let TTI handle expiry -} -``` - ---- - -## 14. LOW — Duplicated SQL in Paginated Search - -### Location - -| File | Line(s) | Symbol | -|------|---------|--------| -| `src/infrastructure/repositories/pg/file_blob_read_repository.rs` | 400–620 | `search_files_paginated()` | - -### Problematic Pattern - -Four nearly identical `match` arms containing: -- Copy-pasted SQL with minor WHERE clause differences -- Each arm has a COUNT query + SELECT query (2 DB round trips per search) -- SQL ORDER BY built via `format!()` string interpolation - -### Impact - -- **Severity**: LOW (correctness) to MEDIUM (maintenance burden). -- The COUNT query is always executed even when the result set is smaller than the limit (i.e., total count could be inferred). - -### Fix Sketch - -Build the query dynamically with a query builder: -```rust -let mut conditions = vec!["fi.user_id = $1::uuid", "fi.is_trashed = false"]; -let mut bind_idx = 2; - -if let Some(fid) = folder_id { - conditions.push(&format!("fi.folder_id = ${bind_idx}::uuid")); - bind_idx += 1; -} -if let Some(name) = &criteria.name_contains { - conditions.push(&format!("LOWER(fi.name) LIKE ${bind_idx}")); - bind_idx += 1; -} -// ... single query construction -``` - -Use `COUNT(*) OVER()` window function to get total count in a single query: -```sql -SELECT fi.*, COUNT(*) OVER() as total_count -FROM storage.files fi -WHERE … -ORDER BY … LIMIT $N OFFSET $M -``` - ---- - -## 15. LOW — Sequential Trash Cleanup Without Batching - -### Location - -| File | Line(s) | Symbol | -|------|---------|--------| -| `src/infrastructure/services/trash_cleanup_service.rs` | 75–95 | `cleanup_expired_items()` | - -### Problematic Pattern - -```rust -for item in expired_items { - trash_service.delete_permanently(&trash_id, &user_id).await; // One at a time -} -``` - -### Impact - -- **Severity**: LOW — cleanup runs periodically in the background, not in the request path. - -### Fix Sketch - -Use `futures::stream::buffer_unordered()` for concurrent deletion, or batch-delete at the SQL level: -```sql -DELETE FROM storage.files WHERE is_trashed = true AND trashed_at < NOW() - INTERVAL '30 days'; -``` - ---- - -## 16. LOW — Search Cache Key Serializes Entire DTO to JSON - -### Location - -| File | Line(s) | Symbol | -|------|---------|--------| -| `src/application/services/search_service.rs` | 170–180 | `create_cache_key()` | - -### Problematic Pattern - -```rust -fn create_cache_key(&self, criteria: &SearchCriteriaDto, user_id: &str) -> Result { - let criteria_str = serde_json::to_string(criteria).map_err(…)?; // Full JSON serialization - Ok(SearchCacheKey { - criteria_hash: criteria_str, // Stored as full JSON string, not a hash - user_id: user_id.to_string(), - }) -} -``` - -The "hash" field is actually the full JSON string, not a hash. This means: -1. Full serde serialization per search request -2. HashMap key comparison is O(n) on string length -3. Unnecessary memory for cache keys - -### Impact - -- **Severity**: LOW — search requests are human-speed, not high-throughput. - -### Fix Sketch - -```rust -use std::hash::{Hash, Hasher, DefaultHasher}; - -fn create_cache_key(&self, criteria: &SearchCriteriaDto, user_id: &str) -> SearchCacheKey { - let mut hasher = DefaultHasher::new(); - criteria.hash(&mut hasher); // Derive Hash on SearchCriteriaDto - user_id.hash(&mut hasher); - SearchCacheKey(hasher.finish()) -} -``` - ---- - -## 17. Positive Patterns — What's Done Well - -These are worth calling out as **exemplary** implementations: - -| Component | File | Pattern | -|-----------|------|---------| -| **Image Transcoding** | `image_transcode_service.rs` | Dedicated `rayon` thread pool (not Tokio blocking pool), `moka` lock-free cache, `AtomicU64` stats, fire-and-forget disk cache writes. **Best-in-class design.** | -| **File Content Cache** | `file_content_cache.rs` | `moka` weight-based cache, lock-free reads, automatic eviction. Clean. | -| **Batch Operations** | `batch_operations.rs` | `Semaphore`-based concurrency control. Correct pattern. | -| **Thumbnail Generation** | `thumbnail_service.rs` | Uses `spawn_blocking` for image processing. Correct (but cache should be moka). | -| **Compression** | `compression_service.rs` | `spawn_blocking` for CPU-bound gzip. Streaming compress. Correct. | -| **Dedup Service** | `dedup_service.rs` | Atomic write via temp+rename, `SELECT FOR UPDATE` for blob refcounting, 2-char hash prefix sharding. Solid CAS implementation. | -| **Multi-Tier Download** | `file_retrieval_service.rs` | Write-behind → hot cache + WebP → mmap → streaming. Well-designed tiered strategy. | -| **Streaming Upload** | `file_handler.rs` | SHA-256 computed during spool, 512KB BufWriter, pre-allocation hints. Good. | -| **DB Pagination** | `file_blob_read_repository.rs` | Non-recursive search uses LIMIT/OFFSET at DB level. Correct. | -| **Content Dedup** | `file_blob_write_repository.rs` | `copy_file` uses CTE for zero-copy blob dedup. `update_file_content` uses atomic CTE with `FOR UPDATE`. | - ---- - -## 18. Summary Matrix - -| # | Issue | Severity | Impact Area | Effort to Fix | -|---|-------|----------|-------------|---------------| -| 1 | `std::sync::Mutex` in async | **CRITICAL** | Latency, deadlock | Small (swap to moka) | -| 2 | ZIP loads files into memory | **CRITICAL** | OOM risk | Medium (streaming ZIP) | -| 3 | Share repo: JSON file I/O | **CRITICAL** | Data loss, O(n) | Medium (migrate to PG) | -| 4 | Blocking FS in async | **HIGH** | Latency on slow storage | Small (use tokio::fs) | -| 5 | Unbounded search tasks | **HIGH** | DB pool exhaustion | Small (add Semaphore) | -| 6 | Thumbnail cache write-lock | **HIGH** | Contention | Small (swap to moka) | -| 7 | HTTP cache buffers 10MB | **HIGH** | Memory | Medium (streaming hash) | -| 8 | N+1 queries | **MEDIUM** | Latency | Small (use RETURNING) | -| 9 | Error string allocations | **MEDIUM** | Allocator pressure | Medium (Cow/lazy) | -| 10 | Redundant path string | **MEDIUM** | Memory | Small (remove field) | -| 11 | Unbounded storage tasks | **MEDIUM** | DB pool | Small (add Semaphore) | -| 12 | Handler re-parses response | **MEDIUM** | CPU waste | Small (refactor) | -| 13 | One-shot cache invalidation | **LOW** | Cache miss rate | Trivial | -| 14 | Duplicated search SQL | **LOW** | Maintenance | Medium | -| 15 | Sequential trash cleanup | **LOW** | Cleanup speed | Small | -| 16 | JSON cache key | **LOW** | Minor alloc | Small | - -### Recommended Priority Order - -1. **Issues 1, 2, 3** — Fix immediately. These can cause production incidents (deadlocks, OOM, data loss). -2. **Issues 4, 5, 6** — Fix before scaling. These create bottlenecks under load. -3. **Issue 7** — Fix when observing memory pressure. -4. **Issues 8–12** — Address as part of normal development. -5. **Issues 13–16** — Clean up opportunistically. diff --git a/src/application/mod.rs b/src/application/mod.rs index 86af78c8..089b4fbe 100644 --- a/src/application/mod.rs +++ b/src/application/mod.rs @@ -2,6 +2,5 @@ pub mod adapters; pub mod dtos; pub mod ports; pub mod services; -pub mod transactions; // Re-exportaciones para facilitar el acceso a los principales puertos diff --git a/src/application/services/batch_operations.rs b/src/application/services/batch_operations.rs index 590e1087..571e2cb0 100644 --- a/src/application/services/batch_operations.rs +++ b/src/application/services/batch_operations.rs @@ -1,13 +1,12 @@ use async_zip::base::write::ZipFileWriter; use async_zip::{Compression, ZipEntryBuilder}; use futures::io::AsyncWriteExt as FuturesWriteExt; -use futures::{Future, StreamExt, future::join_all}; +use futures::{Future, StreamExt, stream}; use std::collections::HashMap; use std::sync::Arc; use tempfile::NamedTempFile; use thiserror::Error; use tokio::io::BufWriter; -use tokio::sync::Semaphore; use tracing::info; use crate::application::dtos::file_dto::FileDto; @@ -71,7 +70,6 @@ pub struct BatchOperationService { folder_service: Arc, trash_service: Option>, config: AppConfig, - semaphore: Arc, } impl BatchOperationService { @@ -82,16 +80,12 @@ impl BatchOperationService { folder_service: Arc, config: AppConfig, ) -> Self { - // Limit concurrency based on configuration - let max_concurrency = config.concurrency.max_concurrent_files; - Self { file_retrieval, file_management, folder_service, trash_service: None, config, - semaphore: Arc::new(Semaphore::new(max_concurrency)), } } @@ -123,6 +117,7 @@ impl BatchOperationService { ) -> Result, BatchOperationError> { info!("Starting batch copy of {} files", file_ids.len()); let start_time = std::time::Instant::now(); + let max_concurrent = self.config.concurrency.max_concurrent_files; // Create result structure let mut result = BatchResult { @@ -134,31 +129,23 @@ impl BatchOperationService { }, }; - // Define the operation to perform for each file - let operations = file_ids.into_iter().map(|file_id| { + // Arc avoids N heap-clones of the same string + let target_folder: Option> = target_folder_id.map(|s| Arc::from(s.as_str())); + + // buffer_unordered materialises only max_concurrent futures at a time + let mut operation_stream = stream::iter(file_ids.into_iter().map(|file_id| { let mgmt = self.file_management.clone(); - let target_folder = target_folder_id.clone(); - let semaphore = self.semaphore.clone(); + let target_folder = target_folder.clone(); async move { - // Acquire semaphore permit - let permit = semaphore.acquire().await.unwrap(); - - let copy_result = mgmt.copy_file(&file_id, target_folder.clone()).await; - - // Release the permit explicitly (also released on drop) - drop(permit); - - // Return the result along with the ID to identify successes/failures + let copy_result = mgmt.copy_file(&file_id, target_folder.map(|s| s.to_string())).await; (file_id, copy_result) } - }); + })) + .buffer_unordered(max_concurrent); - // Execute all operations in parallel with concurrency control - let operation_results = join_all(operations).await; - - // Process the results - for (file_id, operation_result) in operation_results { + // Process results as they complete + while let Some((file_id, operation_result)) = operation_stream.next().await { match operation_result { Ok(file) => { result.successful.push(file); @@ -195,6 +182,7 @@ impl BatchOperationService { ) -> Result, BatchOperationError> { info!("Starting batch move of {} files", file_ids.len()); let start_time = std::time::Instant::now(); + let max_concurrent = self.config.concurrency.max_concurrent_files; // Create result structure let mut result = BatchResult { @@ -206,31 +194,20 @@ impl BatchOperationService { }, }; - // Define the operation to perform for each file - let operations = file_ids.into_iter().map(|file_id| { + let target_folder: Option> = target_folder_id.map(|s| Arc::from(s.as_str())); + + let mut operation_stream = stream::iter(file_ids.into_iter().map(|file_id| { let mgmt = self.file_management.clone(); - let target_folder = target_folder_id.clone(); - let semaphore = self.semaphore.clone(); + let target_folder = target_folder.clone(); async move { - // Acquire semaphore permit - let permit = semaphore.acquire().await.unwrap(); - - let move_result = mgmt.move_file(&file_id, target_folder.clone()).await; - - // Release the permit explicitly - drop(permit); - - // Return the result along with the ID to identify successes/failures + let move_result = mgmt.move_file(&file_id, target_folder.map(|s| s.to_string())).await; (file_id, move_result) } - }); + })) + .buffer_unordered(max_concurrent); - // Execute all operations in parallel with concurrency control - let operation_results = join_all(operations).await; - - // Process the results - for (file_id, operation_result) in operation_results { + while let Some((file_id, operation_result)) = operation_stream.next().await { match operation_result { Ok(file) => { result.successful.push(file); @@ -278,30 +255,19 @@ impl BatchOperationService { }; // Define the operation to perform for each file - let operations = file_ids.into_iter().map(|file_id| { + let mut operation_stream = stream::iter(file_ids.into_iter().map(|file_id| { let mgmt = self.file_management.clone(); - let semaphore = self.semaphore.clone(); - let id_clone = file_id.clone(); async move { - // Acquire semaphore permit - let permit = semaphore.acquire().await.unwrap(); - let delete_result = mgmt.delete_file(&file_id).await; - - // Release the permit explicitly - drop(permit); - - // Return the result along with the ID - (id_clone.clone(), delete_result.map(|_| id_clone)) + let id_for_result = file_id.clone(); + (file_id, delete_result.map(|_| id_for_result)) } - }); + })) + .buffer_unordered(self.config.concurrency.max_concurrent_files); - // Execute all operations in parallel with concurrency control - let operation_results = join_all(operations).await; - - // Process the results - for (file_id, operation_result) in operation_results { + // Process results as they complete + while let Some((file_id, operation_result)) = operation_stream.next().await { match operation_result { Ok(id) => { result.successful.push(id); @@ -349,29 +315,18 @@ impl BatchOperationService { }; // Define the operation to perform for each file - let operations = file_ids.into_iter().map(|file_id| { + let mut operation_stream = stream::iter(file_ids.into_iter().map(|file_id| { let retrieval = self.file_retrieval.clone(); - let semaphore = self.semaphore.clone(); async move { - // Acquire semaphore permit - let permit = semaphore.acquire().await.unwrap(); - let get_result = retrieval.get_file(&file_id).await; - - // Release the permit explicitly - drop(permit); - - // Return the result along with the ID (file_id, get_result) } - }); + })) + .buffer_unordered(self.config.concurrency.max_concurrent_files); - // Execute all operations in parallel with concurrency control - let operation_results = join_all(operations).await; - - // Process the results - for (file_id, operation_result) in operation_results { + // Process results as they complete + while let Some((file_id, operation_result)) = operation_stream.next().await { match operation_result { Ok(file) => { result.successful.push(file); @@ -421,31 +376,22 @@ impl BatchOperationService { }; // Define the operation to perform for each folder - let operations = folder_ids.into_iter().map(|folder_id| { + // Arc avoids N heap-clones of the caller string + let caller: Arc = Arc::from(caller_id); + + let mut operation_stream = stream::iter(folder_ids.into_iter().map(|folder_id| { let folder_service = self.folder_service.clone(); - let semaphore = self.semaphore.clone(); - let id_clone = folder_id.clone(); - let caller = caller_id.to_string(); + let caller = caller.clone(); async move { - // Acquire semaphore permit - let permit = semaphore.acquire().await.unwrap(); - let delete_result = folder_service.delete_folder(&folder_id, &caller).await; - - // Release the permit explicitly - drop(permit); - - // Return the result along with the ID - (id_clone.clone(), delete_result.map(|_| id_clone)) + let id_for_result = folder_id.clone(); + (folder_id, delete_result.map(|_| id_for_result)) } - }); + })) + .buffer_unordered(self.config.concurrency.max_concurrent_files); - // Execute all operations in parallel with concurrency control - let operation_results = join_all(operations).await; - - // Process the results - for (folder_id, operation_result) in operation_results { + while let Some((folder_id, operation_result)) = operation_stream.next().await { match operation_result { Ok(id) => { result.successful.push(id); @@ -497,23 +443,21 @@ impl BatchOperationService { }, }; - let operations = file_ids.into_iter().map(|file_id| { + let uid: Arc = Arc::from(user_id); + + let mut operation_stream = stream::iter(file_ids.into_iter().map(|file_id| { let trash = trash_service.clone(); - let semaphore = self.semaphore.clone(); - let uid = user_id.to_string(); - let id_clone = file_id.clone(); + let uid = uid.clone(); async move { - let permit = semaphore.acquire().await.unwrap(); let trash_result = trash.move_to_trash(&file_id, "file", &uid).await; - drop(permit); - (id_clone.clone(), trash_result.map(|_| id_clone)) + let id_for_result = file_id.clone(); + (file_id, trash_result.map(|_| id_for_result)) } - }); + })) + .buffer_unordered(self.config.concurrency.max_concurrent_files); - let operation_results = join_all(operations).await; - - for (file_id, operation_result) in operation_results { + while let Some((file_id, operation_result)) = operation_stream.next().await { match operation_result { Ok(id) => { result.successful.push(id); @@ -564,23 +508,21 @@ impl BatchOperationService { }, }; - let operations = folder_ids.into_iter().map(|folder_id| { + let uid: Arc = Arc::from(user_id); + + let mut operation_stream = stream::iter(folder_ids.into_iter().map(|folder_id| { let trash = trash_service.clone(); - let semaphore = self.semaphore.clone(); - let uid = user_id.to_string(); - let id_clone = folder_id.clone(); + let uid = uid.clone(); async move { - let permit = semaphore.acquire().await.unwrap(); let trash_result = trash.move_to_trash(&folder_id, "folder", &uid).await; - drop(permit); - (id_clone.clone(), trash_result.map(|_| id_clone)) + let id_for_result = folder_id.clone(); + (folder_id, trash_result.map(|_| id_for_result)) } - }); + })) + .buffer_unordered(self.config.concurrency.max_concurrent_files); - let operation_results = join_all(operations).await; - - for (folder_id, operation_result) in operation_results { + while let Some((folder_id, operation_result)) = operation_stream.next().await { match operation_result { Ok(id) => { result.successful.push(id); @@ -627,24 +569,23 @@ impl BatchOperationService { }, }; - let operations = folder_ids.into_iter().map(|folder_id| { + let target: Option> = target_folder_id.map(|s| Arc::from(s.as_str())); + let caller: Arc = Arc::from(caller_id); + + let mut operation_stream = stream::iter(folder_ids.into_iter().map(|folder_id| { let folder_service = self.folder_service.clone(); - let target = target_folder_id.clone(); - let semaphore = self.semaphore.clone(); - let caller = caller_id.to_string(); + let target = target.clone(); + let caller = caller.clone(); async move { - let permit = semaphore.acquire().await.unwrap(); - let dto = MoveFolderDto { parent_id: target }; + let dto = MoveFolderDto { parent_id: target.map(|s| s.to_string()) }; let move_result = folder_service.move_folder(&folder_id, dto, &caller).await; - drop(permit); (folder_id, move_result) } - }); + })) + .buffer_unordered(self.config.concurrency.max_concurrent_files); - let operation_results = join_all(operations).await; - - for (folder_id, operation_result) in operation_results { + while let Some((folder_id, operation_result)) = operation_stream.next().await { match operation_result { Ok(folder) => { result.successful.push(folder); @@ -873,7 +814,7 @@ impl BatchOperationService { ) -> Result, BatchOperationError> where T: Clone + Send + 'static + std::fmt::Debug, - F: Fn(T, Arc) -> Fut + Clone + Send + Sync + 'static, + F: Fn(T) -> Fut + Clone + Send + Sync + 'static, Fut: Future> + Send + 'static, { info!( @@ -892,33 +833,25 @@ impl BatchOperationService { }, }; - // Convert each item to a task - let tasks = items.iter().map(|item| { - let item_clone = item.clone(); + // buffer_unordered materialises only max_concurrent futures at a time + let mut operation_stream = stream::iter(items.into_iter().map(|item| { let op = operation.clone(); - let semaphore = self.semaphore.clone(); async move { - // The provided function must handle semaphore acquisition - let op_result = op(item_clone.clone(), semaphore).await; - - // Return the result along with the original item for identification - (item_clone, op_result) + let op_result = op(item.clone()).await; + (item, op_result) } - }); + })) + .buffer_unordered(self.config.concurrency.max_concurrent_files); - // Execute all tasks in parallel - let operation_results = join_all(tasks).await; - - // Process results - for (item, operation_result) in operation_results { + // Process results as they complete + while let Some((item, operation_result)) = operation_stream.next().await { match operation_result { Ok(result_item) => { result.successful.push(result_item); result.stats.successful += 1; } Err(e) => { - // Convert item to string for error reporting result.failed.push((format!("{:?}", item), e.to_string())); result.stats.failed += 1; } @@ -960,34 +893,23 @@ impl BatchOperationService { }; // Define the operation for each folder - let operations = folders.into_iter().map(|(name, parent_id)| { + let mut operation_stream = stream::iter(folders.into_iter().map(|(name, parent_id)| { let folder_service = self.folder_service.clone(); - let semaphore = self.semaphore.clone(); async move { - // Acquire semaphore permit - let permit = semaphore.acquire().await.unwrap(); - let dto = crate::application::dtos::folder_dto::CreateFolderDto { name: name.clone(), parent_id: parent_id.clone(), }; let create_result = folder_service.create_folder(dto).await; - - // Release the permit explicitly - drop(permit); - - // Return the result with an identifier for errors let id = format!("{}:{}", name, parent_id.unwrap_or_default()); (id, create_result) } - }); + })) + .buffer_unordered(self.config.concurrency.max_concurrent_files); - // Execute all operations in parallel - let operation_results = join_all(operations).await; - - // Process the results - for (id, operation_result) in operation_results { + // Process results as they complete + while let Some((id, operation_result)) = operation_stream.next().await { match operation_result { Ok(folder) => { result.successful.push(folder); @@ -1035,29 +957,18 @@ impl BatchOperationService { }; // Define the operation for each folder - let operations = folder_ids.into_iter().map(|folder_id| { + let mut operation_stream = stream::iter(folder_ids.into_iter().map(|folder_id| { let folder_service = self.folder_service.clone(); - let semaphore = self.semaphore.clone(); async move { - // Acquire semaphore permit - let permit = semaphore.acquire().await.unwrap(); - let get_result = folder_service.get_folder(&folder_id).await; - - // Release the permit explicitly - drop(permit); - - // Return the result with its ID (folder_id, get_result) } - }); + })) + .buffer_unordered(self.config.concurrency.max_concurrent_files); - // Execute all operations in parallel - let operation_results = join_all(operations).await; - - // Process the results - for (folder_id, operation_result) in operation_results { + // Process results as they complete + while let Some((folder_id, operation_result)) = operation_stream.next().await { match operation_result { Ok(folder) => { result.successful.push(folder); @@ -1105,11 +1016,8 @@ mod tests { AppConfig::default(), ); - // Define a generic test operation - let operation = |item: i32, semaphore: Arc| async move { - // Acquire and release the semaphore - let _permit = semaphore.acquire().await.unwrap(); - + // Define a generic test operation (no more semaphore parameter) + let operation = |item: i32| async move { if item % 2 == 0 { // Simulate success for even numbers Ok(item * 2) diff --git a/src/application/services/folder_service.rs b/src/application/services/folder_service.rs index a695a795..a90be389 100644 --- a/src/application/services/folder_service.rs +++ b/src/application/services/folder_service.rs @@ -3,7 +3,6 @@ use crate::application::dtos::folder_dto::{ }; use crate::application::ports::inbound::FolderUseCase; use crate::application::ports::outbound::FolderStoragePort; -use crate::application::transactions::storage_transaction::StorageTransaction; use crate::common::errors::{DomainError, ErrorKind}; use crate::domain::services::path_service::StoragePath; use async_trait::async_trait; @@ -398,52 +397,11 @@ impl FolderUseCase for FolderService { return Err(DomainError::not_found("Folder", id)); } - // Create transaction for renaming - let mut transaction = StorageTransaction::new("rename_folder"); - - // Main operation: rename folder - // Clone all values to avoid lifetime issues - let folder_storage = self.folder_storage.clone(); - let id_owned = id.to_string(); - let name_owned = dto.name.clone(); - - // Create future with owned values - let rename_op = async move { - folder_storage.rename_folder(&id_owned, name_owned).await?; - Ok(()) - }; - let rollback_op = { - let original_name = existing_folder.name().to_string(); - let storage = self.folder_storage.clone(); - let id_clone = id.to_string(); - - async move { - // In case of failure, restore the original name - storage - .rename_folder(&id_clone, original_name) - .await - .map(|_| ()) - .map_err(|e| { - DomainError::new( - ErrorKind::InternalError, - "Folder", - format!("Failed to rollback folder rename: {}", e), - ) - }) - } - }; - - // Add to the transaction - transaction.add_operation(rename_op, rollback_op); - - // Execute transaction - transaction.commit().await?; - - // Get the renamed folder - let folder = self.folder_storage.get_folder(id).await.map_err(|e| { + // Rename folder — UPDATE RETURNING gives us the updated row directly + let folder = self.folder_storage.rename_folder(id, dto.name).await.map_err(|e| { DomainError::internal_error( "FolderStorage", - format!("Failed to get renamed folder with ID: {}: {}", id, e), + format!("Failed to rename folder with ID: {}: {}", id, e), ) })?; @@ -495,55 +453,12 @@ impl FolderUseCase for FolderService { // TODO: Ideally we should verify the entire hierarchy to prevent cycles } - // Create transaction for moving - let mut transaction = StorageTransaction::new("move_folder"); - - // Main operation: move folder - // Clone all values to avoid lifetime issues - let folder_storage = self.folder_storage.clone(); - let id_owned = id.to_string(); - // Get parent ID as owned string or None - let parent_id_owned = dto.parent_id.as_ref().map(|p| p.to_string()); - - // Create future with owned values - let move_op = async move { - // Convert Option to Option<&str> - let parent_ref = parent_id_owned.as_deref(); - folder_storage.move_folder(&id_owned, parent_ref).await?; - Ok(()) - }; - let rollback_op = { - let original_parent_id = source_folder.parent_id().map(String::from); - let storage = self.folder_storage.clone(); - let id_clone = id.to_string(); - - async move { - // In case of failure, restore the original location - storage - .move_folder(&id_clone, original_parent_id.as_deref()) - .await - .map(|_| ()) - .map_err(|e| { - DomainError::new( - ErrorKind::InternalError, - "Folder", - format!("Failed to rollback folder move: {}", e), - ) - }) - } - }; - - // Add to the transaction - transaction.add_operation(move_op, rollback_op); - - // Execute transaction - transaction.commit().await?; - - // Get the moved folder - let folder = self.folder_storage.get_folder(id).await.map_err(|e| { + // Move folder — UPDATE RETURNING gives us the updated row directly + let parent_ref = dto.parent_id.as_deref(); + let folder = self.folder_storage.move_folder(id, parent_ref).await.map_err(|e| { DomainError::internal_error( "FolderStorage", - format!("Failed to get moved folder with ID: {}: {}", id, e), + format!("Failed to move folder with ID: {}: {}", id, e), ) })?; diff --git a/src/application/services/search_service.rs b/src/application/services/search_service.rs index d77ec418..0559b774 100644 --- a/src/application/services/search_service.rs +++ b/src/application/services/search_service.rs @@ -1,4 +1,5 @@ use async_trait::async_trait; +use std::cmp::Reverse; use std::sync::Arc; use std::time::{Duration, Instant}; @@ -323,15 +324,13 @@ impl SearchUseCase for SearchService { .map(|f| Self::enrich_folder(f, query)) .collect(); - // Sort folders + // Sort folders (cached_key avoids O(N log N) temporary String allocations) match criteria.sort_by.as_str() { "name" => { - enriched_folders - .sort_by(|a, b| a.name.to_lowercase().cmp(&b.name.to_lowercase())); + enriched_folders.sort_by_cached_key(|f| f.name.to_lowercase()); } "name_desc" => { - enriched_folders - .sort_by(|a, b| b.name.to_lowercase().cmp(&a.name.to_lowercase())); + enriched_folders.sort_by_cached_key(|f| Reverse(f.name.to_lowercase())); } "date" => { enriched_folders.sort_by(|a, b| a.modified_at.cmp(&b.modified_at)); @@ -411,13 +410,13 @@ impl SearchUseCase for SearchService { .map(|f| Self::enrich_folder(f, query)) .collect(); - // ── Sort folders (files already sorted by SQL ORDER BY) ── + // ── Sort folders (cached_key avoids O(N log N) temporary String allocations) ── match criteria.sort_by.as_str() { "name" => { - enriched_folders.sort_by(|a, b| a.name.to_lowercase().cmp(&b.name.to_lowercase())); + enriched_folders.sort_by_cached_key(|f| f.name.to_lowercase()); } "name_desc" => { - enriched_folders.sort_by(|a, b| b.name.to_lowercase().cmp(&a.name.to_lowercase())); + enriched_folders.sort_by_cached_key(|f| Reverse(f.name.to_lowercase())); } "date" => { enriched_folders.sort_by(|a, b| a.modified_at.cmp(&b.modified_at)); diff --git a/src/application/transactions/mod.rs b/src/application/transactions/mod.rs deleted file mode 100644 index f9f4dff9..00000000 --- a/src/application/transactions/mod.rs +++ /dev/null @@ -1 +0,0 @@ -pub mod storage_transaction; diff --git a/src/application/transactions/storage_transaction.rs b/src/application/transactions/storage_transaction.rs deleted file mode 100644 index d9aef65b..00000000 --- a/src/application/transactions/storage_transaction.rs +++ /dev/null @@ -1,150 +0,0 @@ -use crate::common::errors::{DomainError, ErrorKind}; -use std::future::Future; -use std::pin::Pin; - -/// Type for async operations and rollbacks -type TransactionOp = Pin> + Send>>; - -/// Transaction for storage operations -/// Allows defining a set of operations and their corresponding rollbacks -pub struct StorageTransaction { - /// Operations to execute - operations: Vec TransactionOp + Send>>, - /// Rollback operations to revert changes in case of error - rollbacks: Vec TransactionOp + Send>>, - /// Transaction name for logging - name: String, -} - -impl StorageTransaction { - /// Creates a new transaction - pub fn new(name: &str) -> Self { - Self { - operations: Vec::new(), - rollbacks: Vec::new(), - name: name.to_string(), - } - } - - /// Adds an operation to the transaction with its corresponding rollback - pub fn add_operation(&mut self, operation: F, rollback: R) - where - F: Future> + Send + 'static, - R: Future> + Send + 'static, - { - self.operations.push(Box::new(move || Box::pin(operation))); - self.rollbacks.push(Box::new(move || Box::pin(rollback))); - } - - /// Adds an operation without rollback (for cleanup or logging) - pub fn add_finalizer(&mut self, finalizer: F) - where - F: Future> + Send + 'static, - { - // The rollback is a no-op - let noop = async { Ok(()) }; - - self.operations.push(Box::new(move || Box::pin(finalizer))); - self.rollbacks.push(Box::new(move || Box::pin(noop))); - } - - /// Executes the transaction by applying all operations in order - /// If any fails, executes rollbacks in reverse order - pub async fn commit(mut self) -> Result<(), DomainError> { - tracing::debug!("Starting transaction: {}", self.name); - - let mut completed_ops = Vec::new(); - - // Extract operations to avoid ownership issues - let operations = std::mem::take(&mut self.operations); - let transaction_name = self.name.clone(); - - // Execute operations - for (i, op) in operations.into_iter().enumerate() { - match op().await { - Ok(()) => { - completed_ops.push(i); - tracing::trace!( - "Operation {} completed in transaction: {}", - i, - transaction_name - ); - } - Err(e) => { - tracing::error!( - "Error in operation {} of transaction {}: {}", - i, - transaction_name, - e - ); - - // Execute rollbacks for completed operations in reverse order - self.rollback(completed_ops).await?; - - return Err(DomainError::new( - ErrorKind::InternalError, - "Transaction", - format!("Transaction '{}' failed: {}", transaction_name, e), - ) - .with_source(e)); - } - } - } - - tracing::debug!("Transaction completed successfully: {}", transaction_name); - Ok(()) - } - - /// Executes rollbacks for completed operations - async fn rollback(mut self, completed_ops: Vec) -> Result<(), DomainError> { - tracing::warn!("Starting rollback for transaction: {}", self.name); - - let mut rollback_errors = Vec::new(); - - // Extract rollbacks to avoid ownership issues - let mut rollbacks = Vec::new(); - std::mem::swap(&mut rollbacks, &mut self.rollbacks); - - // Execute rollbacks in reverse order - for i in completed_ops.into_iter().rev() { - if i < rollbacks.len() { - // Take ownership of the rollback (get a mutable reference) - if let Some(rb) = rollbacks.get_mut(i) { - // Swap with an empty function - let rollback = std::mem::replace(rb, Box::new(|| Box::pin(async { Ok(()) }))); - if let Err(e) = rollback().await { - tracing::error!( - "Error in rollback of operation {} in transaction {}: {}", - i, - self.name, - e - ); - rollback_errors.push(e); - } - } - } - } - - // If there were errors during rollback, report them - if !rollback_errors.is_empty() { - tracing::error!( - "Errors during transaction rollback {}: {} errors", - self.name, - rollback_errors.len() - ); - - return Err(DomainError::new( - ErrorKind::InternalError, - "Transaction", - format!( - "Errors during transaction '{}' rollback: {} errors", - self.name, - rollback_errors.len() - ), - )); - } - - tracing::info!("Transaction rollback completed: {}", self.name); - Ok(()) - } -} diff --git a/src/infrastructure/repositories/pg/folder_db_repository.rs b/src/infrastructure/repositories/pg/folder_db_repository.rs index 7ff073a1..c25c8797 100644 --- a/src/infrastructure/repositories/pg/folder_db_repository.rs +++ b/src/infrastructure/repositories/pg/folder_db_repository.rs @@ -258,6 +258,9 @@ impl FolderRepository for FolderDbRepository { .collect() } + /// Paginated folder listing — single query with `COUNT(*) OVER()` window + /// function so the total matching count comes back alongside the data rows, + /// eliminating a separate COUNT round-trip. async fn list_folders_paginated( &self, parent_id: Option<&str>, @@ -265,34 +268,14 @@ impl FolderRepository for FolderDbRepository { limit: usize, include_total: bool, ) -> Result<(Vec, Option), DomainError> { - let total = if include_total { - let count: i64 = if let Some(pid) = parent_id { - sqlx::query_scalar( - "SELECT COUNT(*) FROM storage.folders WHERE parent_id = $1::uuid AND NOT is_trashed", - ) - .bind(pid) - .fetch_one(self.pool()) - .await - } else { - sqlx::query_scalar( - "SELECT COUNT(*) FROM storage.folders WHERE parent_id IS NULL AND NOT is_trashed", - ) - .fetch_one(self.pool()) - .await - } - .map_err(|e| DomainError::internal_error("FolderDb", format!("count: {e}")))?; - Some(count as usize) - } else { - None - }; - - let rows: Vec<(String, String, String, Option, String, i64, i64)> = + let rows: Vec<(String, String, String, Option, String, i64, i64, i64)> = if let Some(pid) = parent_id { sqlx::query_as( r#" SELECT id::text, name, path, parent_id::text, user_id, EXTRACT(EPOCH FROM created_at)::bigint, - EXTRACT(EPOCH FROM updated_at)::bigint + EXTRACT(EPOCH FROM updated_at)::bigint, + COUNT(*) OVER() AS total_count FROM storage.folders WHERE parent_id = $1::uuid AND NOT is_trashed ORDER BY name @@ -309,7 +292,8 @@ impl FolderRepository for FolderDbRepository { r#" SELECT id::text, name, path, parent_id::text, user_id, EXTRACT(EPOCH FROM created_at)::bigint, - EXTRACT(EPOCH FROM updated_at)::bigint + EXTRACT(EPOCH FROM updated_at)::bigint, + COUNT(*) OVER() AS total_count FROM storage.folders WHERE parent_id IS NULL AND NOT is_trashed ORDER BY name @@ -323,15 +307,24 @@ impl FolderRepository for FolderDbRepository { } .map_err(|e| DomainError::internal_error("FolderDb", format!("paginate: {e}")))?; + // total_count is identical in every row; 0 when the result set is empty. + let total = if include_total { + Some(rows.first().map_or(0, |r| r.7) as usize) + } else { + None + }; + let folders: Result, DomainError> = rows .into_iter() - .map(|(id, name, path, pid, uid, ca, ma)| { + .map(|(id, name, path, pid, uid, ca, ma, _total)| { Self::row_to_folder(id, name, path, pid, Some(uid), ca, ma) }) .collect(); Ok((folders?, total)) } + /// Paginated folder listing filtered by owner — single query with + /// `COUNT(*) OVER()` to avoid a separate COUNT round-trip. async fn list_folders_by_owner_paginated( &self, parent_id: Option<&str>, @@ -340,36 +333,14 @@ impl FolderRepository for FolderDbRepository { limit: usize, include_total: bool, ) -> Result<(Vec, Option), DomainError> { - let total = if include_total { - let count: i64 = if let Some(pid) = parent_id { - sqlx::query_scalar( - "SELECT COUNT(*) FROM storage.folders WHERE parent_id = $1::uuid AND user_id = $2 AND NOT is_trashed", - ) - .bind(pid) - .bind(owner_id) - .fetch_one(self.pool()) - .await - } else { - sqlx::query_scalar( - "SELECT COUNT(*) FROM storage.folders WHERE parent_id IS NULL AND user_id = $1 AND NOT is_trashed", - ) - .bind(owner_id) - .fetch_one(self.pool()) - .await - } - .map_err(|e| DomainError::internal_error("FolderDb", format!("count_by_owner: {e}")))?; - Some(count as usize) - } else { - None - }; - - let rows: Vec<(String, String, String, Option, String, i64, i64)> = + let rows: Vec<(String, String, String, Option, String, i64, i64, i64)> = if let Some(pid) = parent_id { sqlx::query_as( r#" SELECT id::text, name, path, parent_id::text, user_id, EXTRACT(EPOCH FROM created_at)::bigint, - EXTRACT(EPOCH FROM updated_at)::bigint + EXTRACT(EPOCH FROM updated_at)::bigint, + COUNT(*) OVER() AS total_count FROM storage.folders WHERE parent_id = $1::uuid AND user_id = $2 AND NOT is_trashed ORDER BY name @@ -387,7 +358,8 @@ impl FolderRepository for FolderDbRepository { r#" SELECT id::text, name, path, parent_id::text, user_id, EXTRACT(EPOCH FROM created_at)::bigint, - EXTRACT(EPOCH FROM updated_at)::bigint + EXTRACT(EPOCH FROM updated_at)::bigint, + COUNT(*) OVER() AS total_count FROM storage.folders WHERE parent_id IS NULL AND user_id = $1 AND NOT is_trashed ORDER BY name @@ -404,9 +376,15 @@ impl FolderRepository for FolderDbRepository { DomainError::internal_error("FolderDb", format!("paginate_by_owner: {e}")) })?; + let total = if include_total { + Some(rows.first().map_or(0, |r| r.7) as usize) + } else { + None + }; + let folders: Result, DomainError> = rows .into_iter() - .map(|(id, name, path, pid, uid, ca, ma)| { + .map(|(id, name, path, pid, uid, ca, ma, _total)| { Self::row_to_folder(id, name, path, pid, Some(uid), ca, ma) }) .collect(); @@ -417,16 +395,19 @@ impl FolderRepository for FolderDbRepository { // The BEFORE UPDATE trigger recomputes path/lpath for this row; // the AFTER UPDATE cascade trigger then batch-updates all // descendants in a single UPDATE using the GiST lpath index. - sqlx::query( + let row = sqlx::query_as::<_, (String, String, String, Option, String, i64, i64)>( r#" UPDATE storage.folders SET name = $1, updated_at = NOW() WHERE id = $2::uuid AND NOT is_trashed + RETURNING id::text, name, path, parent_id::text, user_id, + EXTRACT(EPOCH FROM created_at)::bigint, + EXTRACT(EPOCH FROM updated_at)::bigint "#, ) .bind(&new_name) .bind(id) - .execute(self.pool()) + .fetch_optional(self.pool()) .await .map_err(|e| { if let sqlx::Error::Database(ref db_err) = e @@ -435,9 +416,10 @@ impl FolderRepository for FolderDbRepository { return DomainError::already_exists("Folder", format!("{new_name} already exists")); } DomainError::internal_error("FolderDb", format!("rename: {e}")) - })?; + })? + .ok_or_else(|| DomainError::not_found("Folder", id))?; - self.get_folder(id).await + Self::row_to_folder(row.0, row.1, row.2, row.3, Some(row.4), row.5, row.6) } async fn move_folder( @@ -448,20 +430,24 @@ impl FolderRepository for FolderDbRepository { // The BEFORE UPDATE trigger recomputes path/lpath for this row; // the AFTER UPDATE cascade trigger then batch-updates all // descendants in a single UPDATE using the GiST lpath index. - sqlx::query( + let row = sqlx::query_as::<_, (String, String, String, Option, String, i64, i64)>( r#" UPDATE storage.folders SET parent_id = $1::uuid, updated_at = NOW() WHERE id = $2::uuid AND NOT is_trashed + RETURNING id::text, name, path, parent_id::text, user_id, + EXTRACT(EPOCH FROM created_at)::bigint, + EXTRACT(EPOCH FROM updated_at)::bigint "#, ) .bind(new_parent_id) .bind(id) - .execute(self.pool()) + .fetch_optional(self.pool()) .await - .map_err(|e| DomainError::internal_error("FolderDb", format!("move: {e}")))?; + .map_err(|e| DomainError::internal_error("FolderDb", format!("move: {e}")))? + .ok_or_else(|| DomainError::not_found("Folder", id))?; - self.get_folder(id).await + Self::row_to_folder(row.0, row.1, row.2, row.3, Some(row.4), row.5, row.6) } async fn delete_folder(&self, id: &str) -> Result<(), DomainError> { diff --git a/src/infrastructure/services/image_transcode_service.rs b/src/infrastructure/services/image_transcode_service.rs index 1e5f73ee..e781bc65 100644 --- a/src/infrastructure/services/image_transcode_service.rs +++ b/src/infrastructure/services/image_transcode_service.rs @@ -26,16 +26,28 @@ use crate::domain::errors::{DomainError, ErrorKind}; /// Maximum file size for transcoding (5MB - larger files stream directly) pub const MAX_TRANSCODE_SIZE: u64 = 5 * 1024 * 1024; -/// Number of threads in the dedicated transcoding pool -const TRANSCODE_POOL_THREADS: usize = 2; +/// Minimum number of threads in the dedicated transcoding pool +const MIN_TRANSCODE_THREADS: usize = 2; + +/// Compute the number of transcoding threads: half the available CPUs, +/// with a floor of `MIN_TRANSCODE_THREADS`. `available_parallelism()` +/// respects cgroup limits (Docker/K8s) and CPU affinity masks. +fn transcode_thread_count() -> usize { + let cpus = std::thread::available_parallelism() + .map(|n| n.get()) + .unwrap_or(MIN_TRANSCODE_THREADS); + (cpus / 2).max(MIN_TRANSCODE_THREADS) +} /// Dedicated rayon thread pool for CPU-bound image transcoding. /// Isolated from Tokio's blocking pool to prevent starvation of other I/O. +/// Thread count scales with available CPUs (half cores, min 2). fn transcode_pool() -> &'static rayon::ThreadPool { static POOL: OnceLock = OnceLock::new(); POOL.get_or_init(|| { + let threads = transcode_thread_count(); rayon::ThreadPoolBuilder::new() - .num_threads(TRANSCODE_POOL_THREADS) + .num_threads(threads) .thread_name(|idx| format!("transcode-{idx}")) .build() .expect("Failed to create transcode thread pool") @@ -171,7 +183,7 @@ impl ImageTranscodeService { fs::create_dir_all(self.cache_dir.join("webp")).await?; tracing::info!( "🖼️ Image transcode service initialized (rayon pool: {} threads, cache dir: {:?})", - TRANSCODE_POOL_THREADS, + transcode_thread_count(), self.cache_dir ); Ok(())