From 7e278a0b32f91f1cf04c3ed1b348bef308d7d911 Mon Sep 17 00:00:00 2001 From: Dionisio Date: Thu, 26 Feb 2026 00:42:41 +0100 Subject: [PATCH] perf: bulk-delete expired trash in 2 SQL queries instead of N+1 loop --- .../services/trash_service_test.rs | 14 +++-- src/common/di.rs | 20 +++---- src/domain/repositories/trash_repository.rs | 5 +- .../repositories/pg/trash_db_repository.rs | 47 ++++++++++------ .../services/trash_cleanup_service.rs | 54 ++++++------------- 5 files changed, 66 insertions(+), 74 deletions(-) diff --git a/src/application/services/trash_service_test.rs b/src/application/services/trash_service_test.rs index 216445ea..02aaa37f 100644 --- a/src/application/services/trash_service_test.rs +++ b/src/application/services/trash_service_test.rs @@ -96,15 +96,13 @@ impl TrashRepository for MockTrashRepository { Ok(()) } - async fn get_expired_items(&self) -> Result> { - let items = self.trash_items.lock().unwrap(); + async fn delete_expired_bulk(&self) -> Result<(u64, u64)> { + let mut items = self.trash_items.lock().unwrap(); let now = Utc::now(); - let expired = items - .values() - .filter(|item| item.deletion_date() <= now) - .cloned() - .collect(); - Ok(expired) + let before = items.len() as u64; + items.retain(|_, item| item.deletion_date() > now); + let deleted = before - items.len() as u64; + Ok((deleted, 0)) } } diff --git a/src/common/di.rs b/src/common/di.rs index a86442e4..3d20b256 100644 --- a/src/common/di.rs +++ b/src/common/di.rs @@ -36,10 +36,10 @@ use crate::application::services::{ use crate::common::config::AppConfig; use crate::common::errors::DomainError; use crate::domain::services::i18n_service::I18nService; -use crate::infrastructure::repositories::pg::SharePgRepository; use crate::infrastructure::repositories::pg::{ FileBlobReadRepository, FileBlobWriteRepository, FolderDbRepository, TrashDbRepository, }; +use crate::infrastructure::repositories::pg::SharePgRepository; use crate::infrastructure::services::file_content_cache::{ FileContentCache, FileContentCacheConfig, }; @@ -306,9 +306,8 @@ impl AppServiceFactory { self.config.storage.trash_retention_days, )); - // Initialize cleanup service + // Initialize cleanup service (bulk-deletes expired items in 2 SQL queries) let cleanup_service = TrashCleanupService::new( - service.clone(), trash_repo.clone(), 24, // Run cleanup every 24 hours ); @@ -334,13 +333,11 @@ impl AppServiceFactory { // Build a password hasher for share password verification let password_hasher: Arc = - Arc::new( - crate::infrastructure::services::password_hasher::Argon2PasswordHasher::new( - self.config.auth.hash_memory_cost, - self.config.auth.hash_time_cost, - self.config.auth.hash_parallelism, - ), - ); + Arc::new(crate::infrastructure::services::password_hasher::Argon2PasswordHasher::new( + self.config.auth.hash_memory_cost, + self.config.auth.hash_time_cost, + self.config.auth.hash_parallelism, + )); let service = Arc::new(ShareService::new( Arc::new(self.config.clone()), @@ -472,8 +469,7 @@ impl AppServiceFactory { recent_service = Some(recent.clone()); apps.recent_service = Some(recent); - storage_usage_service = - Some(self.create_storage_usage_service(&repos, &pool, &maintenance_pool)); + storage_usage_service = Some(self.create_storage_usage_service(&repos, &pool, &maintenance_pool)); // Auth services if self.config.features.enable_auth { diff --git a/src/domain/repositories/trash_repository.rs b/src/domain/repositories/trash_repository.rs index dfe7fe12..8cfff5af 100644 --- a/src/domain/repositories/trash_repository.rs +++ b/src/domain/repositories/trash_repository.rs @@ -12,5 +12,8 @@ pub trait TrashRepository: Send + Sync { async fn restore_from_trash(&self, id: &Uuid, user_id: &Uuid) -> Result<()>; async fn delete_permanently(&self, id: &Uuid, user_id: &Uuid) -> Result<()>; async fn clear_trash(&self, user_id: &Uuid) -> Result<()>; - async fn get_expired_items(&self) -> Result>; + + /// Bulk-delete all expired trash items (files + folders) in a single + /// transaction. Returns `(files_deleted, folders_deleted)`. + async fn delete_expired_bulk(&self) -> Result<(u64, u64)>; } diff --git a/src/infrastructure/repositories/pg/trash_db_repository.rs b/src/infrastructure/repositories/pg/trash_db_repository.rs index ea8bc65a..5c7655f4 100644 --- a/src/infrastructure/repositories/pg/trash_db_repository.rs +++ b/src/infrastructure/repositories/pg/trash_db_repository.rs @@ -147,27 +147,42 @@ impl TrashRepository for TrashDbRepository { Ok(()) } - async fn get_expired_items(&self) -> Result> { + async fn delete_expired_bulk(&self) -> Result<(u64, u64)> { let cutoff = Utc::now() - chrono::Duration::days(self.retention_days); - let rows = sqlx::query_as::<_, (Uuid, String, String, String, Option>)>( - r#" - SELECT id, name, item_type, user_id, trashed_at - FROM storage.trash_items - WHERE trashed_at < $1 - ORDER BY trashed_at ASC - "#, + let mut tx = self.pool.begin().await.map_err(|e| { + DomainError::internal_error("TrashDb", format!("begin tx: {e}")) + })?; + + // 1. Bulk-delete expired trashed files. + // The PG trigger `trg_files_decrement_blob_ref` automatically + // decrements blob ref_count for every deleted row. + let files_deleted = sqlx::query( + "DELETE FROM storage.files WHERE is_trashed = TRUE AND trashed_at < $1", ) .bind(cutoff) - .fetch_all(self.pool.as_ref()) + .execute(&mut *tx) .await - .map_err(|e| DomainError::internal_error("TrashDb", format!("expired: {e}")))?; + .map_err(|e| DomainError::internal_error("TrashDb", format!("bulk delete files: {e}")))? + .rows_affected(); - Ok(rows - .into_iter() - .map(|(id, name, item_type, uid, trashed_at)| { - self.row_to_trashed_item(id, name, item_type, uid, trashed_at) - }) - .collect()) + // 2. Bulk-delete expired trashed folders. + // FK ON DELETE CASCADE handles descendant folders and their files. + let folders_deleted = sqlx::query( + "DELETE FROM storage.folders WHERE is_trashed = TRUE AND trashed_at < $1", + ) + .bind(cutoff) + .execute(&mut *tx) + .await + .map_err(|e| { + DomainError::internal_error("TrashDb", format!("bulk delete folders: {e}")) + })? + .rows_affected(); + + tx.commit().await.map_err(|e| { + DomainError::internal_error("TrashDb", format!("commit tx: {e}")) + })?; + + Ok((files_deleted, folders_deleted)) } } diff --git a/src/infrastructure/services/trash_cleanup_service.rs b/src/infrastructure/services/trash_cleanup_service.rs index b756a5f4..5fbcbe1a 100644 --- a/src/infrastructure/services/trash_cleanup_service.rs +++ b/src/infrastructure/services/trash_cleanup_service.rs @@ -3,25 +3,25 @@ use std::time::Duration; use tokio::time; use tracing::{debug, error, info, instrument}; -use crate::application::ports::trash_ports::TrashUseCase; use crate::common::errors::Result; use crate::domain::repositories::trash_repository::TrashRepository; -/// Service for automatic cleanup of expired items in the trash +/// Service for automatic cleanup of expired items in the trash. +/// +/// Uses `TrashRepository::delete_expired_bulk` to purge all expired items +/// in **2 SQL statements inside a single transaction**, instead of the +/// previous N+1 pattern that issued 3 queries per expired item. pub struct TrashCleanupService { - trash_service: Arc, trash_repository: Arc, cleanup_interval_hours: u64, } impl TrashCleanupService { pub fn new( - trash_service: Arc, trash_repository: Arc, cleanup_interval_hours: u64, ) -> Self { Self { - trash_service, trash_repository, cleanup_interval_hours: cleanup_interval_hours.max(1), // Minimum 1 hour } @@ -31,7 +31,6 @@ impl TrashCleanupService { #[instrument(skip(self))] pub async fn start_cleanup_job(&self) { let trash_repository = self.trash_repository.clone(); - let trash_service = self.trash_service.clone(); let interval_hours = self.cleanup_interval_hours; info!( @@ -44,7 +43,7 @@ impl TrashCleanupService { let mut interval = time::interval(interval_duration); // First immediate execution - Self::cleanup_expired_items(trash_repository.clone(), trash_service.clone()) + Self::cleanup_expired_items(trash_repository.clone()) .await .unwrap_or_else(|e| error!("Error in initial trash cleanup: {:?}", e)); @@ -52,50 +51,31 @@ impl TrashCleanupService { interval.tick().await; debug!("Running scheduled trash cleanup task"); - if let Err(e) = - Self::cleanup_expired_items(trash_repository.clone(), trash_service.clone()) - .await - { + if let Err(e) = Self::cleanup_expired_items(trash_repository.clone()).await { error!("Error in scheduled trash cleanup: {:?}", e); } } }); } - /// Cleans up expired items in the trash - #[instrument(skip(trash_repository, trash_service))] + /// Bulk-delete all expired trash items in a single transaction. + #[instrument(skip(trash_repository))] async fn cleanup_expired_items( trash_repository: Arc, - trash_service: Arc, ) -> Result<()> { - debug!("Starting cleanup of expired items in the trash"); + debug!("Starting bulk cleanup of expired trash items"); - // Get all expired items - let expired_items = trash_repository.get_expired_items().await?; + let (files, folders) = trash_repository.delete_expired_bulk().await?; - if expired_items.is_empty() { + if files == 0 && folders == 0 { debug!("No expired items to clean up"); - return Ok(()); + } else { + info!( + "Trash cleanup completed: {} files + {} folders purged", + files, folders + ); } - info!("Found {} expired items to delete", expired_items.len()); - - // Delete each expired item - for item in expired_items { - let trash_id = item.id().to_string(); - let user_id = item.user_id().to_string(); - - debug!("Deleting expired item: id={}, user={}", trash_id, user_id); - - // If a deletion fails, continue with the rest - if let Err(e) = trash_service.delete_permanently(&trash_id, &user_id).await { - error!("Error deleting expired item {}: {:?}", trash_id, e); - } else { - debug!("Expired item deleted successfully: {}", trash_id); - } - } - - info!("Trash cleanup completed"); Ok(()) } }