cjw/Oxicloud
1
0
Fork 0
Code Issues Pull Requests Actions 4 Packages Projects Releases Wiki Activity

fix: OOM protection, lock-free thumbnail cache, OIDC JWKS TTL

Browse Source 
- Streaming WebDAV PUT: body spooled to tempfile with incremental SHA-256,
  peak RAM ~64KB regardless of file size (Solution 2)
- RequestBodyLimitLayer (1MB) on CalDAV/CardDAV routers (Solution 3)
- All body::to_bytes(body, usize::MAX) replaced with explicit limits:
  PROPFIND/PROPPATCH/LOCK → 1MB, MKCOL → 4KB
- Added AppError::payload_too_large (HTTP 413)
- Added max_upload_size to StorageConfig (default 10GB, env override)
- New streaming update chain: FileWritePort::update_file_content_from_temp
  → FileUploadUseCase::update_file_streaming
- ThumbnailService: migrated from RwLock<LruCache> to moka::future::Cache
  with weight-based eviction — eliminates lock contention on read hot-path
- OIDC: discovery + JWKS caches now expire after 1 hour (Cached<T> wrapper)
  so IdP key rotation no longer requires server restart
This commit is contained in:
Diocrafts
2026-02-22 23:28:03 +01:00
parent b48f2867ac
commit 92e0364a60
12 changed files with 395 additions and 172 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/infrastructure/repositories/pg/file_blob_write_repository.rs
+85 -54
View File
@@ -110,6 +110,74 @@ impl FileBlobWriteRepository {
)),
}
}
/// Atomically swap the blob hash of a file.
///
/// Uses a CTE to capture the old hash before updating so the old blob
/// reference can be decremented afterwards. Compensates on failure by
/// removing the new blob reference.
async fn swap_blob_hash(
&self,
file_id: &str,
new_hash: &str,
new_size: i64,
) -> Result<(), DomainError> {
// Atomic CTE: capture old hash then update in one round-trip, no TOCTOU.
let old_hash = match sqlx::query_scalar::<_, String>(
r#"
WITH old AS (
SELECT id, blob_hash FROM storage.files WHERE id = $3::uuid FOR UPDATE
)
UPDATE storage.files f
SET blob_hash = $1, size = $2, updated_at = NOW()
FROM old
WHERE f.id = old.id
RETURNING old.blob_hash
"#,
)
.bind(new_hash)
.bind(new_size)
.bind(file_id)
.fetch_optional(self.pool.as_ref())
.await
{
Ok(Some(old)) => old,
Ok(None) => {
// File not found — compensate: remove the new blob ref
if let Err(e) = self.dedup.remove_reference(new_hash).await {
tracing::error!("Blob orphaned after missing file: {}", e);
}
return Err(DomainError::not_found("File", file_id));
}
Err(e) => {
// UPDATE failed — compensate: remove the new blob ref
if let Err(rollback_err) = self.dedup.remove_reference(new_hash).await {
tracing::error!(
"Blob orphaned after failed UPDATE — hash: {}, err: {}",
&new_hash[..12],
rollback_err
);
}
return Err(DomainError::internal_error(
"FileBlobWrite",
format!("update: {e}"),
));
}
};
// Decrement old blob ref (only if hash changed, best-effort)
if old_hash != new_hash {
if let Err(e) = self.dedup.remove_reference(&old_hash).await {
tracing::warn!(
"Failed to decrement old blob ref {}: {}",
&old_hash[..12],
e
);
}
}
Ok(())
}
}
#[async_trait]
@@ -476,62 +544,25 @@ impl FileWritePort for FileBlobWriteRepository {
let dedup_result = self.dedup.store_bytes(&content, None).await?;
let new_hash = dedup_result.hash().to_string();
// Atomic CTE: capture old hash then update in one round-trip, no TOCTOU.
// The `old` CTE locks + reads the row *before* the update touches it.
let old_hash = match sqlx::query_scalar::<_, String>(
r#"
WITH old AS (
SELECT id, blob_hash FROM storage.files WHERE id = $3::uuid FOR UPDATE
)
UPDATE storage.files f
SET blob_hash = $1, size = $2, updated_at = NOW()
FROM old
WHERE f.id = old.id
RETURNING old.blob_hash
"#,
)
.bind(&new_hash)
.bind(new_size)
.bind(file_id)
.fetch_optional(self.pool.as_ref())
.await
{
Ok(Some(old)) => old,
Ok(None) => {
// File not found — compensate: remove the new blob ref
if let Err(e) = self.dedup.remove_reference(&new_hash).await {
tracing::error!("Blob orphaned after missing file: {}", e);
}
return Err(DomainError::not_found("File", file_id));
}
Err(e) => {
// UPDATE failed — compensate: remove the new blob ref
if let Err(rollback_err) = self.dedup.remove_reference(&new_hash).await {
tracing::error!(
"Blob orphaned after failed UPDATE — hash: {}, err: {}",
&new_hash[..12],
rollback_err
);
}
return Err(DomainError::internal_error(
"FileBlobWrite",
format!("update: {e}"),
));
}
};
self.swap_blob_hash(file_id, &new_hash, new_size).await
}
// Decrement old blob ref (only if hash changed, best-effort)
if old_hash != new_hash
&& let Err(e) = self.dedup.remove_reference(&old_hash).await
{
tracing::warn!(
"Failed to decrement old blob ref {}: {}",
&old_hash[..12],
e
);
}
async fn update_file_content_from_temp(
&self,
file_id: &str,
temp_path: &std::path::Path,
size: u64,
content_type: Option<String>,
pre_computed_hash: Option<String>,
) -> Result<(), DomainError> {
// Streaming: pass pre-computed hash so dedup skips re-reading the file.
let dedup_result = self
.dedup
.store_from_file(temp_path, content_type, pre_computed_hash)
.await?;
let new_hash = dedup_result.hash().to_string();
Ok(())
self.swap_blob_hash(file_id, &new_hash, size as i64).await
}
async fn register_file_deferred(
src/infrastructure/services/oidc_service.rs
+47 -16
View File
@@ -8,11 +8,16 @@
use async_trait::async_trait;
use serde::Deserialize;
use std::sync::RwLock;
use std::time::{Duration, Instant};
use crate::application::ports::auth_ports::{OidcIdClaims, OidcServicePort, OidcTokenSet};
use crate::common::config::OidcConfig;
use crate::common::errors::{DomainError, ErrorKind};
/// How long discovery/JWKS documents stay cached before re-fetching.
/// 1 hour balances freshness against unnecessary network requests.
const OIDC_CACHE_TTL: Duration = Duration::from_secs(3600);
// ============================================================================
// OIDC Discovery Document
// ============================================================================
@@ -92,13 +97,33 @@ struct UserInfoResponse {
// OIDC Service
// ============================================================================
/// A cached value with a fetch timestamp for TTL-based expiry.
#[derive(Clone)]
struct Cached<T: Clone> {
value: T,
fetched_at: Instant,
}
impl<T: Clone> Cached<T> {
fn new(value: T) -> Self {
Self {
value,
fetched_at: Instant::now(),
}
}
fn is_expired(&self) -> bool {
self.fetched_at.elapsed() > OIDC_CACHE_TTL
}
}
pub struct OidcService {
config: OidcConfig,
http_client: reqwest::Client,
/// Cached discovery document
discovery: RwLock<Option<OidcDiscovery>>,
/// Cached JWKS (typed JWK keys)
jwks: RwLock<Option<JwksDocument>>,
/// Cached discovery document (expires after OIDC_CACHE_TTL)
discovery: RwLock<Option<Cached<OidcDiscovery>>>,
/// Cached JWKS (expires after OIDC_CACHE_TTL)
jwks: RwLock<Option<Cached<JwksDocument>>>,
}
impl OidcService {
@@ -116,16 +141,19 @@ impl OidcService {
}
}
/// Fetch and cache the OIDC discovery document
/// Fetch and cache the OIDC discovery document (TTL: 1 hour)
async fn get_discovery(&self) -> Result<OidcDiscovery, DomainError> {
// Check cache first
// Check cache first (return cached value only if not expired)
{
let cache = self
.discovery
.read()
.map_err(|_| DomainError::new(ErrorKind::InternalError, "OIDC", "Lock poisoned"))?;
if let Some(ref disc) = *cache {
return Ok(disc.clone());
if let Some(ref cached) = *cache {
if !cached.is_expired() {
return Ok(cached.value.clone());
}
tracing::debug!("OIDC discovery cache expired, re-fetching");
}
}
@@ -164,28 +192,31 @@ impl OidcService {
)
})?;
// Cache it
// Cache it with timestamp
{
let mut cache = self
.discovery
.write()
.map_err(|_| DomainError::new(ErrorKind::InternalError, "OIDC", "Lock poisoned"))?;
*cache = Some(discovery.clone());
*cache = Some(Cached::new(discovery.clone()));
}
Ok(discovery)
}
/// Fetch and cache JWKS document for ID token validation
/// Fetch and cache JWKS document for ID token validation (TTL: 1 hour)
async fn get_jwks(&self) -> Result<JwksDocument, DomainError> {
// Check cache first
// Check cache first (return cached value only if not expired)
{
let cache = self
.jwks
.read()
.map_err(|_| DomainError::new(ErrorKind::InternalError, "OIDC", "Lock poisoned"))?;
if let Some(ref jwks) = *cache {
return Ok(jwks.clone());
if let Some(ref cached) = *cache {
if !cached.is_expired() {
return Ok(cached.value.clone());
}
tracing::debug!("OIDC JWKS cache expired, re-fetching");
}
}
@@ -214,13 +245,13 @@ impl OidcService {
)
})?;
// Cache it
// Cache it with timestamp
{
let mut cache = self
.jwks
.write()
.map_err(|_| DomainError::new(ErrorKind::InternalError, "OIDC", "Lock poisoned"))?;
*cache = Some(jwks.clone());
*cache = Some(Cached::new(jwks.clone()));
}
Ok(jwks)
src/infrastructure/services/thumbnail_service.rs
+33 -64
View File
@@ -1,8 +1,6 @@
use async_trait::async_trait;
use bytes::Bytes;
use image::{ImageFormat, imageops::FilterType};
use lru::LruCache;
use std::num::NonZeroUsize;
/**
* Thumbnail Generation Service
*
@@ -12,13 +10,12 @@ use std::num::NonZeroUsize;
* - Background thumbnail generation after upload
* - Multiple sizes (icon 150x150, preview 800x600)
* - WebP output for smaller file sizes
* - LRU cache for hot thumbnails
* - Lock-free moka cache with weight-based eviction
* - Lazy generation on first request if not pre-generated
*/
use std::path::{Path, PathBuf};
use std::sync::Arc;
use tokio::fs;
use tokio::sync::RwLock;
use crate::application::ports::thumbnail_ports::{
ThumbnailPort, ThumbnailSize as PortThumbnailSize, ThumbnailStatsDto,
@@ -76,12 +73,10 @@ struct ThumbnailCacheKey {
pub struct ThumbnailService {
/// Root path for thumbnail storage
thumbnails_root: PathBuf,
/// In-memory LRU cache for hot thumbnails
cache: Arc<RwLock<LruCache<ThumbnailCacheKey, Bytes>>>,
/// Maximum cache size in bytes
max_cache_bytes: usize,
/// Current cache size in bytes
current_cache_bytes: Arc<RwLock<usize>>,
/// Lock-free concurrent cache (moka) with weight-based eviction
cache: moka::future::Cache<ThumbnailCacheKey, Bytes>,
/// Configured maximum cache weight (for stats reporting)
max_cache_bytes: u64,
}
impl ThumbnailService {
@@ -89,18 +84,27 @@ impl ThumbnailService {
///
/// # Arguments
/// * `storage_root` - Root path of file storage
/// * `max_cache_entries` - Maximum number of thumbnails to cache in memory
/// * `max_cache_entries` - (ignored — moka uses weight-based eviction)
/// * `max_cache_bytes` - Maximum total bytes to cache
pub fn new(storage_root: &Path, max_cache_entries: usize, max_cache_bytes: usize) -> Self {
let thumbnails_root = storage_root.join(".thumbnails");
// Ignore max_cache_entries — weight-based eviction is more accurate
// for variable-size thumbnails than entry-count limits.
let _ = max_cache_entries;
let cache = moka::future::Cache::builder()
.max_capacity(max_cache_bytes as u64)
.weigher(|_key: &ThumbnailCacheKey, value: &Bytes| -> u32 {
value.len().min(u32::MAX as usize) as u32
})
.time_to_live(std::time::Duration::from_secs(600))
.build();
Self {
thumbnails_root,
cache: Arc::new(RwLock::new(LruCache::new(
NonZeroUsize::new(max_cache_entries).unwrap_or(NonZeroUsize::new(1000).unwrap()),
))),
max_cache_bytes,
current_cache_bytes: Arc::new(RwLock::new(0)),
cache,
max_cache_bytes: max_cache_bytes as u64,
}
}
@@ -158,13 +162,10 @@ impl ThumbnailService {
size,
};
// Check in-memory cache first
{
let cache = self.cache.read().await;
if let Some(data) = cache.peek(&cache_key) {
tracing::debug!("🔥 Thumbnail cache HIT: {} {:?}", file_id, size);
return Ok(data.clone());
}
// Check lock-free cache first
if let Some(data) = self.cache.get(&cache_key).await {
tracing::debug!("🔥 Thumbnail cache HIT: {} {:?}", file_id, size);
return Ok(data);
}
// Check if thumbnail exists on disk
@@ -177,8 +178,8 @@ impl ThumbnailService {
.map_err(|e| ThumbnailError::IoError(e.to_string()))?;
let bytes = Bytes::from(data);
// Add to cache
self.add_to_cache(cache_key, bytes.clone()).await;
// Add to cache (lock-free insert — moka handles eviction)
self.cache.insert(cache_key, bytes.clone()).await;
tracing::debug!("💾 Thumbnail loaded from disk: {} {:?}", file_id, size);
return Ok(bytes);
@@ -198,8 +199,8 @@ impl ThumbnailService {
.await
.map_err(|e| ThumbnailError::IoError(e.to_string()))?;
// Add to cache
self.add_to_cache(cache_key, bytes.clone()).await;
// Add to cache (lock-free insert)
self.cache.insert(cache_key, bytes.clone()).await;
Ok(bytes)
}
@@ -245,31 +246,6 @@ impl ThumbnailService {
result.map(Bytes::from)
}
/// Add a thumbnail to the in-memory cache
async fn add_to_cache(&self, key: ThumbnailCacheKey, data: Bytes) {
let data_size = data.len();
// Check if adding this would exceed max cache size
let mut current_size = self.current_cache_bytes.write().await;
// Evict items if needed to make room
if *current_size + data_size > self.max_cache_bytes {
let mut cache = self.cache.write().await;
while *current_size + data_size > self.max_cache_bytes && !cache.is_empty() {
if let Some((_, evicted)) = cache.pop_lru() {
*current_size = current_size.saturating_sub(evicted.len());
}
}
}
// Add to cache
let mut cache = self.cache.write().await;
if let Some(old) = cache.put(key, data) {
*current_size = current_size.saturating_sub(old.len());
}
*current_size += data_size;
}
/// Generate all thumbnail sizes for a file in the background
///
/// This is called after file upload to pre-generate thumbnails
@@ -316,16 +292,12 @@ impl ThumbnailService {
.map_err(|e| ThumbnailError::IoError(e.to_string()))?;
}
// Remove from cache
// Remove from cache (lock-free invalidation)
let cache_key = ThumbnailCacheKey {
file_id: file_id.to_string(),
size: *size,
};
let mut cache = self.cache.write().await;
if let Some(removed) = cache.pop(&cache_key) {
let mut current_size = self.current_cache_bytes.write().await;
*current_size = current_size.saturating_sub(removed.len());
}
self.cache.invalidate(&cache_key).await;
}
tracing::debug!("🗑️ Deleted thumbnails for: {}", file_id);
@@ -334,13 +306,10 @@ impl ThumbnailService {
/// Get cache statistics
pub async fn get_stats(&self) -> ThumbnailStats {
let cache = self.cache.read().await;
let current_size = *self.current_cache_bytes.read().await;
ThumbnailStats {
cached_thumbnails: cache.len(),
cache_size_bytes: current_size,
max_cache_bytes: self.max_cache_bytes,
cached_thumbnails: self.cache.entry_count() as usize,
cache_size_bytes: self.cache.weighted_size() as usize,
max_cache_bytes: self.max_cache_bytes as usize,
}
}
}