Merge pull request #429 from AtalayaLabs/perf/auth-me-quota
perf(quota): stop recomputing storage usage on every GET /api/auth/me
This commit is contained in:
@@ -43,6 +43,13 @@ OXICLOUD_SERVER_HOST=127.0.0.1
|
||||
# the OS default.
|
||||
#OXICLOUD_UPLOAD_TMPDIR=/var/lib/oxicloud/tmp
|
||||
|
||||
# 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.
|
||||
|
||||
@@ -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;
|
||||
@@ -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<i64, DomainError> {
|
||||
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);
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
@@ -217,6 +217,11 @@ pub struct StorageConfig {
|
||||
/// memory limit and can trigger OOMKill on large files). Env:
|
||||
/// `OXICLOUD_UPLOAD_TMPDIR`.
|
||||
pub upload_temp_dir: Option<PathBuf>,
|
||||
/// 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`).
|
||||
@@ -355,6 +360,7 @@ impl Default for StorageConfig {
|
||||
trash_retention_days: 30, // 30 days
|
||||
max_upload_size: MAX_UPLOAD_SIZE,
|
||||
upload_temp_dir: None,
|
||||
usage_reconcile_secs: 600, // 10 minutes
|
||||
backend: StorageBackendType::Local,
|
||||
s3: None,
|
||||
azure: None,
|
||||
@@ -1227,6 +1233,14 @@ impl AppConfig {
|
||||
config.storage.upload_temp_dir = Some(PathBuf::from(dir.trim()));
|
||||
}
|
||||
|
||||
// Background storage-usage reconciliation interval
|
||||
if let Ok(secs) =
|
||||
env::var("OXICLOUD_STORAGE_USAGE_RECONCILE_SECS").map(|v| v.parse::<u64>())
|
||||
&& 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() {
|
||||
|
||||
@@ -634,6 +634,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
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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)
|
||||
|
||||
Reference in New Issue
Block a user