From 6ad23e0acc66d48fefae37b45577c9c0078a9ed3 Mon Sep 17 00:00:00 2001 From: Diocrafts Date: Sun, 22 Feb 2026 22:37:36 +0100 Subject: [PATCH] fix(perf): replace std::sync::Mutex with moka lock-free cache in async context MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Eliminates deadlock risk under concurrent load: - SearchService: Arc> → moka::sync::Cache with automatic TTL + LRU - Removed manual cleanup task, TTL checking, eviction logic (~90 lines) - get_from_cache/store_in_cache are now single lock-free calls - clear_search_cache uses invalidate_all() - HttpCache: Arc> → moka::sync::Cache - Removed stats(), cleanup(), evict_oldest() manual methods - Removed CacheEntry.timestamp/max_age fields (moka handles internally) - Removed start_cache_cleanup_task (moka evicts lazily) - routes.rs: Removed dead HttpCache instantiation and unused TTL variables Impact: std::sync::Mutex::lock() blocked Tokio worker threads; N concurrent requests (N = CPU count) could freeze the entire server. moka::sync::Cache is lock-free and designed for async runtimes — zero contention. --- src/application/services/search_service.rs | 125 ++--------------- src/interfaces/api/routes.rs | 13 -- src/interfaces/middleware/cache.rs | 153 ++++----------------- 3 files changed, 38 insertions(+), 253 deletions(-) diff --git a/src/application/services/search_service.rs b/src/application/services/search_service.rs index bbcdd7ed..f2180f0c 100644 --- a/src/application/services/search_service.rs +++ b/src/application/services/search_service.rs @@ -1,9 +1,6 @@ use async_trait::async_trait; -use std::collections::HashMap; use std::sync::Arc; -use std::sync::Mutex; use std::time::{Duration, Instant}; -use tokio::time; use crate::application::dtos::display_helpers::{ category_for, icon_class_for, icon_special_class_for, @@ -43,14 +40,8 @@ pub struct SearchService { /// Repository for folder operations folder_repository: Arc, - /// Search results cache with expiration time - search_cache: Arc>>, - - /// Cache validity duration in seconds - cache_ttl: u64, - - /// Maximum cache size (number of stored results) - max_cache_size: usize, + /// Lock-free concurrent cache with automatic TTL and LRU eviction (moka) + search_cache: moka::sync::Cache, } /// Key for the search cache @@ -63,15 +54,6 @@ struct SearchCacheKey { user_id: String, } -/// Cached search result with expiration time -struct CachedSearchResult { - /// Search results - results: SearchResultsDto, - - /// Time when the cache entry was created - timestamp: Instant, -} - // ─── Utility functions (pure, no self — computed on the server) ───────── /// Compute relevance score (0–100) for a name against a query. @@ -138,47 +120,16 @@ impl SearchService { cache_ttl: u64, max_cache_size: usize, ) -> Self { - let search_service = Self { + let search_cache = moka::sync::Cache::builder() + .max_capacity(max_cache_size as u64) + .time_to_live(Duration::from_secs(cache_ttl)) + .build(); + + Self { file_repository, folder_repository, - search_cache: Arc::new(Mutex::new(HashMap::new())), - cache_ttl, - max_cache_size, - }; - - // Start cache cleanup task if TTL > 0 - if cache_ttl > 0 { - Self::start_cache_cleanup_task(search_service.search_cache.clone(), cache_ttl); + search_cache, } - - search_service - } - - /// Starts an asynchronous task to clean up expired cache entries. - fn start_cache_cleanup_task( - cache_ref: Arc>>, - ttl_seconds: u64, - ) { - tokio::spawn(async move { - let cleanup_interval = Duration::from_secs(ttl_seconds / 2); - let ttl = Duration::from_secs(ttl_seconds); - - loop { - time::sleep(cleanup_interval).await; - - if let Ok(mut cache) = cache_ref.lock() { - let now = Instant::now(); - let expired_keys: Vec = cache - .iter() - .filter(|(_, result)| now.duration_since(result.timestamp) > ttl) - .map(|(key, _)| key.clone()) - .collect(); - for key in expired_keys { - cache.remove(&key); - } - } - } - }); } /// Creates a cache key from the search criteria. @@ -201,62 +152,12 @@ impl SearchService { /// Attempts to retrieve results from the cache. fn get_from_cache(&self, key: &SearchCacheKey) -> Option { - if self.cache_ttl == 0 { - return None; - } - - if let Ok(cache) = self.search_cache.lock() { - if let Some(cached_result) = cache.get(key) { - let now = Instant::now(); - let ttl = Duration::from_secs(self.cache_ttl); - if now.duration_since(cached_result.timestamp) < ttl { - return Some(cached_result.results.clone()); - } - } - } - - None + self.search_cache.get(key) } /// Stores results in the cache. fn store_in_cache(&self, key: SearchCacheKey, results: SearchResultsDto) { - if self.cache_ttl == 0 { - return; - } - - if let Ok(mut cache) = self.search_cache.lock() { - let now = Instant::now(); - let ttl = Duration::from_secs(self.cache_ttl); - - // Remove expired entries - let mut expired_keys = Vec::new(); - for (key, result) in cache.iter() { - if now.duration_since(result.timestamp) > ttl { - expired_keys.push(key.clone()); - } - } - for key in expired_keys { - cache.remove(&key); - } - - // Remove oldest if cache is full - if cache.len() >= self.max_cache_size { - if let Some((oldest_key, _)) = - cache.iter().min_by_key(|(_, result)| result.timestamp) - { - let key_to_remove = oldest_key.clone(); - cache.remove(&key_to_remove); - } - } - - cache.insert( - key, - CachedSearchResult { - results, - timestamp: Instant::now(), - }, - ); - } + self.search_cache.insert(key, results); } /// Enrich a FileDto → SearchFileResultDto with server-computed metadata. @@ -795,9 +696,7 @@ impl SearchUseCase for SearchService { /// Clears the search results cache. async fn clear_search_cache(&self) -> Result<()> { - if let Ok(mut cache) = self.search_cache.lock() { - cache.clear(); - } + self.search_cache.invalidate_all(); Ok(()) } } diff --git a/src/interfaces/api/routes.rs b/src/interfaces/api/routes.rs index 1786dc36..253fab04 100644 --- a/src/interfaces/api/routes.rs +++ b/src/interfaces/api/routes.rs @@ -17,8 +17,6 @@ async fn get_version() -> AxumJson { })) } -use crate::interfaces::middleware::cache::{HttpCache, start_cache_cleanup_task}; - use crate::application::services::batch_operations::BatchOperationService; use crate::interfaces::api::handlers::admin_handler; @@ -110,17 +108,6 @@ pub fn create_api_routes(app_state: &AppState) -> Router { batch_service: batch_service.clone(), }; - // Implement HTTP Cache - let http_cache = HttpCache::new(); - - // Define TTL values for different resource types (in seconds) - let _folders_ttl = 300; // 5 minutes - let _files_list_ttl = 300; // 5 minutes - let _i18n_ttl = 3600; // 1 hour - - // Start the cleanup task for HTTP cache - start_cache_cleanup_task(http_cache.clone()); - // Create the basic folders router with service operations let folders_basic_router = Router::new() .route("/", post(FolderHandler::create_folder)) diff --git a/src/interfaces/middleware/cache.rs b/src/interfaces/middleware/cache.rs index 5be87000..d162e0be 100644 --- a/src/interfaces/middleware/cache.rs +++ b/src/interfaces/middleware/cache.rs @@ -6,18 +6,16 @@ use axum::{ use bytes::Bytes; use chrono::{DateTime, Utc}; use serde::Serialize; -use std::collections::HashMap; use std::collections::hash_map::DefaultHasher; use std::future::Future; use std::hash::{Hash, Hasher}; use std::pin::Pin; -use std::sync::{Arc, Mutex}; use std::task::{Context, Poll}; -use std::time::{Duration, SystemTime}; +use std::time::Duration; use tower::{Layer, Service}; -use tracing::{debug, info}; +use tracing::debug; -const MAX_CACHE_ENTRIES: usize = 1000; // Maximum number of cache entries +const MAX_CACHE_ENTRIES: u64 = 1000; // Maximum number of cache entries const DEFAULT_MAX_AGE: u64 = 60; // Default time-to-live in seconds // Type definitions for clarity @@ -33,18 +31,18 @@ struct CacheEntry { data: Option, /// The original headers headers: HeaderMap, - /// Timestamp of when it was stored - timestamp: SystemTime, - /// Time-to-live in seconds - max_age: u64, } -/// Cache for HTTP responses with ETag support +/// Lock-free HTTP response cache with ETag support. +/// +/// Backed by `moka::sync::Cache` — all reads and writes are lock-free and +/// safe to call from async Tokio tasks without risking worker-thread stalls. +/// TTL expiration and LRU eviction are handled automatically. #[derive(Clone)] pub struct HttpCache { - /// Cache entry storage - cache: Arc>>, - /// Default time-to-live for entries + /// Concurrent cache (lock-free, automatic TTL + LRU) + cache: moka::sync::Cache, + /// Default max-age value used in HTTP Cache-Control headers default_max_age: u64, } @@ -55,10 +53,13 @@ impl Default for HttpCache { } impl HttpCache { - /// Creates a new cache instance + /// Creates a new cache instance with the default TTL pub fn new() -> Self { Self { - cache: Arc::new(Mutex::new(HashMap::with_capacity(100))), + cache: moka::sync::Cache::builder() + .max_capacity(MAX_CACHE_ENTRIES) + .time_to_live(Duration::from_secs(DEFAULT_MAX_AGE)) + .build(), default_max_age: DEFAULT_MAX_AGE, } } @@ -66,47 +67,14 @@ impl HttpCache { /// Creates a new instance with a specified time-to-live pub fn with_max_age(max_age: u64) -> Self { Self { - cache: Arc::new(Mutex::new(HashMap::with_capacity(100))), + cache: moka::sync::Cache::builder() + .max_capacity(MAX_CACHE_ENTRIES) + .time_to_live(Duration::from_secs(max_age)) + .build(), default_max_age: max_age, } } - /// Gets cache statistics - pub fn stats(&self) -> (usize, usize) { - let lock = self.cache.lock().unwrap(); - let total = lock.len(); - - // Count valid entries - let _now = SystemTime::now(); - let valid = lock - .values() - .filter(|entry| match entry.timestamp.elapsed() { - Ok(elapsed) => elapsed.as_secs() < entry.max_age, - Err(_) => false, - }) - .count(); - - (total, valid) - } - - /// Cleans up expired entries - pub fn cleanup(&self) -> usize { - let mut lock = self.cache.lock().unwrap(); - let initial_count = lock.len(); - - // Remove expired entries - let _now = SystemTime::now(); - lock.retain(|_, entry| match entry.timestamp.elapsed() { - Ok(elapsed) => elapsed.as_secs() < entry.max_age, - Err(_) => false, - }); - - let removed = initial_count - lock.len(); - debug!("HttpCache cleanup: removed {} expired entries", removed); - - removed - } - /// Sets an entry in the cache fn set( &self, @@ -114,76 +82,27 @@ impl HttpCache { etag: EntityTag, data: Option, headers: HeaderMap, - max_age: Option, ) { - let mut lock = self.cache.lock().unwrap(); - - // Apply eviction policy if the cache is full - if lock.len() >= MAX_CACHE_ENTRIES { - debug!("Cache full, removing oldest entries"); - // Remove the oldest 10% of entries - self.evict_oldest(&mut lock, MAX_CACHE_ENTRIES / 10); - } - - // Store the new entry - lock.insert( + self.cache.insert( key.to_string(), CacheEntry { etag, data, headers, - timestamp: SystemTime::now(), - max_age: max_age.unwrap_or(self.default_max_age), }, ); } - /// Removes the oldest entries from the cache - fn evict_oldest(&self, cache: &mut HashMap, count: usize) { - // Sort by timestamp - let mut entries: Vec<(CacheKey, SystemTime)> = cache - .iter() - .map(|(key, entry)| (key.clone(), entry.timestamp)) - .collect(); - - // Sort by timestamp (oldest first) - entries.sort_by(|a, b| a.1.cmp(&b.1)); - - // Remove the oldest entries - for (key, _) in entries.iter().take(count) { - cache.remove(key); - } - } - - /// Gets an entry from the cache + /// Gets an entry from the cache (returns None for expired / missing) fn get(&self, key: &str) -> Option { - let lock = self.cache.lock().unwrap(); - - // Look up the entry - if let Some(entry) = lock.get(key) { - // Check if it has expired - match entry.timestamp.elapsed() { - Ok(elapsed) if elapsed.as_secs() < entry.max_age => { - // Entry is still valid - return Some(entry.clone()); - } - _ => { - // Entry has expired - return None; - } - } - } - - None + self.cache.get(key) } /// Generates a simple ETag for a block of bytes fn calculate_etag_for_bytes(&self, bytes: &[u8]) -> EntityTag { - // Calculate hash let mut hasher = DefaultHasher::new(); bytes.hash(&mut hasher); let hash = hasher.finish(); - format!("\"{}\"", hash) } } @@ -239,7 +158,7 @@ where set_cache_headers( &mut response, &cache_entry.etag, - max_age.unwrap_or(cache_entry.max_age), + max_age.unwrap_or(cache.default_max_age), ); return Ok(response); @@ -270,7 +189,6 @@ where etag.clone(), Some(bytes.clone()), parts.headers.clone(), - max_age, ); // Create the response with ETag @@ -439,7 +357,7 @@ where set_cache_headers( &mut response, &cache_entry.etag, - max_age.unwrap_or(cache_entry.max_age), + max_age.unwrap_or(cache_clone.default_max_age), ); Box::pin(async move { Ok(response) }) @@ -474,7 +392,6 @@ where etag.clone(), Some(bytes.clone()), parts.headers.clone(), - max_age, ); // Create the response with ETag @@ -515,24 +432,6 @@ where Response::from_parts(parts, Body::from(collected)) } -/// Starts a periodic cleanup task for the cache -pub fn start_cache_cleanup_task(cache: HttpCache) { - tokio::spawn(async move { - let mut interval = tokio::time::interval(Duration::from_secs(300)); // Every 5 minutes - - loop { - interval.tick().await; - let removed = cache.cleanup(); - let (total, valid) = cache.stats(); - - info!( - "HTTP Cache cleanup: removed {}, current: {}/{}", - removed, valid, total - ); - } - }); -} - #[cfg(test)] mod tests { use super::*; @@ -584,7 +483,7 @@ mod tests { let headers1 = HeaderMap::new(); let etag1 = cache.calculate_etag_for_bytes(&bytes1); - cache.set("test", etag1.clone(), Some(bytes1.clone()), headers1, None); + cache.set("test", etag1.clone(), Some(bytes1.clone()), headers1); // Verify cache hit let entry = cache.get("test").unwrap();