cjw/Oxicloud
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

fix: prevent blob storage leak on folder deletion

Browse Source 
- Add PG trigger trg_files_decrement_blob_ref (AFTER DELETE ON storage.files)
  that auto-decrements storage.blobs.ref_count for every deleted file row.
  Covers all paths: explicit DELETE, ON DELETE CASCADE, trash emptying.

- Remove manual remove_reference() call from delete_file() in
  file_blob_write_repository — trigger is now the single source of truth.

- Fix double-decrement bug in FileManagementService::delete_with_cleanup:
  was decrementing ref_count on trash (soft-delete) when the file row still
  existed, causing premature blob GC and potential data corruption on restore.

- Remove dead fields (file_read, dedup_service) from FileManagementService
  and simplify constructors — ref_count fully handled by PG trigger.
This commit is contained in:
Diocrafts
2026-02-22 22:07:46 +01:00
parent 35cfbba335
commit 2dd3dc0b54
5 changed files with 867 additions and 91 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/application/services/file_management_service.rs
+19 -75
View File
@@ -2,94 +2,40 @@ use async_trait::async_trait;
use std::sync::Arc;
use crate::application::dtos::file_dto::FileDto;
use crate::application::ports::dedup_ports::DedupPort;
use crate::application::ports::file_ports::FileManagementUseCase;
use crate::application::ports::storage_ports::{FileReadPort, FileWritePort};
use crate::application::ports::storage_ports::FileWritePort;
use crate::application::ports::trash_ports::TrashUseCase;
use crate::common::errors::DomainError;
use tracing::{debug, error, info, warn};
use tracing::{error, info, warn};
/// Service for file management operations (move, delete).
///
/// The `delete_with_cleanup` method internalises:
/// 1. Blob-hash lookup for dedup tracking (O(1) DB read)
/// 2. Trash-first soft-delete
/// 3. Fallback to permanent delete
/// 4. Dedup reference-count decrement
/// Blob ref_count bookkeeping on deletion is handled by the PG trigger
/// `trg_files_decrement_blob_ref` (fires on DELETE FROM storage.files).
/// This service only orchestrates trash vs. permanent delete — it never
/// touches ref_count directly.
pub struct FileManagementService {
file_repository: Arc<dyn FileWritePort>,
file_read: Option<Arc<dyn FileReadPort>>,
trash_service: Option<Arc<dyn TrashUseCase>>,
dedup_service: Option<Arc<dyn DedupPort>>,
}
impl FileManagementService {
/// Backward-compatible constructor (no trash, no dedup).
/// Creates a new FileManagementService.
pub fn new(file_repository: Arc<dyn FileWritePort>) -> Self {
Self {
file_repository,
file_read: None,
trash_service: None,
dedup_service: None,
}
}
/// Full constructor with trash + dedup ports.
pub fn new_full(
/// Creates a FileManagementService with a trash service.
pub fn with_trash(
file_repository: Arc<dyn FileWritePort>,
file_read: Arc<dyn FileReadPort>,
trash_service: Option<Arc<dyn TrashUseCase>>,
dedup_service: Arc<dyn DedupPort>,
) -> Self {
Self {
file_repository,
file_read: Some(file_read),
trash_service,
dedup_service: Some(dedup_service),
}
}
/// Setter for late-bound trash service.
pub fn with_trash_service(mut self, trash_service: Arc<dyn TrashUseCase>) -> Self {
self.trash_service = Some(trash_service);
self
}
// ── private helpers ──────────────────────────────────────────
/// Look up the blob hash from the database. Returns `None` when
/// dedup is inactive or the file is not found.
///
/// This is O(1) — a single `SELECT blob_hash` by primary key.
/// It replaces the old `compute_content_hash` which loaded the
/// entire file into RAM just to re-derive the same hash.
async fn lookup_blob_hash(&self, id: &str) -> Option<String> {
self.dedup_service.as_ref()?; // dedup must be active
let file_read = self.file_read.as_ref()?;
match file_read.get_blob_hash(id).await {
Ok(hash) => {
info!("🔗 DEDUP: File {} has blob hash: {}", id, &hash[..12]);
Some(hash)
}
Err(e) => {
warn!("Could not get blob hash for dedup: {}", e);
None
}
}
}
/// Decrement dedup reference count; log result.
async fn decrement_dedup_ref(&self, hash: &str) {
let Some(dedup) = &self.dedup_service else {
return;
};
match dedup.remove_reference(hash).await {
Ok(true) => info!(
"🗑️ DEDUP: Blob {} deleted (no more references)",
&hash[..12]
),
Ok(false) => debug!("🔗 DEDUP: Reference removed from blob {}", &hash[..12]),
Err(e) => warn!("⚠️ DEDUP: Failed to decrement reference: {}", e),
}
}
}
@@ -180,19 +126,21 @@ impl FileManagementUseCase for FileManagementService {
}
/// Smart delete: trash-first with dedup reference cleanup.
///
/// Blob ref_count bookkeeping is handled entirely by the PG trigger
/// `trg_files_decrement_blob_ref` which fires on DELETE FROM storage.files.
/// We do NOT decrement here — trashing is a soft-delete (UPDATE, not DELETE)
/// so the blob must remain referenced until the file is permanently deleted.
async fn delete_with_cleanup(&self, id: &str, user_id: &str) -> Result<bool, DomainError> {
// Step 1: Look up blob hash for dedup tracking (O(1) DB read)
let content_hash = self.lookup_blob_hash(id).await;
// Step 2: Try trash (soft delete)
// Step 1: Try trash (soft delete — file row stays, blob stays referenced)
if let Some(trash) = &self.trash_service {
info!("Moving file to trash: {}", id);
match trash.move_to_trash(id, "file", user_id).await {
Ok(_) => {
info!("File successfully moved to trash: {}", id);
if let Some(hash) = &content_hash {
self.decrement_dedup_ref(hash).await;
}
// Do NOT decrement blob ref here — the file row still exists
// (is_trashed = TRUE). The trigger will decrement when the
// row is actually DELETEd during trash emptying.
return Ok(true); // trashed
}
Err(err) => {
@@ -205,15 +153,11 @@ impl FileManagementUseCase for FileManagementService {
warn!("Trash service not available, using permanent delete");
}
// Step 3: Permanent delete
// Step 2: Permanent delete — trigger handles blob ref_count
warn!("Permanently deleting file: {}", id);
self.file_repository.delete_file(id).await?;
info!("File permanently deleted: {}", id);
if let Some(hash) = &content_hash {
self.decrement_dedup_ref(hash).await;
}
Ok(false) // permanently deleted
}
}
src/common/di.rs
+2 -4
View File
@@ -241,12 +241,10 @@ impl AppServiceFactory {
core.image_transcode_service.clone(),
));
// FileManagementService with dedup and trash
let file_management_service = Arc::new(FileManagementService::new_full(
// FileManagementService — ref_count handled by PG trigger, no dedup port needed
let file_management_service = Arc::new(FileManagementService::with_trash(
repos.file_write_repository.clone(),
repos.file_read_repository.clone(),
trash_service.clone(),
core.dedup_service.clone(),
));
let file_use_case_factory = Arc::new(AppFileUseCaseFactory::new(
src/infrastructure/repositories/pg/file_blob_write_repository.rs
+11 -12
View File
@@ -449,19 +449,18 @@ impl FileWritePort for FileBlobWriteRepository {
}
async fn delete_file(&self, id: &str) -> Result<(), DomainError> {
// Atomic DELETE RETURNING — one round-trip instead of SELECT + DELETE
let hash = sqlx::query_scalar::<_, String>(
"DELETE FROM storage.files WHERE id = $1::uuid RETURNING blob_hash",
)
.bind(id)
.fetch_optional(self.pool.as_ref())
.await
.map_err(|e| DomainError::internal_error("FileBlobWrite", format!("delete: {e}")))?
.ok_or_else(|| DomainError::not_found("File", id))?;
// The PG trigger `trg_files_decrement_blob_ref` automatically
// decrements storage.blobs.ref_count for the deleted row's blob_hash.
// Disk cleanup of orphaned blobs (ref_count = 0) is handled by
// garbage_collect().
let result = sqlx::query("DELETE FROM storage.files WHERE id = $1::uuid")
.bind(id)
.execute(self.pool.as_ref())
.await
.map_err(|e| DomainError::internal_error("FileBlobWrite", format!("delete: {e}")))?;
// Decrement blob reference (best-effort after successful DELETE)
if let Err(e) = self.dedup.remove_reference(&hash).await {
tracing::warn!("Failed to decrement blob ref for {}: {}", &hash[..12], e);
if result.rows_affected() == 0 {
return Err(DomainError::not_found("File", id));
}
Ok(())