perf: bulk-delete expired trash in 2 SQL queries instead of N+1 loop
This commit is contained in:
@@ -96,15 +96,13 @@ impl TrashRepository for MockTrashRepository {
|
|||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
async fn get_expired_items(&self) -> Result<Vec<TrashedItem>> {
|
async fn delete_expired_bulk(&self) -> Result<(u64, u64)> {
|
||||||
let items = self.trash_items.lock().unwrap();
|
let mut items = self.trash_items.lock().unwrap();
|
||||||
let now = Utc::now();
|
let now = Utc::now();
|
||||||
let expired = items
|
let before = items.len() as u64;
|
||||||
.values()
|
items.retain(|_, item| item.deletion_date() > now);
|
||||||
.filter(|item| item.deletion_date() <= now)
|
let deleted = before - items.len() as u64;
|
||||||
.cloned()
|
Ok((deleted, 0))
|
||||||
.collect();
|
|
||||||
Ok(expired)
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
+8
-12
@@ -36,10 +36,10 @@ use crate::application::services::{
|
|||||||
use crate::common::config::AppConfig;
|
use crate::common::config::AppConfig;
|
||||||
use crate::common::errors::DomainError;
|
use crate::common::errors::DomainError;
|
||||||
use crate::domain::services::i18n_service::I18nService;
|
use crate::domain::services::i18n_service::I18nService;
|
||||||
use crate::infrastructure::repositories::pg::SharePgRepository;
|
|
||||||
use crate::infrastructure::repositories::pg::{
|
use crate::infrastructure::repositories::pg::{
|
||||||
FileBlobReadRepository, FileBlobWriteRepository, FolderDbRepository, TrashDbRepository,
|
FileBlobReadRepository, FileBlobWriteRepository, FolderDbRepository, TrashDbRepository,
|
||||||
};
|
};
|
||||||
|
use crate::infrastructure::repositories::pg::SharePgRepository;
|
||||||
use crate::infrastructure::services::file_content_cache::{
|
use crate::infrastructure::services::file_content_cache::{
|
||||||
FileContentCache, FileContentCacheConfig,
|
FileContentCache, FileContentCacheConfig,
|
||||||
};
|
};
|
||||||
@@ -306,9 +306,8 @@ impl AppServiceFactory {
|
|||||||
self.config.storage.trash_retention_days,
|
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(
|
let cleanup_service = TrashCleanupService::new(
|
||||||
service.clone(),
|
|
||||||
trash_repo.clone(),
|
trash_repo.clone(),
|
||||||
24, // Run cleanup every 24 hours
|
24, // Run cleanup every 24 hours
|
||||||
);
|
);
|
||||||
@@ -334,13 +333,11 @@ impl AppServiceFactory {
|
|||||||
|
|
||||||
// Build a password hasher for share password verification
|
// Build a password hasher for share password verification
|
||||||
let password_hasher: Arc<dyn crate::application::ports::auth_ports::PasswordHasherPort> =
|
let password_hasher: Arc<dyn crate::application::ports::auth_ports::PasswordHasherPort> =
|
||||||
Arc::new(
|
Arc::new(crate::infrastructure::services::password_hasher::Argon2PasswordHasher::new(
|
||||||
crate::infrastructure::services::password_hasher::Argon2PasswordHasher::new(
|
self.config.auth.hash_memory_cost,
|
||||||
self.config.auth.hash_memory_cost,
|
self.config.auth.hash_time_cost,
|
||||||
self.config.auth.hash_time_cost,
|
self.config.auth.hash_parallelism,
|
||||||
self.config.auth.hash_parallelism,
|
));
|
||||||
),
|
|
||||||
);
|
|
||||||
|
|
||||||
let service = Arc::new(ShareService::new(
|
let service = Arc::new(ShareService::new(
|
||||||
Arc::new(self.config.clone()),
|
Arc::new(self.config.clone()),
|
||||||
@@ -472,8 +469,7 @@ impl AppServiceFactory {
|
|||||||
recent_service = Some(recent.clone());
|
recent_service = Some(recent.clone());
|
||||||
apps.recent_service = Some(recent);
|
apps.recent_service = Some(recent);
|
||||||
|
|
||||||
storage_usage_service =
|
storage_usage_service = Some(self.create_storage_usage_service(&repos, &pool, &maintenance_pool));
|
||||||
Some(self.create_storage_usage_service(&repos, &pool, &maintenance_pool));
|
|
||||||
|
|
||||||
// Auth services
|
// Auth services
|
||||||
if self.config.features.enable_auth {
|
if self.config.features.enable_auth {
|
||||||
|
|||||||
@@ -12,5 +12,8 @@ pub trait TrashRepository: Send + Sync {
|
|||||||
async fn restore_from_trash(&self, id: &Uuid, user_id: &Uuid) -> Result<()>;
|
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 delete_permanently(&self, id: &Uuid, user_id: &Uuid) -> Result<()>;
|
||||||
async fn clear_trash(&self, user_id: &Uuid) -> Result<()>;
|
async fn clear_trash(&self, user_id: &Uuid) -> Result<()>;
|
||||||
async fn get_expired_items(&self) -> Result<Vec<TrashedItem>>;
|
|
||||||
|
/// 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)>;
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -147,27 +147,42 @@ impl TrashRepository for TrashDbRepository {
|
|||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
async fn get_expired_items(&self) -> Result<Vec<TrashedItem>> {
|
async fn delete_expired_bulk(&self) -> Result<(u64, u64)> {
|
||||||
let cutoff = Utc::now() - chrono::Duration::days(self.retention_days);
|
let cutoff = Utc::now() - chrono::Duration::days(self.retention_days);
|
||||||
|
|
||||||
let rows = sqlx::query_as::<_, (Uuid, String, String, String, Option<DateTime<Utc>>)>(
|
let mut tx = self.pool.begin().await.map_err(|e| {
|
||||||
r#"
|
DomainError::internal_error("TrashDb", format!("begin tx: {e}"))
|
||||||
SELECT id, name, item_type, user_id, trashed_at
|
})?;
|
||||||
FROM storage.trash_items
|
|
||||||
WHERE trashed_at < $1
|
// 1. Bulk-delete expired trashed files.
|
||||||
ORDER BY trashed_at ASC
|
// 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)
|
.bind(cutoff)
|
||||||
.fetch_all(self.pool.as_ref())
|
.execute(&mut *tx)
|
||||||
.await
|
.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
|
// 2. Bulk-delete expired trashed folders.
|
||||||
.into_iter()
|
// FK ON DELETE CASCADE handles descendant folders and their files.
|
||||||
.map(|(id, name, item_type, uid, trashed_at)| {
|
let folders_deleted = sqlx::query(
|
||||||
self.row_to_trashed_item(id, name, item_type, uid, trashed_at)
|
"DELETE FROM storage.folders WHERE is_trashed = TRUE AND trashed_at < $1",
|
||||||
})
|
)
|
||||||
.collect())
|
.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))
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -3,25 +3,25 @@ use std::time::Duration;
|
|||||||
use tokio::time;
|
use tokio::time;
|
||||||
use tracing::{debug, error, info, instrument};
|
use tracing::{debug, error, info, instrument};
|
||||||
|
|
||||||
use crate::application::ports::trash_ports::TrashUseCase;
|
|
||||||
use crate::common::errors::Result;
|
use crate::common::errors::Result;
|
||||||
use crate::domain::repositories::trash_repository::TrashRepository;
|
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 {
|
pub struct TrashCleanupService {
|
||||||
trash_service: Arc<dyn TrashUseCase>,
|
|
||||||
trash_repository: Arc<dyn TrashRepository>,
|
trash_repository: Arc<dyn TrashRepository>,
|
||||||
cleanup_interval_hours: u64,
|
cleanup_interval_hours: u64,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl TrashCleanupService {
|
impl TrashCleanupService {
|
||||||
pub fn new(
|
pub fn new(
|
||||||
trash_service: Arc<dyn TrashUseCase>,
|
|
||||||
trash_repository: Arc<dyn TrashRepository>,
|
trash_repository: Arc<dyn TrashRepository>,
|
||||||
cleanup_interval_hours: u64,
|
cleanup_interval_hours: u64,
|
||||||
) -> Self {
|
) -> Self {
|
||||||
Self {
|
Self {
|
||||||
trash_service,
|
|
||||||
trash_repository,
|
trash_repository,
|
||||||
cleanup_interval_hours: cleanup_interval_hours.max(1), // Minimum 1 hour
|
cleanup_interval_hours: cleanup_interval_hours.max(1), // Minimum 1 hour
|
||||||
}
|
}
|
||||||
@@ -31,7 +31,6 @@ impl TrashCleanupService {
|
|||||||
#[instrument(skip(self))]
|
#[instrument(skip(self))]
|
||||||
pub async fn start_cleanup_job(&self) {
|
pub async fn start_cleanup_job(&self) {
|
||||||
let trash_repository = self.trash_repository.clone();
|
let trash_repository = self.trash_repository.clone();
|
||||||
let trash_service = self.trash_service.clone();
|
|
||||||
let interval_hours = self.cleanup_interval_hours;
|
let interval_hours = self.cleanup_interval_hours;
|
||||||
|
|
||||||
info!(
|
info!(
|
||||||
@@ -44,7 +43,7 @@ impl TrashCleanupService {
|
|||||||
let mut interval = time::interval(interval_duration);
|
let mut interval = time::interval(interval_duration);
|
||||||
|
|
||||||
// First immediate execution
|
// First immediate execution
|
||||||
Self::cleanup_expired_items(trash_repository.clone(), trash_service.clone())
|
Self::cleanup_expired_items(trash_repository.clone())
|
||||||
.await
|
.await
|
||||||
.unwrap_or_else(|e| error!("Error in initial trash cleanup: {:?}", e));
|
.unwrap_or_else(|e| error!("Error in initial trash cleanup: {:?}", e));
|
||||||
|
|
||||||
@@ -52,50 +51,31 @@ impl TrashCleanupService {
|
|||||||
interval.tick().await;
|
interval.tick().await;
|
||||||
debug!("Running scheduled trash cleanup task");
|
debug!("Running scheduled trash cleanup task");
|
||||||
|
|
||||||
if let Err(e) =
|
if let Err(e) = Self::cleanup_expired_items(trash_repository.clone()).await {
|
||||||
Self::cleanup_expired_items(trash_repository.clone(), trash_service.clone())
|
|
||||||
.await
|
|
||||||
{
|
|
||||||
error!("Error in scheduled trash cleanup: {:?}", e);
|
error!("Error in scheduled trash cleanup: {:?}", e);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Cleans up expired items in the trash
|
/// Bulk-delete all expired trash items in a single transaction.
|
||||||
#[instrument(skip(trash_repository, trash_service))]
|
#[instrument(skip(trash_repository))]
|
||||||
async fn cleanup_expired_items(
|
async fn cleanup_expired_items(
|
||||||
trash_repository: Arc<dyn TrashRepository>,
|
trash_repository: Arc<dyn TrashRepository>,
|
||||||
trash_service: Arc<dyn TrashUseCase>,
|
|
||||||
) -> Result<()> {
|
) -> Result<()> {
|
||||||
debug!("Starting cleanup of expired items in the trash");
|
debug!("Starting bulk cleanup of expired trash items");
|
||||||
|
|
||||||
// Get all expired items
|
let (files, folders) = trash_repository.delete_expired_bulk().await?;
|
||||||
let expired_items = trash_repository.get_expired_items().await?;
|
|
||||||
|
|
||||||
if expired_items.is_empty() {
|
if files == 0 && folders == 0 {
|
||||||
debug!("No expired items to clean up");
|
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(())
|
Ok(())
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user