diff --git a/example.env b/example.env index bbff9ffb..f0da51cc 100644 --- a/example.env +++ b/example.env @@ -34,6 +34,13 @@ OXICLOUD_SERVER_HOST=127.0.0.1 # Maximum upload size in bytes (default: 10 GB on 64-bit) #OXICLOUD_MAX_UPLOAD_SIZE=10737418240 +# How often (seconds) the background sweep reconciles each user's cached +# storage usage with the real sum of their files (default: 600 = 10 min). +# GET /api/auth/me serves the cached value instead of recomputing per request; +# this sweep keeps it fresh for deletes/trash too. Lower = fresher quota, +# higher = less background DB work. Minimum enforced: 30s. +#OXICLOUD_STORAGE_USAGE_RECONCILE_SECS=600 + # Allow multiple processes to bind to the same port (SO_REUSEPORT). # DISABLED by default — leaving this off means a second accidental instance # will fail immediately with "address already in use", which is the safe behaviour. diff --git a/migrations/20260625000000_files_user_size_index.sql b/migrations/20260625000000_files_user_size_index.sql new file mode 100644 index 00000000..d4d3cd46 --- /dev/null +++ b/migrations/20260625000000_files_user_size_index.sql @@ -0,0 +1,18 @@ +-- Covering partial index for per-user storage-usage accounting. +-- +-- The usage calculation is: +-- SELECT COALESCE(SUM(size), 0) FROM storage.files +-- WHERE user_id = $1 AND NOT is_trashed; +-- +-- Without an index that carries `size`, this is a heap scan over every file +-- the user owns. This index lets PostgreSQL satisfy it with an index-only scan: +-- * keyed by user_id → only the target user's rows are visited +-- * INCLUDE (size) → the sum is read straight from the index +-- * WHERE NOT is_trashed → matches the query predicate exactly and keeps +-- the index small (trashed files are excluded) +-- +-- Used by the per-upload usage update and the periodic background +-- reconciliation sweep (GET /api/auth/me no longer recomputes usage inline). +CREATE INDEX IF NOT EXISTS idx_files_user_size_active + ON storage.files (user_id) INCLUDE (size) + WHERE NOT is_trashed; diff --git a/src/application/services/storage_usage_service.rs b/src/application/services/storage_usage_service.rs index 3aed2399..ced34892 100644 --- a/src/application/services/storage_usage_service.rs +++ b/src/application/services/storage_usage_service.rs @@ -52,7 +52,12 @@ impl StorageUsageService { } /// Calculates a user's storage usage by summing all their file sizes. - /// Uses a direct SQL query for O(1) performance. + /// + /// This is `SUM(size)` over the user's non-trashed files — O(number of + /// files), backed by the `idx_files_user_size_active` covering partial + /// index so it runs as an index-only scan. It is NOT called on the request + /// path; only by the per-upload update and the background reconciliation + /// sweep. async fn calculate_user_storage_usage(&self, user_id: Uuid) -> Result { debug!("Calculating storage for user: {}", user_id); @@ -105,6 +110,37 @@ impl StorageUsageService { Ok(total_usage) } + + /// Spawn a background task that periodically reconciles every user's cached + /// `storage_used_bytes` against the actual sum of their files. + /// + /// `GET /api/auth/me` no longer recomputes usage on the request path; this + /// sweep (plus the per-upload update) keeps the cached value current for + /// all mutations — including deletes and trash — without any O(N) work on a + /// hot endpoint. Runs on the maintenance pool. The first sweep is deferred + /// by one interval so it never adds load at boot. + pub fn start_reconciliation_job(&self, interval_secs: u64) { + // Floor the interval so a misconfiguration can't busy-loop the sweep. + let interval_secs = interval_secs.max(30); + let service = self.clone(); + info!( + "Starting storage-usage reconciliation job (every {}s)", + interval_secs + ); + task::spawn(async move { + let mut ticker = tokio::time::interval(std::time::Duration::from_secs(interval_secs)); + // tokio's first `tick()` fires immediately — consume it so the + // first real sweep happens one interval after startup. + ticker.tick().await; + loop { + ticker.tick().await; + debug!("Running scheduled storage-usage reconciliation"); + if let Err(e) = service.update_all_users_storage_usage().await { + error!("Scheduled storage-usage reconciliation failed: {}", e); + } + } + }); + } } /** diff --git a/src/common/config.rs b/src/common/config.rs index 5a3963f9..079de563 100644 --- a/src/common/config.rs +++ b/src/common/config.rs @@ -211,6 +211,11 @@ pub struct StorageConfig { /// Maximum upload file size in bytes (default: 10 GB). /// Applied as a hard limit to WebDAV PUT and streaming uploads. pub max_upload_size: usize, + /// Interval (seconds) of the background sweep that reconciles every user's + /// cached `storage_used_bytes` with the real sum of their files. Keeps the + /// quota fresh for all mutations without recomputing on the request path. + /// Default: 600 (10 min). Env: `OXICLOUD_STORAGE_USAGE_RECONCILE_SECS`. + pub usage_reconcile_secs: u64, /// Which blob storage backend to use (`local`, `s3`, or `azure`). pub backend: StorageBackendType, /// S3-compatible backend configuration (used when `backend == S3`). @@ -348,6 +353,7 @@ impl Default for StorageConfig { parallel_threshold: 100 * 1024 * 1024, // 100 MB trash_retention_days: 30, // 30 days max_upload_size: MAX_UPLOAD_SIZE, + usage_reconcile_secs: 600, // 10 minutes backend: StorageBackendType::Local, s3: None, azure: None, @@ -1212,6 +1218,14 @@ impl AppConfig { config.storage.max_upload_size = val; } + // Background storage-usage reconciliation interval + if let Ok(secs) = + env::var("OXICLOUD_STORAGE_USAGE_RECONCILE_SECS").map(|v| v.parse::()) + && let Ok(val) = secs + { + config.storage.usage_reconcile_secs = val; + } + // Storage backend selection if let Ok(backend) = env::var("OXICLOUD_STORAGE_BACKEND") { match backend.to_lowercase().as_str() { diff --git a/src/common/di.rs b/src/common/di.rs index 6a7a0e71..32d17b59 100644 --- a/src/common/di.rs +++ b/src/common/di.rs @@ -633,6 +633,10 @@ impl AppServiceFactory { user_repository, ), ); + // Keep cached storage usage fresh off the request path: GET /api/auth/me + // no longer recomputes the O(N) SUM per call; a periodic sweep does it + // instead (on the maintenance pool). + service.start_reconciliation_job(self.config.storage.usage_reconcile_secs); tracing::info!("Storage usage service initialized"); service } diff --git a/src/infrastructure/repositories/pg/user_pg_repository.rs b/src/infrastructure/repositories/pg/user_pg_repository.rs index c64c7f9a..82132fad 100644 --- a/src/infrastructure/repositories/pg/user_pg_repository.rs +++ b/src/infrastructure/repositories/pg/user_pg_repository.rs @@ -411,7 +411,11 @@ impl UserRepository for UserPgRepository { Ok(user) } - /// Updates only the storage usage of a user + /// Updates only the storage usage of a user. + /// + /// The `IS DISTINCT FROM` guard makes this a no-op when the value is + /// unchanged — which is the common case for the periodic reconciliation + /// sweep — so it produces no dead tuple and no WAL when nothing changed. async fn update_storage_usage( &self, user_id: Uuid, @@ -420,10 +424,10 @@ impl UserRepository for UserPgRepository { sqlx::query( r#" UPDATE auth.users - SET + SET storage_used_bytes = $2, updated_at = NOW() - WHERE id = $1 + WHERE id = $1 AND storage_used_bytes IS DISTINCT FROM $2 "#, ) .bind(user_id) diff --git a/src/interfaces/api/handlers/auth_handler.rs b/src/interfaces/api/handlers/auth_handler.rs index 0388eebc..14ccc22d 100644 --- a/src/interfaces/api/handlers/auth_handler.rs +++ b/src/interfaces/api/handlers/auth_handler.rs @@ -434,7 +434,7 @@ pub async fn refresh_token( Ok(response) } -/// Return the authenticated user's profile, including live storage usage. +/// Return the authenticated user's profile, including cached storage usage. #[utoipa::path( get, path = "/api/auth/me", @@ -454,29 +454,13 @@ pub async fn get_current_user( .as_ref() .ok_or_else(|| AppError::internal_error("Authentication service not configured"))?; - // First, update the storage usage statistics - // IMPORTANT: We await the calculation to return updated data - if let Some(storage_usage_service) = state.storage_usage_service.as_ref() { - // Calculate storage synchronously (we await the result) - match storage_usage_service - .update_user_storage_usage(user_id) - .await - { - Ok(usage) => { - tracing::info!( - "Updated storage usage for user {}: {} bytes", - user_id, - usage - ); - } - Err(e) => { - // Only log a warning, don't fail the entire request - tracing::warn!("Failed to update storage usage for user {}: {}", user_id, e); - } - } - } - - // Now get the user data WITH the updated storage + // Storage usage is served from the cached `storage_used_bytes` column — + // it is NOT recomputed here. Recomputing on this hot endpoint meant an + // O(N) `SUM(size)` over all the user's files plus an `UPDATE` of + // `auth.users` on every single call (one of the most frequent endpoints). + // The cached value is kept current by the per-upload update and a periodic + // background reconciliation sweep + // (see `StorageUsageService::start_reconciliation_job`). let user = auth_service .auth_application_service .get_user_by_id(user_id) diff --git a/src/interfaces/api/handlers/carddav_handler.rs b/src/interfaces/api/handlers/carddav_handler.rs index 3524605a..3a6d273e 100644 --- a/src/interfaces/api/handlers/carddav_handler.rs +++ b/src/interfaces/api/handlers/carddav_handler.rs @@ -141,12 +141,10 @@ fn strip_username_prefix(path: &str) -> &str { } else { &path[pos + 1..] } + } else if uuid::Uuid::parse_str(path).is_ok() { + path } else { - if uuid::Uuid::parse_str(path).is_ok() { - path - } else { - "" - } + "" } }