perf: cache-stampede coalescing + DB safeguards; ui/i18n fixes
Backend — tail latency & throughput:
- FileContentCache, image transcode, and search now use moka single-flight
(try_get_with / get_or_load) so N concurrent misses for the same key
collapse to one disk read / transcode / query instead of a thundering herd.
Microbenchmark (128 concurrent on one hot key): 128 loads / p99 ~1023ms
before vs 1 load / p99 ~32ms after.
- DB: configurable per-statement timeout on the primary pool
(OXICLOUD_DB_STATEMENT_TIMEOUT_SECS, default 30; maintenance pool exempt) so
a runaway query can't pin a connection and starve the pool.
- DB: background pool-saturation monitor
(OXICLOUD_DB_POOL_MONITOR_INTERVAL_SECS) that WARNs as the primary pool nears
exhaustion — the early signal before tail latency cliffs.
- mimalloc: set MIMALLOC_PURGE_DELAY=0 (Dockerfile + compose) so freed pages
return to the OS and RSS tracks the live working set; benchmarked on
musl/aarch64 at ~400MB reclaimed vs 0MB with the default.
Frontend — UI / i18n fixes:
- i18n: fix literal "{{count}}" and "{{percentage}}/{{used}}/{{total}}" in the
selection toolbar and storage line — the call sites passed param names that
didn't match the locale placeholders; unify on `count` and pass the storage
template its params. Add es files.selected_count.
- sidebar: hide the drive picker when there's only one drive (the redundant
"Personal" row); remove the coloured left accent on the active nav item.
- logo: stop clipping the cloud's left bulge — viewBox recentred on the cloud's
true bbox with proportional SVG size so it keeps the same rendered scale.
- user menu: drop the default <a> underline on the link rows.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
DioCrafts
@@ -92,6 +92,14 @@ COPY --from=builder --chown=oxicloud:oxicloud /app/static-dist /app/static
|
||||
# Create storage directory with proper permissions
|
||||
RUN mkdir -p /app/storage && chown -R oxicloud:oxicloud /app/storage
|
||||
|
||||
# Allocator tuning — make RSS track the live working set.
|
||||
# mimalloc retains freed pages by default, so process RSS clamps at the peak
|
||||
# even after the in-memory caches (file content, thumbnails, transcode) expire
|
||||
# by TTL. Purging immediately returns those pages to the kernel at no throughput
|
||||
# cost. Measured on this musl/aarch64 image: a 400 MB alloc→free spike returns
|
||||
# 0 MB by default vs ~400 MB with this set (idle RSS back to a few MB).
|
||||
ENV MIMALLOC_PURGE_DELAY=0
|
||||
|
||||
# Set working directory
|
||||
WORKDIR /app
|
||||
|
||||
|
||||
@@ -21,6 +21,14 @@ services:
|
||||
oxicloud:
|
||||
image: diocrafts/oxicloud:latest
|
||||
restart: always
|
||||
environment:
|
||||
# Return freed heap pages to the OS promptly. mimalloc retains them by
|
||||
# default, so idle RSS clamps at the peak even after the in-memory caches
|
||||
# expire; purging immediately makes RSS track the live working set
|
||||
# (benchmarked: ~400 MB reclaimed on musl/aarch64) at no throughput cost.
|
||||
# Also baked into the Dockerfile; set here so it applies to the prebuilt
|
||||
# image too.
|
||||
MIMALLOC_PURGE_DELAY: "0"
|
||||
build:
|
||||
context: .
|
||||
dockerfile: Dockerfile
|
||||
|
||||
@@ -261,7 +261,7 @@
|
||||
<div class="sidebar" class:open={sidebarOpen}>
|
||||
<a href="/files" class="logo-container">
|
||||
<div class="logo">
|
||||
<svg viewBox="120 120 280 280" aria-hidden="true">
|
||||
<svg viewBox="95 67 320 320" aria-hidden="true">
|
||||
<path
|
||||
d="M345 310c32 0 58-26 58-58s-26-58-58-58c-6.2 0-12 0.9-17.5 2.7C318 166 289 143 255 143c-34.3 0-63.1 22.6-73 53.7C176.9 195.7 171 195 165 195c-32 0-58 26-58 58s26 58 58 58h180z"
|
||||
/>
|
||||
@@ -302,7 +302,7 @@
|
||||
session.user.storage_quota_bytes
|
||||
)}
|
||||
{:else}
|
||||
{formatBytes(session.user.storage_used_bytes)} {t('storage.used', 'used')}
|
||||
{formatBytes(session.user.storage_used_bytes)}
|
||||
{/if}
|
||||
</div>
|
||||
</div>
|
||||
@@ -526,7 +526,19 @@
|
||||
<div class="user-menu-storage-fill" style:width="{storagePct}%"></div>
|
||||
</div>
|
||||
<div class="user-menu-storage-text">
|
||||
{Math.round(storagePct)}% {t('storage.used', 'used')}
|
||||
{#if session.user.storage_quota_bytes > 0}
|
||||
{t(
|
||||
'storage.used',
|
||||
{
|
||||
percentage: Math.round(storagePct),
|
||||
used: formatBytes(session.user.storage_used_bytes),
|
||||
total: formatBytes(session.user.storage_quota_bytes)
|
||||
},
|
||||
'{{percentage}}% used ({{used}} / {{total}})'
|
||||
)}
|
||||
{:else}
|
||||
{formatBytes(session.user.storage_used_bytes)}
|
||||
{/if}
|
||||
</div>
|
||||
</div>
|
||||
{/if}
|
||||
@@ -661,7 +673,7 @@
|
||||
<div class="about-overlay" onclick={(e) => e.target === e.currentTarget && (aboutOpen = false)}>
|
||||
<div class="about-modal" role="dialog" aria-modal="true" aria-labelledby="about-modal-title">
|
||||
<div class="about-modal__logo">
|
||||
<svg viewBox="120 120 280 280" aria-hidden="true">
|
||||
<svg viewBox="95 67 320 320" aria-hidden="true">
|
||||
<path
|
||||
d="M345 310c32 0 58-26 58-58s-26-58-58-58c-6.2 0-12 0.9-17.5 2.7C318 166 289 143 255 143c-34.3 0-63.1 22.6-73 53.7C176.9 195.7 171 195 165 195c-32 0-58 26-58 58s26 58 58 58h180z"
|
||||
/>
|
||||
@@ -1071,8 +1083,10 @@
|
||||
}
|
||||
|
||||
.about-modal__logo {
|
||||
width: 64px;
|
||||
height: 64px;
|
||||
/* 73px (not 64) so the cloud keeps its rendered scale after the viewBox
|
||||
grew 280→320 to stop clipping its left bulge: 73/320 ≈ 64/280. */
|
||||
width: 73px;
|
||||
height: 73px;
|
||||
color: var(--color-accent);
|
||||
}
|
||||
|
||||
|
||||
@@ -63,7 +63,11 @@
|
||||
});
|
||||
</script>
|
||||
|
||||
{#if drivesStore.loaded && drivesStore.drives.length > 0}
|
||||
<!-- Only show the drive switcher when there's an actual choice to make. With a
|
||||
single drive (the default personal one) the picker just repeats "Personal"
|
||||
under the Files nav row, so hide it; it reappears the moment a second drive
|
||||
(e.g. a shared one) exists. -->
|
||||
{#if drivesStore.loaded && drivesStore.drives.length > 1}
|
||||
<ul class="drive-picker" aria-label={t('drive.picker', 'Drives')}>
|
||||
{#each sortedDrives as d (d.id)}
|
||||
<li class="drive-picker__row" class:drive-picker__row--active={isActive(d)}>
|
||||
|
||||
@@ -91,8 +91,10 @@
|
||||
}
|
||||
|
||||
.logo svg {
|
||||
width: 22px;
|
||||
height: 22px;
|
||||
/* 25px (not 22) so the cloud keeps its rendered scale after the viewBox
|
||||
grew 280→320 to stop clipping its left bulge: 25/320 ≈ 22/280. */
|
||||
width: 25px;
|
||||
height: 25px;
|
||||
fill: var(--color-sidebar-text-active);
|
||||
}
|
||||
|
||||
@@ -147,13 +149,10 @@
|
||||
.nav-item.active {
|
||||
background-color: var(--color-sidebar-active-bg);
|
||||
color: var(--color-sidebar-text-active);
|
||||
border-left-color: var(--color-accent);
|
||||
font-weight: var(--weight-semibold);
|
||||
|
||||
[dir="rtl"] & {
|
||||
border-left-color: transparent;
|
||||
border-right-color: var(--color-accent);
|
||||
}
|
||||
/* No coloured left accent — the tinted background + bold weight already
|
||||
read as "active". The base `border-left: 3px solid transparent` is kept
|
||||
on every row so removing the accent causes no horizontal text shift. */
|
||||
}
|
||||
|
||||
.nav-item i,
|
||||
|
||||
@@ -177,6 +177,10 @@
|
||||
cursor: pointer;
|
||||
transition: background 0.15s ease;
|
||||
text-align: left;
|
||||
/* The link rows (`<a class="user-menu-item">` — admin panel, groups,
|
||||
profile) are styled as menu rows, not hyperlinks; drop the default
|
||||
<a> underline so they match the <button> rows. */
|
||||
text-decoration: none;
|
||||
}
|
||||
|
||||
.user-menu-item:hover,
|
||||
|
||||
@@ -1213,7 +1213,7 @@
|
||||
<Icon name="times" />
|
||||
</button>
|
||||
<span class="batch-bar-count"
|
||||
>{t('files.selected_count', { n: selectedCount }, '{{n}} selected')}</span
|
||||
>{t('files.selected_count', { count: selectedCount }, '{{count}} selected')}</span
|
||||
>
|
||||
</div>
|
||||
<div class="batch-selection-info">
|
||||
|
||||
@@ -456,7 +456,7 @@
|
||||
|
||||
{#if selected.size > 0}
|
||||
<div class="batch-bar">
|
||||
<span>{t('files.selected_count', { n: selected.size }, '{{n}} selected')}</span>
|
||||
<span>{t('files.selected_count', { count: selected.size }, '{{count}} selected')}</span>
|
||||
<div class="batch-bar__actions">
|
||||
<Button onclick={downloadSelected}>{t('common.download', 'Download')}</Button>
|
||||
<Button onclick={() => selected.clear()}>{t('common.clear', 'Clear')}</Button>
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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<Bytes, DomainError> {
|
||||
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<str> = format!("\"{}\"", cache_key).into();
|
||||
let ct: Arc<str> = 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
|
||||
|
||||
@@ -193,16 +193,6 @@ impl SearchService {
|
||||
hasher.finish()
|
||||
}
|
||||
|
||||
/// Attempts to retrieve results from the cache.
|
||||
async fn get_from_cache(&self, key: u64) -> Option<Arc<SearchResultsDto>> {
|
||||
self.search_cache.get(&key).await
|
||||
}
|
||||
|
||||
/// Stores results in the cache.
|
||||
async fn store_in_cache(&self, key: u64, results: Arc<SearchResultsDto>) {
|
||||
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<Arc<SearchResultsDto>> {
|
||||
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<FileDto> = files.into_iter().map(FileDto::from).collect();
|
||||
let mut enriched_files: Vec<SearchFileResultDto> = file_dtos
|
||||
.iter()
|
||||
.map(|f| Self::enrich_file(f, &query_lower))
|
||||
.collect();
|
||||
// Convert to DTOs and enrich with metadata
|
||||
let file_dtos: Vec<FileDto> = files.into_iter().map(FileDto::from).collect();
|
||||
let mut enriched_files: Vec<SearchFileResultDto> = 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<FolderDto> =
|
||||
folders.into_iter().map(FolderDto::from).collect();
|
||||
|
||||
// For folders, apply sorting and pagination in memory (usually fewer folders)
|
||||
let mut enriched_folders: Vec<SearchFolderResultDto> = 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<Folder> = 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<FileDto> = found_files.into_iter().map(FileDto::from).collect();
|
||||
let mut enriched_files: Vec<SearchFileResultDto> = file_dtos
|
||||
.iter()
|
||||
.map(|f| Self::enrich_file(f, &query_lower))
|
||||
.collect();
|
||||
|
||||
let folder_dtos: Vec<FolderDto> =
|
||||
found_folders.into_iter().map(FolderDto::from).collect();
|
||||
let mut enriched_folders: Vec<SearchFolderResultDto> = 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>| {
|
||||
crate::common::errors::DomainError::new(
|
||||
shared.kind,
|
||||
shared.entity_type,
|
||||
shared.message.clone(),
|
||||
)
|
||||
.await?;
|
||||
|
||||
let filtered_folders: Vec<FolderDto> =
|
||||
folders.into_iter().map(FolderDto::from).collect();
|
||||
|
||||
// For folders, apply sorting and pagination in memory (usually fewer folders)
|
||||
let mut enriched_folders: Vec<SearchFolderResultDto> = 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<Folder> = 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<FileDto> = found_files.into_iter().map(FileDto::from).collect();
|
||||
let mut enriched_files: Vec<SearchFileResultDto> = file_dtos
|
||||
.iter()
|
||||
.map(|f| Self::enrich_file(f, &query_lower))
|
||||
.collect();
|
||||
|
||||
let folder_dtos: Vec<FolderDto> = found_folders.into_iter().map(FolderDto::from).collect();
|
||||
let mut enriched_folders: Vec<SearchFolderResultDto> = 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.
|
||||
|
||||
@@ -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::<u64>())
|
||||
&& 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::<u64>())
|
||||
&& let Ok(val) = interval
|
||||
{
|
||||
config.database.pool_monitor_interval_secs = val;
|
||||
}
|
||||
|
||||
// Auth configuration
|
||||
if let Some(jwt_secret) = env::var("OXICLOUD_JWT_SECRET")
|
||||
.ok()
|
||||
|
||||
@@ -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<PgPool>) {
|
||||
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;
|
||||
|
||||
@@ -45,6 +45,7 @@ pub async fn create_database_pools(config: &AppConfig) -> Result<DbPools> {
|
||||
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<DbPools> {
|
||||
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<DbPools> {
|
||||
}
|
||||
|
||||
/// 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<PgPool> {
|
||||
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);
|
||||
|
||||
@@ -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<AtomicU32>,
|
||||
}
|
||||
|
||||
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));
|
||||
}
|
||||
}
|
||||
@@ -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<F>(
|
||||
&self,
|
||||
cache_key: String,
|
||||
etag: Arc<str>,
|
||||
content_type: Arc<str>,
|
||||
load: F,
|
||||
) -> Result<(Bytes, Arc<str>, Arc<str>), DomainError>
|
||||
where
|
||||
F: Future<Output = Result<Bytes, DomainError>>,
|
||||
{
|
||||
// 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, DomainError>(CacheEntry {
|
||||
content,
|
||||
etag,
|
||||
content_type,
|
||||
})
|
||||
})
|
||||
.await
|
||||
// try_get_with hands back `Arc<DomainError>` 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>| {
|
||||
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<Semaphore>,
|
||||
loads: Arc<AtomicUsize>,
|
||||
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"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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<String>` across waiters; DomainError
|
||||
// here is just a String, so hand callers an owned clone.
|
||||
.map_err(|shared: Arc<String>| (*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<Bytes, String> {
|
||||
// ── 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
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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;
|
||||
|
||||
|
||||
Reference in New Issue
Block a user