fix(perf): replace std::sync::Mutex with moka lock-free cache in async context

Eliminates deadlock risk under concurrent load:
- SearchService: Arc<Mutex<HashMap>> → 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<Mutex<HashMap>> → 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.
This commit is contained in:
Diocrafts
2026-02-22 22:37:36 +01:00
parent 5b4cd30e2b
commit 6ad23e0acc
3 changed files with 38 additions and 253 deletions
+12 -113
View File
@@ -1,9 +1,6 @@
use async_trait::async_trait; use async_trait::async_trait;
use std::collections::HashMap;
use std::sync::Arc; use std::sync::Arc;
use std::sync::Mutex;
use std::time::{Duration, Instant}; use std::time::{Duration, Instant};
use tokio::time;
use crate::application::dtos::display_helpers::{ use crate::application::dtos::display_helpers::{
category_for, icon_class_for, icon_special_class_for, category_for, icon_class_for, icon_special_class_for,
@@ -43,14 +40,8 @@ pub struct SearchService {
/// Repository for folder operations /// Repository for folder operations
folder_repository: Arc<dyn FolderStoragePort>, folder_repository: Arc<dyn FolderStoragePort>,
/// Search results cache with expiration time /// Lock-free concurrent cache with automatic TTL and LRU eviction (moka)
search_cache: Arc<Mutex<HashMap<SearchCacheKey, CachedSearchResult>>>, search_cache: moka::sync::Cache<SearchCacheKey, SearchResultsDto>,
/// Cache validity duration in seconds
cache_ttl: u64,
/// Maximum cache size (number of stored results)
max_cache_size: usize,
} }
/// Key for the search cache /// Key for the search cache
@@ -63,15 +54,6 @@ struct SearchCacheKey {
user_id: String, 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) ───────── // ─── Utility functions (pure, no self — computed on the server) ─────────
/// Compute relevance score (0–100) for a name against a query. /// Compute relevance score (0–100) for a name against a query.
@@ -138,47 +120,16 @@ impl SearchService {
cache_ttl: u64, cache_ttl: u64,
max_cache_size: usize, max_cache_size: usize,
) -> Self { ) -> 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, file_repository,
folder_repository, folder_repository,
search_cache: Arc::new(Mutex::new(HashMap::new())), search_cache,
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_service
}
/// Starts an asynchronous task to clean up expired cache entries.
fn start_cache_cleanup_task(
cache_ref: Arc<Mutex<HashMap<SearchCacheKey, CachedSearchResult>>>,
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<SearchCacheKey> = 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. /// Creates a cache key from the search criteria.
@@ -201,62 +152,12 @@ impl SearchService {
/// Attempts to retrieve results from the cache. /// Attempts to retrieve results from the cache.
fn get_from_cache(&self, key: &SearchCacheKey) -> Option<SearchResultsDto> { fn get_from_cache(&self, key: &SearchCacheKey) -> Option<SearchResultsDto> {
if self.cache_ttl == 0 { self.search_cache.get(key)
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
} }
/// Stores results in the cache. /// Stores results in the cache.
fn store_in_cache(&self, key: SearchCacheKey, results: SearchResultsDto) { fn store_in_cache(&self, key: SearchCacheKey, results: SearchResultsDto) {
if self.cache_ttl == 0 { self.search_cache.insert(key, results);
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(),
},
);
}
} }
/// Enrich a FileDto → SearchFileResultDto with server-computed metadata. /// Enrich a FileDto → SearchFileResultDto with server-computed metadata.
@@ -795,9 +696,7 @@ impl SearchUseCase for SearchService {
/// Clears the search results cache. /// Clears the search results cache.
async fn clear_search_cache(&self) -> Result<()> { async fn clear_search_cache(&self) -> Result<()> {
if let Ok(mut cache) = self.search_cache.lock() { self.search_cache.invalidate_all();
cache.clear();
}
Ok(()) Ok(())
} }
} }
-13
View File
@@ -17,8 +17,6 @@ async fn get_version() -> AxumJson<serde_json::Value> {
})) }))
} }
use crate::interfaces::middleware::cache::{HttpCache, start_cache_cleanup_task};
use crate::application::services::batch_operations::BatchOperationService; use crate::application::services::batch_operations::BatchOperationService;
use crate::interfaces::api::handlers::admin_handler; use crate::interfaces::api::handlers::admin_handler;
@@ -110,17 +108,6 @@ pub fn create_api_routes(app_state: &AppState) -> Router<AppState> {
batch_service: batch_service.clone(), 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 // Create the basic folders router with service operations
let folders_basic_router = Router::new() let folders_basic_router = Router::new()
.route("/", post(FolderHandler::create_folder)) .route("/", post(FolderHandler::create_folder))
+26 -127
View File
@@ -6,18 +6,16 @@ use axum::{
use bytes::Bytes; use bytes::Bytes;
use chrono::{DateTime, Utc}; use chrono::{DateTime, Utc};
use serde::Serialize; use serde::Serialize;
use std::collections::HashMap;
use std::collections::hash_map::DefaultHasher; use std::collections::hash_map::DefaultHasher;
use std::future::Future; use std::future::Future;
use std::hash::{Hash, Hasher}; use std::hash::{Hash, Hasher};
use std::pin::Pin; use std::pin::Pin;
use std::sync::{Arc, Mutex};
use std::task::{Context, Poll}; use std::task::{Context, Poll};
use std::time::{Duration, SystemTime}; use std::time::Duration;
use tower::{Layer, Service}; 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 const DEFAULT_MAX_AGE: u64 = 60; // Default time-to-live in seconds
// Type definitions for clarity // Type definitions for clarity
@@ -33,18 +31,18 @@ struct CacheEntry {
data: Option<Bytes>, data: Option<Bytes>,
/// The original headers /// The original headers
headers: HeaderMap, 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)] #[derive(Clone)]
pub struct HttpCache { pub struct HttpCache {
/// Cache entry storage /// Concurrent cache (lock-free, automatic TTL + LRU)
cache: Arc<Mutex<HashMap<CacheKey, CacheEntry>>>, cache: moka::sync::Cache<CacheKey, CacheEntry>,
/// Default time-to-live for entries /// Default max-age value used in HTTP Cache-Control headers
default_max_age: u64, default_max_age: u64,
} }
@@ -55,10 +53,13 @@ impl Default for HttpCache {
} }
impl HttpCache { impl HttpCache {
/// Creates a new cache instance /// Creates a new cache instance with the default TTL
pub fn new() -> Self { pub fn new() -> Self {
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, default_max_age: DEFAULT_MAX_AGE,
} }
} }
@@ -66,47 +67,14 @@ impl HttpCache {
/// Creates a new instance with a specified time-to-live /// Creates a new instance with a specified time-to-live
pub fn with_max_age(max_age: u64) -> Self { pub fn with_max_age(max_age: u64) -> Self {
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, 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 /// Sets an entry in the cache
fn set( fn set(
&self, &self,
@@ -114,76 +82,27 @@ impl HttpCache {
etag: EntityTag, etag: EntityTag,
data: Option<Bytes>, data: Option<Bytes>,
headers: HeaderMap, headers: HeaderMap,
max_age: Option<u64>,
) { ) {
let mut lock = self.cache.lock().unwrap(); self.cache.insert(
// 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(
key.to_string(), key.to_string(),
CacheEntry { CacheEntry {
etag, etag,
data, data,
headers, headers,
timestamp: SystemTime::now(),
max_age: max_age.unwrap_or(self.default_max_age),
}, },
); );
} }
/// Removes the oldest entries from the cache /// Gets an entry from the cache (returns None for expired / missing)
fn evict_oldest(&self, cache: &mut HashMap<CacheKey, CacheEntry>, 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
fn get(&self, key: &str) -> Option<CacheEntry> { fn get(&self, key: &str) -> Option<CacheEntry> {
let lock = self.cache.lock().unwrap(); self.cache.get(key)
// 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
} }
/// Generates a simple ETag for a block of bytes /// Generates a simple ETag for a block of bytes
fn calculate_etag_for_bytes(&self, bytes: &[u8]) -> EntityTag { fn calculate_etag_for_bytes(&self, bytes: &[u8]) -> EntityTag {
// Calculate hash
let mut hasher = DefaultHasher::new(); let mut hasher = DefaultHasher::new();
bytes.hash(&mut hasher); bytes.hash(&mut hasher);
let hash = hasher.finish(); let hash = hasher.finish();
format!("\"{}\"", hash) format!("\"{}\"", hash)
} }
} }
@@ -239,7 +158,7 @@ where
set_cache_headers( set_cache_headers(
&mut response, &mut response,
&cache_entry.etag, &cache_entry.etag,
max_age.unwrap_or(cache_entry.max_age), max_age.unwrap_or(cache.default_max_age),
); );
return Ok(response); return Ok(response);
@@ -270,7 +189,6 @@ where
etag.clone(), etag.clone(),
Some(bytes.clone()), Some(bytes.clone()),
parts.headers.clone(), parts.headers.clone(),
max_age,
); );
// Create the response with ETag // Create the response with ETag
@@ -439,7 +357,7 @@ where
set_cache_headers( set_cache_headers(
&mut response, &mut response,
&cache_entry.etag, &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) }) Box::pin(async move { Ok(response) })
@@ -474,7 +392,6 @@ where
etag.clone(), etag.clone(),
Some(bytes.clone()), Some(bytes.clone()),
parts.headers.clone(), parts.headers.clone(),
max_age,
); );
// Create the response with ETag // Create the response with ETag
@@ -515,24 +432,6 @@ where
Response::from_parts(parts, Body::from(collected)) 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)] #[cfg(test)]
mod tests { mod tests {
use super::*; use super::*;
@@ -584,7 +483,7 @@ mod tests {
let headers1 = HeaderMap::new(); let headers1 = HeaderMap::new();
let etag1 = cache.calculate_etag_for_bytes(&bytes1); 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 // Verify cache hit
let entry = cache.get("test").unwrap(); let entry = cache.get("test").unwrap();