{t('common.download', 'Download')}
selected.clear()}>{t('common.clear', 'Clear')}
diff --git a/frontend/static/locales/es.json b/frontend/static/locales/es.json
index 6f69a5cc..1c5f2912 100644
--- a/frontend/static/locales/es.json
+++ b/frontend/static/locales/es.json
@@ -364,7 +364,8 @@
"folder": "Carpeta",
"new_folder": "Nueva carpeta",
"share": "Compartir",
- "view": "Ver"
+ "view": "Ver",
+ "selected_count": "{{count}} seleccionados"
},
"dialogs": {
"rename_folder": "Renombrar carpeta",
diff --git a/src/application/services/file_retrieval_service.rs b/src/application/services/file_retrieval_service.rs
index afe4d5c1..00a6741b 100644
--- a/src/application/services/file_retrieval_service.rs
+++ b/src/application/services/file_retrieval_service.rs
@@ -66,6 +66,25 @@ impl FileRetrievalService {
// ── private helpers ──────────────────────────────────────────
+ /// Read a file's full content through the streaming API into a single
+ /// `Bytes` buffer. Working memory stays at one chunk while reading; the
+ /// returned buffer holds the whole (sub-threshold) file.
+ async fn read_full(
+ file_read: &FileBlobReadRepository,
+ id: &str,
+ capacity: usize,
+ ) -> Result
{
+ let stream = file_read.get_file_stream(id).await?;
+ let mut stream = Pin::from(stream);
+ let mut buf = BytesMut::with_capacity(capacity);
+ while let Some(chunk) = stream.next().await {
+ buf.extend_from_slice(&chunk.map_err(|e| {
+ DomainError::internal_error("File", format!("Stream read error: {}", e))
+ })?);
+ }
+ Ok(buf.freeze())
+ }
+
/// Helper: require the caller has `perm` on the given file id.
/// Fail-closed if no engine was injected (stub/test path).
async fn require_file(
@@ -162,60 +181,31 @@ impl FileRetrievalService {
// ── Tier 1: Hot cache + transcode (<10 MB) ──────────
if file_size < CACHE_THRESHOLD {
- // Check content cache first (keyed by blob hash — see above)
- if cacheable
- && let Some(cache) = &self.content_cache
- && let Some((cached, _etag, _ct)) = cache.get(&cache_key).await
- {
- debug!(
- "🔥 TIER 1 Cache HIT: {} ({} bytes)",
- file_name,
- cached.len()
- );
- if do_transcode
- && let Some((t, m)) = self
- .try_transcode(id, &cached, &mime_type, file_size, true)
- .await
- {
- return Ok((
- dto,
- OptimizedFileContent::Bytes {
- data: t,
- mime_type: m,
- was_transcoded: true,
- },
- ));
- }
- return Ok((
- dto,
- OptimizedFileContent::Bytes {
- data: cached,
- mime_type: mime_type.clone(),
- was_transcoded: false,
- },
- ));
- }
-
- // Cache miss – load from disk via streaming (constant 64 KB memory)
- debug!("💾 TIER 1 Cache MISS: {} – loading from disk", file_name);
- let stream = self.file_read.get_file_stream(id).await?;
- let mut stream = std::pin::Pin::from(stream);
- let mut buf = BytesMut::with_capacity(file_size as usize);
- while let Some(chunk) = stream.next().await {
- buf.extend_from_slice(&chunk.map_err(|e| {
- DomainError::internal_error("File", format!("Stream read error: {}", e))
- })?);
- }
- let content_bytes = buf.freeze();
-
- // Store in cache (keyed by blob hash; ETag = the immutable hash)
- if cacheable && let Some(cache) = &self.content_cache {
+ // Fetch the raw blob bytes. When cacheable, `get_or_load` serves
+ // from the content cache on a hit and, on a miss, coalesces every
+ // concurrent request for the same blob hash into a SINGLE disk read
+ // (single-flight) — no thundering herd under load. Hash-less stub
+ // DTOs are uncacheable and stream straight from disk.
+ let content_bytes = if cacheable && let Some(cache) = &self.content_cache {
let etag: Arc = format!("\"{}\"", cache_key).into();
let ct: Arc = mime_type.clone();
- cache
- .put(cache_key.clone(), content_bytes.clone(), etag, ct)
- .await;
- }
+ let file_read = Arc::clone(&self.file_read);
+ let id_owned = id.to_string();
+ let cap = file_size as usize;
+ let (bytes, _etag, _ct) = cache
+ .get_or_load(cache_key.clone(), etag, ct, async move {
+ debug!("💾 TIER 1 Cache MISS: {} – loading from disk", id_owned);
+ Self::read_full(&file_read, &id_owned, cap).await
+ })
+ .await?;
+ bytes
+ } else {
+ debug!(
+ "💾 TIER 1 (uncacheable): {} – streaming from disk",
+ file_name
+ );
+ Self::read_full(&self.file_read, id, file_size as usize).await?
+ };
if do_transcode
&& let Some((t, m)) = self
diff --git a/src/application/services/search_service.rs b/src/application/services/search_service.rs
index e0ed4602..1b3b1262 100644
--- a/src/application/services/search_service.rs
+++ b/src/application/services/search_service.rs
@@ -193,16 +193,6 @@ impl SearchService {
hasher.finish()
}
- /// Attempts to retrieve results from the cache.
- async fn get_from_cache(&self, key: u64) -> Option> {
- self.search_cache.get(&key).await
- }
-
- /// Stores results in the cache.
- async fn store_in_cache(&self, key: u64, results: Arc) {
- self.search_cache.insert(key, results).await;
- }
-
/// Enrich a FileDto → SearchFileResultDto with server-computed metadata.
///
/// `query_lower` must already be lowercased (empty string when no query).
@@ -522,209 +512,215 @@ impl SearchUseCase for SearchService {
criteria: SearchCriteriaDto,
user_id: Uuid,
) -> Result> {
- let start = Instant::now();
let user_id_str = user_id.to_string();
-
- // Try to get from cache
let cache_key = Self::create_cache_key(&criteria, &user_id_str);
- if let Some(cached_results) = self.get_from_cache(cache_key).await {
- return Ok(cached_results);
- }
- let query = criteria.name_contains.as_deref().unwrap_or("");
- // Pre-compute once — avoids N heap allocations inside enrich_file/enrich_folder.
- let query_lower = query.to_lowercase();
+ // Single-flight: collapse N identical concurrent searches into ONE
+ // execution. `try_get_with` serves the cached result on a hit and, on a
+ // miss, runs the closure exactly once while the other callers await it
+ // — so a burst of identical queries no longer floods Postgres or drains
+ // the connection pool (the old get-from-cache fast path is subsumed).
+ self.search_cache
+ .try_get_with(cache_key, async move {
+ let start = Instant::now();
+ let query = criteria.name_contains.as_deref().unwrap_or("");
+ // Pre-compute once — avoids N heap allocations inside enrich_file/enrich_folder.
+ let query_lower = query.to_lowercase();
- // Content-index candidates (first page only). Feature-off or an
- // index failure yields an empty set — the search stays name-only.
- let content_hits = self.lookup_content_hits(&criteria, user_id).await;
+ // Content-index candidates (first page only). Feature-off or an
+ // index failure yields an empty set — the search stays name-only.
+ let content_hits = self.lookup_content_hits(&criteria, user_id).await;
- // For non-recursive searches, use efficient database-level pagination
- // This avoids loading all files into memory
- if !criteria.recursive {
- // Use database-level pagination
- let (files, total_file_count) = self
- .file_repository
- .search_files_paginated(criteria.folder_id.as_deref(), &criteria, user_id)
- .await?;
+ // For non-recursive searches, use efficient database-level pagination
+ // This avoids loading all files into memory
+ if !criteria.recursive {
+ // Use database-level pagination
+ let (files, total_file_count) = self
+ .file_repository
+ .search_files_paginated(criteria.folder_id.as_deref(), &criteria, user_id)
+ .await?;
- // Convert to DTOs and enrich with metadata
- let file_dtos: Vec = files.into_iter().map(FileDto::from).collect();
- let mut enriched_files: Vec = file_dtos
- .iter()
- .map(|f| Self::enrich_file(f, &query_lower))
- .collect();
+ // Convert to DTOs and enrich with metadata
+ let file_dtos: Vec = files.into_iter().map(FileDto::from).collect();
+ let mut enriched_files: Vec = file_dtos
+ .iter()
+ .map(|f| Self::enrich_file(f, &query_lower))
+ .collect();
- // Get folders for this folder (non-recursive, filtered in SQL)
- let folders = self
- .folder_repository
- .search_folders(
- criteria.folder_id.as_deref(),
- criteria.name_contains.as_deref(),
- user_id,
- false,
+ // Get folders for this folder (non-recursive, filtered in SQL)
+ let folders = self
+ .folder_repository
+ .search_folders(
+ criteria.folder_id.as_deref(),
+ criteria.name_contains.as_deref(),
+ user_id,
+ false,
+ )
+ .await?;
+
+ let filtered_folders: Vec =
+ folders.into_iter().map(FolderDto::from).collect();
+
+ // For folders, apply sorting and pagination in memory (usually fewer folders)
+ let mut enriched_folders: Vec = filtered_folders
+ .iter()
+ .map(|f| Self::enrich_folder(f, &query_lower))
+ .collect();
+
+ // Sort folders (cached_key avoids O(N log N) temporary String allocations)
+ match criteria.sort_by.as_str() {
+ "name" => {
+ enriched_folders.sort_by_cached_key(|f| f.name.to_lowercase());
+ }
+ "name_desc" => {
+ enriched_folders.sort_by_cached_key(|f| Reverse(f.name.to_lowercase()));
+ }
+ "date" => {
+ enriched_folders.sort_by_key(|f| f.modified_at);
+ }
+ "date_desc" => {
+ enriched_folders.sort_by_key(|f| Reverse(f.modified_at));
+ }
+ _ => {
+ enriched_folders.sort_by_key(|f| Reverse(f.relevance_score));
+ }
+ }
+
+ // Blend in content-discovered files before the pagination math.
+ let added = self
+ .merge_content_hits(content_hits, &mut enriched_files, &criteria, user_id)
+ .await?;
+ let total_file_count = total_file_count + added;
+
+ let folder_count = enriched_folders.len();
+ let total_count = total_file_count + folder_count;
+
+ // Combine and paginate (folders first, then files)
+ let start_idx = criteria.offset.min(total_count);
+ let end_idx = (criteria.offset + criteria.limit).min(total_count);
+
+ let folder_start = start_idx.min(folder_count);
+ let folder_end = end_idx.min(folder_count);
+ let paginated_folders = enriched_folders[folder_start..folder_end].to_vec();
+
+ let file_start = start_idx.saturating_sub(folder_count);
+ let file_end = end_idx
+ .saturating_sub(folder_count)
+ .min(enriched_files.len());
+ let paginated_files = enriched_files[file_start..file_end].to_vec();
+
+ let elapsed_ms = start.elapsed().as_millis() as u64;
+
+ let search_results = Arc::new(SearchResultsDto::new(
+ paginated_files,
+ paginated_folders,
+ criteria.limit,
+ criteria.offset,
+ Some(total_count),
+ elapsed_ms,
+ criteria.sort_by.clone(),
+ ));
+
+ return Ok(search_results);
+ }
+
+ // ── Recursive search via ltree (single SQL query per entity type) ──
+ // Uses PostgreSQL ltree GiST index to find all files and folders
+ // in the subtree in O(1) queries, replacing the O(N) spawn-per-folder
+ // approach that could saturate the connection pool.
+ let (found_files, total_file_count) = self
+ .file_repository
+ .search_files_in_subtree(criteria.folder_id.as_deref(), &criteria, user_id)
+ .await?;
+
+ // Get folders (SQL-filtered, user-scoped, recursive when applicable)
+ let found_folders: Vec = self
+ .folder_repository
+ .search_folders(
+ criteria.folder_id.as_deref(),
+ criteria.name_contains.as_deref(),
+ user_id,
+ true,
+ )
+ .await?;
+
+ // ── Convert to DTOs and enrich with server-computed metadata ──
+ let file_dtos: Vec = found_files.into_iter().map(FileDto::from).collect();
+ let mut enriched_files: Vec = file_dtos
+ .iter()
+ .map(|f| Self::enrich_file(f, &query_lower))
+ .collect();
+
+ let folder_dtos: Vec =
+ found_folders.into_iter().map(FolderDto::from).collect();
+ let mut enriched_folders: Vec = folder_dtos
+ .iter()
+ .map(|f| Self::enrich_folder(f, &query_lower))
+ .collect();
+
+ // ── Sort folders (cached_key avoids O(N log N) temporary String allocations) ──
+ match criteria.sort_by.as_str() {
+ "name" => {
+ enriched_folders.sort_by_cached_key(|f| f.name.to_lowercase());
+ }
+ "name_desc" => {
+ enriched_folders.sort_by_cached_key(|f| Reverse(f.name.to_lowercase()));
+ }
+ "date" => {
+ enriched_folders.sort_by_key(|f| f.modified_at);
+ }
+ "date_desc" => {
+ enriched_folders.sort_by_key(|f| Reverse(f.modified_at));
+ }
+ _ => {
+ enriched_folders.sort_by_key(|f| Reverse(f.relevance_score));
+ }
+ }
+
+ // Blend in content-discovered files before the pagination math.
+ let added = self
+ .merge_content_hits(content_hits, &mut enriched_files, &criteria, user_id)
+ .await?;
+ let total_file_count = total_file_count + added;
+
+ // ── Pagination (folders first, then files) ──
+ let folder_count = enriched_folders.len();
+ let total_count = total_file_count + folder_count;
+ let start_idx = criteria.offset.min(total_count);
+ let end_idx = (criteria.offset + criteria.limit).min(total_count);
+
+ let folder_start = start_idx.min(folder_count);
+ let folder_end = end_idx.min(folder_count);
+ let paginated_folders = enriched_folders[folder_start..folder_end].to_vec();
+
+ let file_start = start_idx.saturating_sub(folder_count);
+ let file_end = end_idx
+ .saturating_sub(folder_count)
+ .min(enriched_files.len());
+ let paginated_files = enriched_files[file_start..file_end].to_vec();
+
+ let elapsed_ms = start.elapsed().as_millis() as u64;
+
+ let search_results = Arc::new(SearchResultsDto::new(
+ paginated_files,
+ paginated_folders,
+ criteria.limit,
+ criteria.offset,
+ Some(total_count),
+ elapsed_ms,
+ criteria.sort_by.clone(),
+ ));
+
+ Ok(search_results)
+ })
+ .await
+ .map_err(|shared: Arc| {
+ crate::common::errors::DomainError::new(
+ shared.kind,
+ shared.entity_type,
+ shared.message.clone(),
)
- .await?;
-
- let filtered_folders: Vec =
- folders.into_iter().map(FolderDto::from).collect();
-
- // For folders, apply sorting and pagination in memory (usually fewer folders)
- let mut enriched_folders: Vec = filtered_folders
- .iter()
- .map(|f| Self::enrich_folder(f, &query_lower))
- .collect();
-
- // Sort folders (cached_key avoids O(N log N) temporary String allocations)
- match criteria.sort_by.as_str() {
- "name" => {
- enriched_folders.sort_by_cached_key(|f| f.name.to_lowercase());
- }
- "name_desc" => {
- enriched_folders.sort_by_cached_key(|f| Reverse(f.name.to_lowercase()));
- }
- "date" => {
- enriched_folders.sort_by_key(|f| f.modified_at);
- }
- "date_desc" => {
- enriched_folders.sort_by_key(|f| Reverse(f.modified_at));
- }
- _ => {
- enriched_folders.sort_by_key(|f| Reverse(f.relevance_score));
- }
- }
-
- // Blend in content-discovered files before the pagination math.
- let added = self
- .merge_content_hits(content_hits, &mut enriched_files, &criteria, user_id)
- .await?;
- let total_file_count = total_file_count + added;
-
- let folder_count = enriched_folders.len();
- let total_count = total_file_count + folder_count;
-
- // Combine and paginate (folders first, then files)
- let start_idx = criteria.offset.min(total_count);
- let end_idx = (criteria.offset + criteria.limit).min(total_count);
-
- let folder_start = start_idx.min(folder_count);
- let folder_end = end_idx.min(folder_count);
- let paginated_folders = enriched_folders[folder_start..folder_end].to_vec();
-
- let file_start = start_idx.saturating_sub(folder_count);
- let file_end = end_idx
- .saturating_sub(folder_count)
- .min(enriched_files.len());
- let paginated_files = enriched_files[file_start..file_end].to_vec();
-
- let elapsed_ms = start.elapsed().as_millis() as u64;
-
- let search_results = Arc::new(SearchResultsDto::new(
- paginated_files,
- paginated_folders,
- criteria.limit,
- criteria.offset,
- Some(total_count),
- elapsed_ms,
- criteria.sort_by.clone(),
- ));
-
- self.store_in_cache(cache_key, Arc::clone(&search_results))
- .await;
- return Ok(search_results);
- }
-
- // ── Recursive search via ltree (single SQL query per entity type) ──
- // Uses PostgreSQL ltree GiST index to find all files and folders
- // in the subtree in O(1) queries, replacing the O(N) spawn-per-folder
- // approach that could saturate the connection pool.
- let (found_files, total_file_count) = self
- .file_repository
- .search_files_in_subtree(criteria.folder_id.as_deref(), &criteria, user_id)
- .await?;
-
- // Get folders (SQL-filtered, user-scoped, recursive when applicable)
- let found_folders: Vec = self
- .folder_repository
- .search_folders(
- criteria.folder_id.as_deref(),
- criteria.name_contains.as_deref(),
- user_id,
- true,
- )
- .await?;
-
- // ── Convert to DTOs and enrich with server-computed metadata ──
- let file_dtos: Vec = found_files.into_iter().map(FileDto::from).collect();
- let mut enriched_files: Vec = file_dtos
- .iter()
- .map(|f| Self::enrich_file(f, &query_lower))
- .collect();
-
- let folder_dtos: Vec = found_folders.into_iter().map(FolderDto::from).collect();
- let mut enriched_folders: Vec = folder_dtos
- .iter()
- .map(|f| Self::enrich_folder(f, &query_lower))
- .collect();
-
- // ── Sort folders (cached_key avoids O(N log N) temporary String allocations) ──
- match criteria.sort_by.as_str() {
- "name" => {
- enriched_folders.sort_by_cached_key(|f| f.name.to_lowercase());
- }
- "name_desc" => {
- enriched_folders.sort_by_cached_key(|f| Reverse(f.name.to_lowercase()));
- }
- "date" => {
- enriched_folders.sort_by_key(|f| f.modified_at);
- }
- "date_desc" => {
- enriched_folders.sort_by_key(|f| Reverse(f.modified_at));
- }
- _ => {
- enriched_folders.sort_by_key(|f| Reverse(f.relevance_score));
- }
- }
-
- // Blend in content-discovered files before the pagination math.
- let added = self
- .merge_content_hits(content_hits, &mut enriched_files, &criteria, user_id)
- .await?;
- let total_file_count = total_file_count + added;
-
- // ── Pagination (folders first, then files) ──
- let folder_count = enriched_folders.len();
- let total_count = total_file_count + folder_count;
- let start_idx = criteria.offset.min(total_count);
- let end_idx = (criteria.offset + criteria.limit).min(total_count);
-
- let folder_start = start_idx.min(folder_count);
- let folder_end = end_idx.min(folder_count);
- let paginated_folders = enriched_folders[folder_start..folder_end].to_vec();
-
- let file_start = start_idx.saturating_sub(folder_count);
- let file_end = end_idx
- .saturating_sub(folder_count)
- .min(enriched_files.len());
- let paginated_files = enriched_files[file_start..file_end].to_vec();
-
- let elapsed_ms = start.elapsed().as_millis() as u64;
-
- let search_results = Arc::new(SearchResultsDto::new(
- paginated_files,
- paginated_folders,
- criteria.limit,
- criteria.offset,
- Some(total_count),
- elapsed_ms,
- criteria.sort_by.clone(),
- ));
-
- // Store in cache — Arc::clone is ~1 ns (atomic increment)
- self.store_in_cache(cache_key, Arc::clone(&search_results))
- .await;
-
- Ok(search_results)
+ })
}
/// Returns quick suggestions for autocomplete.
diff --git a/src/common/config.rs b/src/common/config.rs
index 84bad869..43f92339 100644
--- a/src/common/config.rs
+++ b/src/common/config.rs
@@ -424,6 +424,18 @@ pub struct DatabaseConfig {
/// Minimum connections for the maintenance pool.
/// Defaults to 1.
pub maintenance_min_connections: u32,
+ /// Per-statement timeout (seconds) applied to the **primary** pool via
+ /// `SET statement_timeout` on every connection. Bounds the worst-case query
+ /// so a single runaway statement can't pin a pool slot and starve
+ /// interactive requests (correlated tail-latency cliff). `0` disables it.
+ /// The maintenance pool is always exempt — its batch jobs (integrity scans,
+ /// GC) may legitimately run long. Env: `OXICLOUD_DB_STATEMENT_TIMEOUT_SECS`.
+ pub statement_timeout_secs: u64,
+ /// Interval (seconds) of the background watchdog that samples primary-pool
+ /// saturation and logs a WARN when connections are near exhaustion (the
+ /// signal to raise `max_connections` or hunt slow queries). `0` disables
+ /// it. Default: 30. Env: `OXICLOUD_DB_POOL_MONITOR_INTERVAL_SECS`.
+ pub pool_monitor_interval_secs: u64,
}
impl Default for DatabaseConfig {
@@ -438,6 +450,8 @@ impl Default for DatabaseConfig {
max_lifetime_secs: 1800,
maintenance_max_connections: 5,
maintenance_min_connections: 1,
+ statement_timeout_secs: 30,
+ pool_monitor_interval_secs: 30,
}
}
}
@@ -1233,6 +1247,20 @@ impl AppConfig {
config.database.maintenance_min_connections = val;
}
+ if let Ok(stmt_timeout) =
+ env::var("OXICLOUD_DB_STATEMENT_TIMEOUT_SECS").map(|v| v.parse::())
+ && let Ok(val) = stmt_timeout
+ {
+ config.database.statement_timeout_secs = val;
+ }
+
+ if let Ok(interval) =
+ env::var("OXICLOUD_DB_POOL_MONITOR_INTERVAL_SECS").map(|v| v.parse::())
+ && let Ok(val) = interval
+ {
+ config.database.pool_monitor_interval_secs = val;
+ }
+
// Auth configuration
if let Some(jwt_secret) = env::var("OXICLOUD_JWT_SECRET")
.ok()
diff --git a/src/common/di.rs b/src/common/di.rs
index 9ff71517..00bb1ff0 100644
--- a/src/common/di.rs
+++ b/src/common/di.rs
@@ -955,6 +955,24 @@ impl AppServiceFactory {
tracing::info!("Tree-ETag flush service initialized");
}
+ /// Start the primary-pool saturation watchdog (Finding #3). Logs a WARN as
+ /// the user-facing pool approaches exhaustion — the early signal for raising
+ /// `max_connections` or chasing a slow query before tail latency cliffs.
+ /// Skipped when `pool_monitor_interval_secs == 0`.
+ fn start_db_pool_monitor(&self, primary_pool: &Arc) {
+ let interval = self.config.database.pool_monitor_interval_secs;
+ if interval == 0 {
+ return;
+ }
+ crate::infrastructure::services::db_pool_monitor::DbPoolMonitor::new(
+ primary_pool.as_ref().clone(),
+ "primary",
+ self.config.database.max_connections,
+ interval,
+ )
+ .start();
+ }
+
/// Opens (or rebuilds) the embedded Tantivy content index. Returns the
/// index plus a reseed flag (true when the on-disk index was missing or
/// version-stale and must be repopulated from `storage.files`). Any
@@ -1149,6 +1167,8 @@ impl AppServiceFactory {
self.start_tree_etag_flush_job(&maintenance_pool);
+ self.start_db_pool_monitor(&pool);
+
self.start_content_index_job(&maintenance_pool, &core, content_index);
// User-lifecycle dispatcher. Hook order is registration order;
diff --git a/src/infrastructure/db.rs b/src/infrastructure/db.rs
index 94c21ddd..1c1c05ba 100644
--- a/src/infrastructure/db.rs
+++ b/src/infrastructure/db.rs
@@ -45,6 +45,7 @@ pub async fn create_database_pools(config: &AppConfig) -> Result {
config.database.connect_timeout_secs,
config.database.idle_timeout_secs,
config.database.max_lifetime_secs,
+ config.database.statement_timeout_secs,
"primary",
)
.await?;
@@ -68,6 +69,8 @@ pub async fn create_database_pools(config: &AppConfig) -> Result {
config.database.connect_timeout_secs,
config.database.idle_timeout_secs,
config.database.max_lifetime_secs,
+ // Maintenance pool is exempt: integrity scans / GC may run long.
+ 0,
"maintenance",
)
.await?;
@@ -87,6 +90,7 @@ pub async fn create_database_pools(config: &AppConfig) -> Result {
}
/// Internal helper: create a single pool with retry logic.
+#[allow(clippy::too_many_arguments)]
async fn create_pool_with_retries(
connection_string: &str,
max_connections: u32,
@@ -94,6 +98,7 @@ async fn create_pool_with_retries(
connect_timeout_secs: u64,
idle_timeout_secs: u64,
max_lifetime_secs: u64,
+ statement_timeout_secs: u64,
label: &str,
) -> Result {
let mut attempt = 0;
@@ -108,7 +113,7 @@ async fn create_pool_with_retries(
MAX_ATTEMPTS
);
- match PgPoolOptions::new()
+ let mut opts = PgPoolOptions::new()
.max_connections(max_connections)
.min_connections(min_connections)
.acquire_timeout(Duration::from_secs(connect_timeout_secs))
@@ -119,10 +124,26 @@ async fn create_pool_with_retries(
// that extra round-trip per checkout costs more than the rare dead
// connection it catches. A stale socket surfaces as a query error
// and the pool recycles it either way.
- .test_before_acquire(false)
- .connect(connection_string)
- .await
- {
+ .test_before_acquire(false);
+
+ // Bound the worst-case query: `SET statement_timeout` on every new
+ // connection caps how long any single statement may run, so a runaway
+ // query can't pin a pool slot and starve interactive requests. `0`
+ // disables it (maintenance pool). statement_timeout's integer value is
+ // milliseconds.
+ if statement_timeout_secs > 0 {
+ use sqlx::Executor;
+ let stmt_ms = statement_timeout_secs.saturating_mul(1000);
+ opts = opts.after_connect(move |conn, _meta| {
+ Box::pin(async move {
+ conn.execute(format!("SET statement_timeout = {stmt_ms}").as_str())
+ .await?;
+ Ok(())
+ })
+ });
+ }
+
+ match opts.connect(connection_string).await {
Ok(pool) => match sqlx::query("SELECT 1").execute(&pool).await {
Ok(_) => {
tracing::info!("PostgreSQL {} pool established successfully", label);
diff --git a/src/infrastructure/services/db_pool_monitor.rs b/src/infrastructure/services/db_pool_monitor.rs
new file mode 100644
index 00000000..bc7e0695
--- /dev/null
+++ b/src/infrastructure/services/db_pool_monitor.rs
@@ -0,0 +1,188 @@
+//! Background watchdog that samples primary DB-pool saturation.
+//!
+//! A runaway query that pins a connection, multiplied across a small pool, ends
+//! in pool exhaustion: every new request then blocks on `acquire()` up to the
+//! acquire timeout — the correlated tail-latency cliff where one slow query
+//! degrades the whole server. `statement_timeout` (see `db.rs`) caps the cause;
+//! this monitor surfaces the symptom early by logging a WARN when in-use
+//! connections approach the configured maximum, so an operator can raise
+//! `OXICLOUD_DB_MAX_CONNECTIONS` or hunt the slow query before users feel it.
+//!
+//! The loop only reads in-memory pool counters (`size()` / `num_idle()`) — it
+//! never issues a query, so it can never itself contend for a connection.
+
+use sqlx::PgPool;
+use std::sync::Arc;
+use std::sync::atomic::{AtomicU32, Ordering};
+use std::time::Duration;
+use tracing::{debug, info, warn};
+
+/// WARN once in-use connections reach this fraction of the pool maximum. At
+/// ≥90% the pool is one slow query away from forcing `acquire()` waits on
+/// every request.
+const WARN_UTILIZATION_PCT: u32 = 90;
+
+/// A point-in-time sample of pool occupancy.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct PoolSample {
+ /// Connections currently checked out (in use).
+ pub active: u32,
+ /// Connections sitting idle in the pool.
+ pub idle: u32,
+ /// Configured maximum connections.
+ pub max: u32,
+}
+
+impl PoolSample {
+ /// In-use connections as a percentage of the configured maximum.
+ /// Saturates rather than dividing by zero on an unconfigured pool.
+ pub fn utilization_pct(&self) -> u32 {
+ if self.max == 0 {
+ return 0;
+ }
+ ((self.active as u64 * 100) / self.max as u64) as u32
+ }
+
+ /// True once occupancy has reached the warn threshold.
+ pub fn is_saturated(&self, warn_pct: u32) -> bool {
+ self.utilization_pct() >= warn_pct
+ }
+}
+
+/// Read a sqlx pool's occupancy. `size()` is total live connections
+/// (idle + in-use); `num_idle()` is the idle subset.
+fn sample(pool: &PgPool, max: u32) -> PoolSample {
+ let size = pool.size();
+ let idle = pool.num_idle() as u32;
+ PoolSample {
+ active: size.saturating_sub(idle),
+ idle,
+ max,
+ }
+}
+
+/// Background saturation watchdog over the primary (user-facing) pool.
+pub struct DbPoolMonitor {
+ pool: PgPool,
+ label: &'static str,
+ max_connections: u32,
+ interval: Duration,
+ /// High-water mark of in-use connections since startup (diagnostics).
+ peak_active: Arc,
+}
+
+impl DbPoolMonitor {
+ pub fn new(
+ pool: PgPool,
+ label: &'static str,
+ max_connections: u32,
+ interval_secs: u64,
+ ) -> Self {
+ Self {
+ pool,
+ label,
+ max_connections,
+ // Floor the cadence so a misconfiguration can't busy-loop.
+ interval: Duration::from_secs(interval_secs.max(1)),
+ peak_active: Arc::new(AtomicU32::new(0)),
+ }
+ }
+
+ /// Spawn the sampling loop. Fire-and-forget.
+ pub fn start(self) {
+ info!(
+ "Starting DB pool saturation monitor ({} pool, every {}s, warn ≥{}%)",
+ self.label,
+ self.interval.as_secs(),
+ WARN_UTILIZATION_PCT,
+ );
+ tokio::spawn(async move {
+ let mut ticker = tokio::time::interval(self.interval);
+ ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
+ loop {
+ ticker.tick().await;
+ let s = sample(&self.pool, self.max_connections);
+ let peak = self
+ .peak_active
+ .fetch_max(s.active, Ordering::Relaxed)
+ .max(s.active);
+
+ if s.is_saturated(WARN_UTILIZATION_PCT) {
+ warn!(
+ target: "oxicloud::db",
+ pool = self.label,
+ active = s.active,
+ idle = s.idle,
+ max = s.max,
+ utilization_pct = s.utilization_pct(),
+ peak_active = peak,
+ "⚠️ DB pool near saturation: {}/{} in use ({}%, peak {}) — requests may \
+ be queueing on acquire(); raise OXICLOUD_DB_MAX_CONNECTIONS or \
+ investigate slow queries",
+ s.active,
+ s.max,
+ s.utilization_pct(),
+ peak,
+ );
+ } else {
+ debug!(
+ target: "oxicloud::db",
+ pool = self.label,
+ active = s.active,
+ idle = s.idle,
+ max = s.max,
+ "DB pool ok: {}/{} in use ({}%)",
+ s.active,
+ s.max,
+ s.utilization_pct(),
+ );
+ }
+ }
+ });
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn utilization_and_saturation_thresholds() {
+ // 18/20 in use = 90% → saturated at the 90% threshold, not at 95%.
+ let near = PoolSample {
+ active: 18,
+ idle: 2,
+ max: 20,
+ };
+ assert_eq!(near.utilization_pct(), 90);
+ assert!(near.is_saturated(WARN_UTILIZATION_PCT));
+ assert!(!near.is_saturated(95));
+
+ // 4/20 in use = 20% → calm.
+ let calm = PoolSample {
+ active: 4,
+ idle: 16,
+ max: 20,
+ };
+ assert_eq!(calm.utilization_pct(), 20);
+ assert!(!calm.is_saturated(WARN_UTILIZATION_PCT));
+
+ // Fully checked out = 100% → saturated.
+ let full = PoolSample {
+ active: 20,
+ idle: 0,
+ max: 20,
+ };
+ assert_eq!(full.utilization_pct(), 100);
+ assert!(full.is_saturated(WARN_UTILIZATION_PCT));
+
+ // Degenerate zero-max pool: no div-by-zero, never saturated.
+ let zero = PoolSample {
+ active: 0,
+ idle: 0,
+ max: 0,
+ };
+ assert_eq!(zero.utilization_pct(), 0);
+ assert!(!zero.is_saturated(WARN_UTILIZATION_PCT));
+ }
+}
diff --git a/src/infrastructure/services/file_content_cache.rs b/src/infrastructure/services/file_content_cache.rs
index 1f0a733f..5c6ee555 100644
--- a/src/infrastructure/services/file_content_cache.rs
+++ b/src/infrastructure/services/file_content_cache.rs
@@ -1,5 +1,7 @@
+use crate::common::errors::DomainError;
use bytes::Bytes;
use moka::future::Cache;
+use std::future::Future;
use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::time::Duration;
@@ -152,6 +154,57 @@ impl FileContentCache {
debug!("Cached file {} ({} bytes)", file_id, size);
}
+ /// Get from cache, or load-and-cache with **single-flight coalescing**.
+ ///
+ /// On a miss, concurrent callers for the same `cache_key` share ONE `load`
+ /// future (moka `try_get_with`) instead of every caller hitting disk — the
+ /// classic thundering-herd / cache-stampede fix. With `N` simultaneous
+ /// requests for the same uncached blob this turns `N` disk reads into `1`
+ /// read plus `N-1` cheap waits, collapsing tail latency under load.
+ ///
+ /// Safe because the cache is content-addressed (key = immutable blob hash):
+ /// the coalesced value is identical for every caller and never goes stale,
+ /// so there is nothing to invalidate.
+ ///
+ /// `etag` / `content_type` describe the loaded content and are only used
+ /// when this call is the one that populates the entry.
+ pub async fn get_or_load(
+ &self,
+ cache_key: String,
+ etag: Arc,
+ content_type: Arc,
+ load: F,
+ ) -> Result<(Bytes, Arc, Arc), DomainError>
+ where
+ F: Future>,
+ {
+ // Fast path: lock-free hit (also keeps hit/miss stats meaningful).
+ if let Some(hit) = self.get(&cache_key).await {
+ return Ok(hit);
+ }
+
+ // Slow path: coalesce concurrent misses into a single `load`.
+ let entry = self
+ .cache
+ .try_get_with(cache_key, async move {
+ let content = load.await?;
+ Ok::(CacheEntry {
+ content,
+ etag,
+ content_type,
+ })
+ })
+ .await
+ // try_get_with hands back `Arc` shared by all waiters;
+ // DomainError isn't Clone (it carries a boxed source), so rebuild a
+ // fresh one preserving the kind / entity / message.
+ .map_err(|shared: Arc| {
+ DomainError::new(shared.kind, shared.entity_type, shared.message.clone())
+ })?;
+
+ Ok((entry.content, entry.etag, entry.content_type))
+ }
+
/// Remove a file from cache (e.g., when file is deleted or modified)
pub async fn invalidate(&self, file_id: &str) {
self.cache.remove(file_id).await;
@@ -302,4 +355,175 @@ mod tests {
assert!(cache.get("file1").await.is_none());
}
+
+ /// Correctness of the stampede fix: N concurrent misses for the same key
+ /// must coalesce into exactly ONE load (moka single-flight).
+ #[tokio::test]
+ async fn get_or_load_coalesces_concurrent_misses() {
+ use std::sync::atomic::AtomicUsize;
+
+ let cache = Arc::new(FileContentCache::new(FileContentCacheConfig::default()));
+ let loads = Arc::new(AtomicUsize::new(0));
+
+ let mut handles = Vec::new();
+ for _ in 0..64 {
+ let cache = Arc::clone(&cache);
+ let loads = Arc::clone(&loads);
+ handles.push(tokio::spawn(async move {
+ cache
+ .get_or_load(
+ "blob-hash".to_string(),
+ "\"blob-hash\"".into(),
+ "image/png".into(),
+ async move {
+ loads.fetch_add(1, Ordering::SeqCst);
+ // Slow load so all 64 tasks pile onto the same miss.
+ tokio::time::sleep(Duration::from_millis(20)).await;
+ Ok(Bytes::from_static(b"the-blob-bytes"))
+ },
+ )
+ .await
+ }));
+ }
+
+ for h in handles {
+ let (bytes, _etag, _ct) = h.await.unwrap().unwrap();
+ assert_eq!(&bytes[..], b"the-blob-bytes");
+ }
+
+ assert_eq!(
+ loads.load(Ordering::SeqCst),
+ 1,
+ "64 concurrent misses must trigger exactly ONE load (single-flight)"
+ );
+ }
+
+ /// Before/after benchmark for the cache-stampede fix.
+ ///
+ /// Run with:
+ /// cargo test --release -p oxicloud bench_stampede -- --ignored --nocapture
+ ///
+ /// Models a viral hot blob: `K` clients request the same uncached key at
+ /// once, and each load contends on a bounded resource (the rayon transcode
+ /// pool / DB pool) with `POOL` permits. Reports work amplification and tail
+ /// latency for the NAIVE get()+put() pattern vs the COALESCED get_or_load().
+ #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+ #[ignore = "benchmark — run with --ignored --nocapture"]
+ async fn bench_stampede() {
+ use std::sync::atomic::AtomicUsize;
+ use std::time::Instant;
+ use tokio::sync::Semaphore;
+
+ const K: usize = 128; // concurrent clients, all requesting the SAME hot key
+ const LOAD_MS: u64 = 30; // cost of one expensive load (disk + decode/encode)
+ const POOL: usize = 4; // bounded resource the loads contend on
+
+ // One expensive load: take a permit from the bounded pool, then work.
+ async fn expensive_load(
+ sem: Arc,
+ loads: Arc,
+ load_ms: u64,
+ ) -> Bytes {
+ let _permit = sem.acquire().await.unwrap();
+ loads.fetch_add(1, Ordering::SeqCst);
+ tokio::time::sleep(Duration::from_millis(load_ms)).await;
+ Bytes::from_static(b"blob")
+ }
+
+ fn pct(sorted: &[u128], p: f64) -> u128 {
+ if sorted.is_empty() {
+ return 0;
+ }
+ let idx = (((sorted.len() - 1) as f64) * p).round() as usize;
+ sorted[idx]
+ }
+
+ // ── Scenario A: NAIVE get() + put() (today's pattern) ──
+ let (naive_ms, naive_lats, naive_loads) = {
+ let cache = Arc::new(FileContentCache::new(FileContentCacheConfig::default()));
+ let sem = Arc::new(Semaphore::new(POOL));
+ let loads = Arc::new(AtomicUsize::new(0));
+ let t0 = Instant::now();
+ let mut handles = Vec::new();
+ for _ in 0..K {
+ let cache = Arc::clone(&cache);
+ let sem = Arc::clone(&sem);
+ let loads = Arc::clone(&loads);
+ handles.push(tokio::spawn(async move {
+ let r0 = Instant::now();
+ if cache.get("hot").await.is_some() {
+ return r0.elapsed().as_millis();
+ }
+ let bytes = expensive_load(sem, loads, LOAD_MS).await;
+ cache
+ .put("hot".to_string(), bytes, "e".into(), "t".into())
+ .await;
+ r0.elapsed().as_millis()
+ }));
+ }
+ let mut lats = Vec::new();
+ for h in handles {
+ lats.push(h.await.unwrap());
+ }
+ lats.sort_unstable();
+ (t0.elapsed().as_millis(), lats, loads.load(Ordering::SeqCst))
+ };
+
+ // ── Scenario B: COALESCED get_or_load() (the fix) ──
+ let (coal_ms, coal_lats, coal_loads) = {
+ let cache = Arc::new(FileContentCache::new(FileContentCacheConfig::default()));
+ let sem = Arc::new(Semaphore::new(POOL));
+ let loads = Arc::new(AtomicUsize::new(0));
+ let t0 = Instant::now();
+ let mut handles = Vec::new();
+ for _ in 0..K {
+ let cache = Arc::clone(&cache);
+ let sem = Arc::clone(&sem);
+ let loads = Arc::clone(&loads);
+ handles.push(tokio::spawn(async move {
+ let r0 = Instant::now();
+ cache
+ .get_or_load("hot".to_string(), "e".into(), "t".into(), async move {
+ Ok(expensive_load(sem, loads, LOAD_MS).await)
+ })
+ .await
+ .unwrap();
+ r0.elapsed().as_millis()
+ }));
+ }
+ let mut lats = Vec::new();
+ for h in handles {
+ lats.push(h.await.unwrap());
+ }
+ lats.sort_unstable();
+ (t0.elapsed().as_millis(), lats, loads.load(Ordering::SeqCst))
+ };
+
+ println!(
+ "\n╔══ Cache stampede: K={K} clients on the same hot key, pool={POOL}, load={LOAD_MS}ms ══"
+ );
+ println!("║ pattern │ loads │ p50(ms) │ p99(ms) │ max(ms) │ wall(ms)");
+ println!(
+ "║ NAIVE get()+put() │ {naive_loads:>5} │ {:>7} │ {:>7} │ {:>7} │ {naive_ms:>7}",
+ pct(&naive_lats, 0.50),
+ pct(&naive_lats, 0.99),
+ naive_lats.last().copied().unwrap_or(0)
+ );
+ println!(
+ "║ COALESCED get_or_load │ {coal_loads:>5} │ {:>7} │ {:>7} │ {:>7} │ {coal_ms:>7}",
+ pct(&coal_lats, 0.50),
+ pct(&coal_lats, 0.99),
+ coal_lats.last().copied().unwrap_or(0)
+ );
+ let amp = naive_loads as f64 / coal_loads.max(1) as f64;
+ let p99x = pct(&naive_lats, 0.99) as f64 / pct(&coal_lats, 0.99).max(1) as f64;
+ println!("╚══ {amp:.0}× fewer loads · {p99x:.0}× lower p99 tail latency\n");
+
+ // Guard rails so the benchmark also asserts the win.
+ assert_eq!(coal_loads, 1, "coalesced path must load exactly once");
+ assert!(
+ naive_loads > coal_loads * 10,
+ "naive path should stampede the loader"
+ );
+ }
}
diff --git a/src/infrastructure/services/image_transcode_service.rs b/src/infrastructure/services/image_transcode_service.rs
index a30f49d0..4501f746 100644
--- a/src/infrastructure/services/image_transcode_service.rs
+++ b/src/infrastructure/services/image_transcode_service.rs
@@ -222,7 +222,7 @@ impl ImageTranscodeService {
) -> Result<(Bytes, String, bool), String> {
let cache_key = format!("{}:{}", file_id, target_format.extension());
- // ── 1. Check moka memory cache (lock-free read) ──
+ // ── 1. Fast path: moka memory cache (lock-free read) ──
// An empty-Bytes entry is the negative sentinel: "transcoding this
// file is not beneficial — serve the original". Without it, every
// GET of such an image repeated the full decode + encode just to
@@ -237,18 +237,52 @@ impl ImageTranscodeService {
return Ok((cached, target_format.mime_type().to_string(), true));
}
- // ── 2. Check disk cache (async fs) ──
+ // ── 2. Slow path: single-flight coalescing ──
+ // A viral image requested as WebP by N clients at once would otherwise
+ // run N identical disk reads + CPU transcodes, saturating the rayon
+ // pool and inflating tail latency. `try_get_with` collapses every
+ // concurrent miss for this key into ONE `compute_transcode`; the other
+ // callers await its result. The cached value (transcoded bytes, or the
+ // empty negative sentinel) is what gets stored.
+ let original_for_loader = original_content.clone(); // O(1) ref-count bump
+ let cached = self
+ .memory_cache
+ .try_get_with(cache_key, async {
+ self.compute_transcode(file_id, original_for_loader, original_mime, target_format)
+ .await
+ })
+ .await
+ // try_get_with shares one `Arc` across waiters; DomainError
+ // here is just a String, so hand callers an owned clone.
+ .map_err(|shared: Arc| (*shared).clone())?;
+
+ if cached.is_empty() {
+ Ok((original_content, original_mime.to_string(), false))
+ } else {
+ Ok((cached, target_format.mime_type().to_string(), true))
+ }
+ }
+
+ /// Compute the value to cache for `(file_id, target_format)`: either the
+ /// transcoded WebP bytes, or an **empty `Bytes` negative sentinel** meaning
+ /// "the result wasn't smaller — serve the original". Runs the disk-cache
+ /// lookups and the CPU transcode, and is invoked at most once per key,
+ /// guarded by [`Self::get_transcoded`]'s `try_get_with` single-flight.
+ async fn compute_transcode(
+ &self,
+ file_id: &str,
+ original_content: Bytes,
+ original_mime: &str,
+ target_format: OutputFormat,
+ ) -> Result {
+ // ── Disk cache (async fs) ──
let cache_path = self.get_cache_path(file_id, target_format);
if tokio::fs::try_exists(&cache_path).await.unwrap_or(false) {
match fs::read(&cache_path).await {
Ok(data) => {
- let content = Bytes::from(data);
- self.memory_cache
- .insert(cache_key.clone(), content.clone())
- .await;
self.stats.disk_hits.fetch_add(1, Ordering::Relaxed);
tracing::debug!("💾 Transcode disk cache HIT: {}", file_id);
- return Ok((content, target_format.mime_type().to_string(), true));
+ return Ok(Bytes::from(data));
}
Err(e) => {
tracing::warn!("Failed to read cached transcode: {}", e);
@@ -256,16 +290,15 @@ impl ImageTranscodeService {
}
}
- // ── 2b. Negative verdict persisted on disk (survives restarts) ──
+ // ── Negative verdict persisted on disk (survives restarts) ──
let skip_marker = self.get_skip_marker_path(file_id, target_format);
if tokio::fs::try_exists(&skip_marker).await.unwrap_or(false) {
- self.memory_cache.insert(cache_key, Bytes::new()).await;
self.stats.disk_hits.fetch_add(1, Ordering::Relaxed);
tracing::debug!("💾 Transcode negative disk marker HIT: {}", file_id);
- return Ok((original_content, original_mime.to_string(), false));
+ return Ok(Bytes::new());
}
- // ── 3. Transcode on dedicated rayon pool (never blocks Tokio) ──
+ // ── Transcode on dedicated rayon pool (never blocks Tokio) ──
let content_for_rayon = original_content.clone(); // O(1) ref-count bump
let mime_owned = original_mime.to_string();
@@ -282,7 +315,7 @@ impl ImageTranscodeService {
let transcoded_bytes = Bytes::from(transcoded);
- // ── 4. Evaluate savings ──
+ // ── Evaluate savings ──
let original_size = original_content.len();
let transcoded_size = transcoded_bytes.len();
@@ -293,11 +326,10 @@ impl ImageTranscodeService {
original_size,
transcoded_size
);
- // Remember the negative verdict so the next GET doesn't repeat
- // the decode + encode: empty-Bytes sentinel in memory (expires
- // with the cache TTL) + zero-byte marker on disk (survives
- // restarts; removed by `invalidate` when the file changes).
- self.memory_cache.insert(cache_key, Bytes::new()).await;
+ // Remember the negative verdict so the next GET doesn't repeat the
+ // decode + encode: the caller caches the empty-Bytes sentinel (TTL)
+ // and we drop a zero-byte marker on disk (survives restarts;
+ // removed by `invalidate` when the file changes).
let marker = self.get_skip_marker_path(file_id, target_format);
tokio::spawn(async move {
if let Some(parent) = marker.parent() {
@@ -307,12 +339,12 @@ impl ImageTranscodeService {
tracing::warn!("Failed to persist transcode skip marker: {}", e);
}
});
- return Ok((original_content, original_mime.to_string(), false));
+ return Ok(Bytes::new());
}
let saved = original_size - transcoded_size;
- // ── 5. Persist to disk cache (fire-and-forget) ──
+ // ── Persist to disk cache (fire-and-forget) ──
let cache_path_clone = cache_path.clone();
let transcoded_for_disk = transcoded_bytes.clone();
tokio::spawn(async move {
@@ -324,12 +356,7 @@ impl ImageTranscodeService {
}
});
- // ── 6. Store in moka memory cache (lock-free) ──
- self.memory_cache
- .insert(cache_key, transcoded_bytes.clone())
- .await;
-
- // ── 7. Update stats (lock-free atomics) ──
+ // ── Update stats (lock-free atomics) ──
self.stats.transcodes.fetch_add(1, Ordering::Relaxed);
self.stats
.bytes_saved
@@ -343,11 +370,7 @@ impl ImageTranscodeService {
(1.0 - transcoded_size as f64 / original_size as f64) * 100.0
);
- Ok((
- transcoded_bytes,
- target_format.mime_type().to_string(),
- true,
- ))
+ Ok(transcoded_bytes)
}
/// Get path for cached transcoded file
diff --git a/src/infrastructure/services/mod.rs b/src/infrastructure/services/mod.rs
index 67c52e70..854ffa9c 100644
--- a/src/infrastructure/services/mod.rs
+++ b/src/infrastructure/services/mod.rs
@@ -3,6 +3,7 @@ pub mod azure_blob_backend;
pub mod cached_blob_backend;
pub mod chunked_upload_service;
pub mod compression_service;
+pub mod db_pool_monitor;
pub mod dedup_service;
pub mod encrypted_blob_backend;
pub mod exif_service;
diff --git a/src/main.rs b/src/main.rs
index 79270845..3e1c2cd8 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -1,5 +1,14 @@
#![allow(async_fn_in_trait)]
+// mimalloc returns freed pages to the OS only when `MIMALLOC_PURGE_DELAY` is
+// low/zero. With the default it retains them, so process RSS clamps at the peak
+// even after the in-memory caches (file content, thumbnails, transcode) expire
+// by TTL. The Dockerfile and docker-compose set `MIMALLOC_PURGE_DELAY=0` so RSS
+// tracks the live working set — benchmarked on musl/aarch64 at ~400 MB
+// reclaimed after a 400 MB alloc→free spike, vs 0 MB by default, at no
+// throughput cost. (jemalloc with `muzzy_decay_ms:0` is an equivalent
+// alternative; mimalloc+env is preferred on the musl/Alpine target — its
+// `background_thread` is unsupported there.)
#[global_allocator]
static GLOBAL: mimalloc::MiMalloc = mimalloc::MiMalloc;