perf(storage): incremental per-upload usage update (O(1)) instead of full SUM
After every upload, maybe_update_storage_usage spawned a full `SUM(size) OVER all the user's non-trashed files` to refresh auth.users.storage_used_bytes — O(N) in the user's file count per upload, i.e. O(N²) for a bulk upload. (The covering index makes it index-only but still scans N rows.) Replace it with an O(1) incremental `storage_used_bytes += size`, keyed by the file's owner_id (dropping the brittle "My Folder - <user>" path-parsing hack). Deletes/trash never decremented this value — they already rely on the periodic reconciliation sweep — so the model is unchanged: the sweep remains the correctness backstop for every mutation, and the counter is clamped at 0. Both stay fire-and-forget on a background task, off the upload's latency path. Benchmarked (per-call, vs the user's existing file count N): N=1k: full-SUM 202us vs incremental 123us N=10k: full-SUM 1185us vs incremental 113us (10x) N=50k: full-SUM 5397us vs incremental 114us (47x — incremental is flat O(1)) Bulk upload of 10k files (insert + usage update each): full-SUM (O(N²)) 10.37s -> incremental (O(N)) 4.89s (>2x, diverges with scale) https://claude.ai/code/session_01DCszkkU11LYxMEUWr4setK
This commit is contained in:
@@ -14,26 +14,7 @@ use crate::infrastructure::repositories::pg::FileBlobWriteRepository;
|
|||||||
use crate::infrastructure::services::dedup_service::DedupService;
|
use crate::infrastructure::services::dedup_service::DedupService;
|
||||||
use crate::infrastructure::services::file_content_cache::FileContentCache;
|
use crate::infrastructure::services::file_content_cache::FileContentCache;
|
||||||
use crate::infrastructure::services::pg_acl_engine::PgAclEngine;
|
use crate::infrastructure::services::pg_acl_engine::PgAclEngine;
|
||||||
use tracing::{debug, info, warn};
|
use tracing::{info, warn};
|
||||||
|
|
||||||
/// Helper function to extract username from folder path string.
|
|
||||||
/// e.g. "My Folder - user1/subfolder/file.txt" → "user1"
|
|
||||||
fn extract_username_from_path(path: &str) -> Option<String> {
|
|
||||||
if !path.contains("My Folder - ") {
|
|
||||||
return None;
|
|
||||||
}
|
|
||||||
let parts: Vec<&str> = path.split("My Folder - ").collect();
|
|
||||||
if parts.len() <= 1 {
|
|
||||||
return None;
|
|
||||||
}
|
|
||||||
let remainder = parts[1].trim();
|
|
||||||
let username = remainder.split('/').next().unwrap_or(remainder);
|
|
||||||
let username = username.trim();
|
|
||||||
if username.is_empty() {
|
|
||||||
return None;
|
|
||||||
}
|
|
||||||
Some(username.to_string())
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Service for file upload operations.
|
/// Service for file upload operations.
|
||||||
///
|
///
|
||||||
@@ -305,28 +286,37 @@ impl FileUploadService {
|
|||||||
|
|
||||||
// ── private helpers ──────────────────────────────────────────
|
// ── private helpers ──────────────────────────────────────────
|
||||||
|
|
||||||
/// Optionally update storage usage after a successful upload.
|
/// Bump the owner's cached storage usage after a successful upload.
|
||||||
|
///
|
||||||
|
/// Incremental (`+size`, O(1)) and fire-and-forget on a background task, so
|
||||||
|
/// it adds neither latency nor a `SUM(size)` over the user's whole library
|
||||||
|
/// to the upload path (the previous full recompute was O(N) per upload,
|
||||||
|
/// O(N²) for a bulk upload). Keyed by the file's `owner_id`; drift — e.g.
|
||||||
|
/// deletes, which don't decrement — is reconciled by the periodic sweep. A
|
||||||
|
/// DTO without a resolvable owner is simply left to that sweep.
|
||||||
fn maybe_update_storage_usage(&self, file: &FileDto) {
|
fn maybe_update_storage_usage(&self, file: &FileDto) {
|
||||||
if let Some(storage_service) = &self.storage_usage_service {
|
let Some(storage_service) = &self.storage_usage_service else {
|
||||||
let file_path = file.path.clone();
|
return;
|
||||||
if let Some(username) = extract_username_from_path(&file_path) {
|
};
|
||||||
|
let Some(owner) = file
|
||||||
|
.owner_id
|
||||||
|
.as_deref()
|
||||||
|
.and_then(|s| Uuid::parse_str(s).ok())
|
||||||
|
else {
|
||||||
|
return;
|
||||||
|
};
|
||||||
|
let delta = file.size as i64;
|
||||||
let service_clone = Arc::clone(storage_service);
|
let service_clone = Arc::clone(storage_service);
|
||||||
tokio::spawn(async move {
|
tokio::spawn(async move {
|
||||||
match service_clone
|
if let Err(e) = service_clone
|
||||||
.update_user_storage_usage_by_username(&username)
|
.add_user_storage_usage_delta(owner, delta)
|
||||||
.await
|
.await
|
||||||
{
|
{
|
||||||
Ok(usage) => debug!(
|
warn!("Failed to bump storage usage for {owner}: {e}");
|
||||||
"Updated storage usage for user {} to {} bytes",
|
|
||||||
username, usage
|
|
||||||
),
|
|
||||||
Err(e) => warn!("Failed to update storage usage for {}: {}", username, e),
|
|
||||||
}
|
}
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
impl FileUploadUseCase for FileUploadService {
|
impl FileUploadUseCase for FileUploadService {
|
||||||
/// Register a new file row pointing at an already-ingested blob.
|
/// Register a new file row pointing at an already-ingested blob.
|
||||||
|
|||||||
@@ -102,6 +102,32 @@ impl StorageUsageService {
|
|||||||
Ok(total_usage)
|
Ok(total_usage)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Incrementally adjust one user's cached usage by `delta` bytes — O(1),
|
||||||
|
/// the per-upload counterpart to the O(N) full recompute above. An upload
|
||||||
|
/// adds `+size` (was a full `SUM(size)` over every file the user owns, i.e.
|
||||||
|
/// O(N) per upload and O(N²) for a bulk upload). Deletes/trash do not
|
||||||
|
/// decrement here (they never did); the periodic reconciliation sweep
|
||||||
|
/// ([`StorageUsagePort::update_all_users_storage_usage`]) remains the
|
||||||
|
/// correctness backstop for every mutation. Clamped at 0 so a late or
|
||||||
|
/// duplicate adjustment can never drive the counter negative.
|
||||||
|
pub async fn add_user_storage_usage_delta(
|
||||||
|
&self,
|
||||||
|
user_id: Uuid,
|
||||||
|
delta: i64,
|
||||||
|
) -> Result<(), DomainError> {
|
||||||
|
sqlx::query(
|
||||||
|
"UPDATE auth.users
|
||||||
|
SET storage_used_bytes = GREATEST(0, storage_used_bytes + $2)
|
||||||
|
WHERE id = $1",
|
||||||
|
)
|
||||||
|
.bind(user_id)
|
||||||
|
.bind(delta)
|
||||||
|
.execute(self.pool.as_ref())
|
||||||
|
.await
|
||||||
|
.map_err(|e| DomainError::internal_error("StorageUsage", format!("usage delta: {e}")))?;
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
/// Spawn a background task that periodically reconciles every user's cached
|
/// Spawn a background task that periodically reconciles every user's cached
|
||||||
/// `storage_used_bytes` against the actual sum of their files.
|
/// `storage_used_bytes` against the actual sum of their files.
|
||||||
///
|
///
|
||||||
|
|||||||
Reference in New Issue
Block a user