diff --git a/src/application/ports/blob_reference_ports.rs b/src/application/ports/blob_reference_ports.rs index 38b7da61..aa949ba3 100644 --- a/src/application/ports/blob_reference_ports.rs +++ b/src/application/ports/blob_reference_ports.rs @@ -99,6 +99,19 @@ pub trait BlobReferenceSource: Send + Sync { /// by a request; no fragment may interpolate caller input. fn ref_count_sql(&self, level: RefLevel, outer_hash_expr: &str) -> Option; + /// Existence form of [`Self::ref_count_sql`] — a boolean fragment, true + /// when this source holds at least one reference at `level`. + /// + /// Defaults to `() > 0`. Override when the source can express a + /// short-circuiting `EXISTS`, which the planner can stop at the first + /// matching row: `dedup_gc`'s reap predicate runs this per candidate + /// manifest, and a heavily-deduplicated blob has many referrers, so + /// counting all of them where existence would do is a real regression. + fn ref_exists_sql(&self, level: RefLevel, outer_hash_expr: &str) -> Option { + self.ref_count_sql(level, outer_hash_expr) + .map(|fragment| format!("{fragment} > 0")) + } + /// Count of references this source holds on `blob_hash`, across both /// levels. /// @@ -171,6 +184,28 @@ impl BlobReferenceRegistry { } } + /// Predicate selecting rows that **no** registered source references at + /// `level` — i.e. reap candidates. + /// + /// Returns `None` when no source contributes at this level, and callers + /// **must** treat that as "refuse to act" rather than substituting a + /// default. The natural default would be the sum-equals-zero form, which + /// on an empty registry reduces to `0 = 0` — vacuously true for every + /// row, i.e. "delete everything". Returning `None` makes that + /// unrepresentable at the call site instead of merely discouraged. + pub fn no_reference_predicate(&self, level: RefLevel, outer_hash_expr: &str) -> Option { + let fragments: Vec = self + .sources + .iter() + .filter_map(|s| s.ref_exists_sql(level, outer_hash_expr)) + .collect(); + + if fragments.is_empty() { + return None; + } + Some(format!("NOT ({})", fragments.join("\n OR "))) + } + /// Total references held on `hash` across every source. /// /// On-demand path only — see [`BlobReferenceSource::count_references`]. diff --git a/src/infrastructure/repositories/pg/blob_reference_sources.rs b/src/infrastructure/repositories/pg/blob_reference_sources.rs index fc7e5aee..4424671b 100644 --- a/src/infrastructure/repositories/pg/blob_reference_sources.rs +++ b/src/infrastructure/repositories/pg/blob_reference_sources.rs @@ -51,6 +51,27 @@ fn files_ref_sql(level: RefLevel, outer_hash_expr: &str) -> Option { } } +/// Short-circuiting existence form of [`files_ref_sql`]. +/// +/// `dedup_gc` evaluates this per candidate manifest, so counting every +/// referrer where existence would do is a real cost on a heavily-deduplicated +/// blob. This is also the exact shape the reap predicate used before the +/// registry existed, so wiring it in changes no plan. +fn files_exists_sql(level: RefLevel, outer_hash_expr: &str) -> Option { + let f = FILES_ALIAS; + match level { + RefLevel::Chunk => Some(format!( + "EXISTS (SELECT 1 FROM storage.files {f} \ + WHERE {f}.blob_hash = {outer_hash_expr} \ + AND NOT EXISTS (SELECT 1 FROM storage.chunk_manifests {MANIFEST_ALIAS} \ + WHERE {MANIFEST_ALIAS}.file_hash = {f}.blob_hash))" + )), + RefLevel::Manifest => Some(format!( + "EXISTS (SELECT 1 FROM storage.files {f} WHERE {f}.blob_hash = {outer_hash_expr})" + )), + } +} + /// Fragment for [`ChunksReferenceSource`]. See [`files_ref_sql`]. fn chunks_ref_sql(level: RefLevel, outer_hash_expr: &str) -> Option { match level { @@ -99,6 +120,10 @@ impl BlobReferenceSource for FilesReferenceSource { files_ref_sql(level, outer_hash_expr) } + fn ref_exists_sql(&self, level: RefLevel, outer_hash_expr: &str) -> Option { + files_exists_sql(level, outer_hash_expr) + } + async fn count_references(&self, blob_hash: &str) -> Result { // No level split here: the question is "how many file rows name this // exact hash", and a hash names either a manifest or a legacy blob, diff --git a/src/infrastructure/services/dedup_service.rs b/src/infrastructure/services/dedup_service.rs index c8b84fa3..3f7b6a62 100644 --- a/src/infrastructure/services/dedup_service.rs +++ b/src/infrastructure/services/dedup_service.rs @@ -55,6 +55,7 @@ use std::sync::Arc; use tokio_util::io::StreamReader; use crate::application::ports::blob_lifecycle::BlobLifecycleHook; +use crate::application::ports::blob_reference_ports::{BlobReferenceRegistry, RefLevel}; use crate::application::ports::blob_storage_ports::BlobStorageBackend; use crate::application::ports::dedup_ports::{ BlobMetadataDto, DedupPort, DedupResultDto, DedupStatsDto, @@ -424,6 +425,51 @@ async fn populate_integrity_blob_sizes<'a>( IntegrityBlobSizes { hashes, sizes } } +/// Build the manifest reap statement from the registered reference sources. +/// +/// A manifest is collectible when either: +/// * `ref_count` reached 0 via `cleanup_if_orphaned` on the single-file +/// delete path, **or** +/// * nothing references it any more — the bulk-delete path (user cascade, +/// `empty_trash`), where the PG trigger only touches `storage.blobs` and +/// the per-file `cleanup_if_orphaned` call is skipped, so `ref_count` is +/// never decremented and the second clause is the only thing that reaps. +/// +/// The second clause used to name `storage.files` directly, which hardcoded +/// "files is the only thing that can reference a manifest". Any new referring +/// table — thumbnails via `storage.content_derived_blobs`, previews via +/// `storage.file_attached_blobs` — would then have its manifests reaped on the +/// next sweep *despite a correct `ref_count`*: clause one false, clause two +/// true, `OR` fires, bytes gone. See `docs/plan/derived-blobs.md`. +/// +/// # Panics +/// +/// If no source contributes at [`RefLevel::Manifest`]. That is a wiring bug, +/// and it must be loud: with no source, "nothing references it" is vacuously +/// true for every row and this statement would delete every manifest in the +/// database. `DedupService::new` always registers `FilesReferenceSource`, so +/// the only way to reach this is to pass a deliberately empty registry. +fn manifest_reap_sql(registry: &BlobReferenceRegistry) -> String { + let orphaned = registry + .no_reference_predicate(RefLevel::Manifest, "m.file_hash") + .expect( + "no manifest-level blob reference source registered: the reap \ + predicate would match every manifest", + ); + + format!( + "DELETE FROM storage.chunk_manifests + WHERE ctid = ANY( + SELECT ctid + FROM storage.chunk_manifests m + WHERE m.ref_count <= 0 + OR {orphaned} + LIMIT $1 + ) + RETURNING file_hash, chunk_hashes, total_size" + ) +} + pub struct DedupService { /// Pluggable blob storage backend (local FS, S3, …). backend: Arc, @@ -442,6 +488,16 @@ pub struct DedupService { /// seen immediately), weight-bounded (a manifest is ~72 B per chunk), /// short TTL so GC'd manifests age out fast (benches/MANIFEST-CACHE.md). manifest_cache: moka::future::Cache>, + /// Every table that holds blob references, so GC agrees with the + /// consistency jobs on what "referenced" means. Defaults to the two + /// built-in sources; DI replaces it once more tables exist. Never + /// optional — an empty registry would make "nothing references it" + /// vacuously true and the manifest sweep would reap everything. + reference_registry: Arc, + /// The manifest reap statement, built once from `reference_registry`. + /// Kept as a field so `garbage_collect` runs a fixed statement rather + /// than assembling SQL inside a delete loop — see `manifest_reap_sql`. + manifest_reap_sql: String, } impl DedupService { @@ -455,15 +511,32 @@ impl DedupService { pool: Arc, maintenance_pool: Arc, ) -> Self { + let registry = Arc::new(Self::default_reference_registry(pool.clone())); Self { backend, pool, maintenance_pool, blob_lifecycle: None, manifest_cache: Self::build_manifest_cache(), + reference_registry: registry.clone(), + manifest_reap_sql: manifest_reap_sql(®istry), } } + /// The two sources that were implicit before the registry existed. + /// Keeping this as the default means every construction path — including + /// tests — has a manifest-level source, so the reap predicate can never + /// degenerate to "nothing references anything". + fn default_reference_registry(pool: Arc) -> BlobReferenceRegistry { + use crate::infrastructure::repositories::pg::blob_reference_sources::{ + ChunksReferenceSource, FilesReferenceSource, + }; + let mut registry = BlobReferenceRegistry::new(); + registry.register(Arc::new(FilesReferenceSource::new(pool.clone()))); + registry.register(Arc::new(ChunksReferenceSource::new(pool))); + registry + } + /// See the `manifest_cache` field docs. Weight ≈ real heap bytes of one /// entry; 32 MiB cap ≈ tens of thousands of typical (sub-1 GB) files. fn build_manifest_cache() -> moka::future::Cache> { @@ -476,6 +549,15 @@ impl DedupService { .build() } + /// Registers the blob-reference registry used by the manifest reap + /// predicate. Without it `garbage_collect` skips manifest collection + /// entirely — see `docs/plan/derived-blobs.md`. + pub fn with_reference_registry(mut self, registry: Arc) -> Self { + self.manifest_reap_sql = manifest_reap_sql(®istry); + self.reference_registry = registry; + self + } + /// Registers the blob lifecycle dispatcher (thumbnail cleanup, …). pub fn with_blob_lifecycle(mut self, lifecycle: Arc) -> Self { self.blob_lifecycle = Some(lifecycle); @@ -510,12 +592,15 @@ impl DedupService { .connect_lazy("postgres://invalid:5432/none") .unwrap(), ); + let stub_registry = Arc::new(Self::default_reference_registry(stub_pool.clone())); Self { backend: Arc::new(LocalBlobBackend::new(Path::new("/tmp/oxicloud_stub_blobs"))), pool: stub_pool.clone(), - maintenance_pool: stub_pool, + maintenance_pool: stub_pool.clone(), blob_lifecycle: None, manifest_cache: Self::build_manifest_cache(), + reference_registry: stub_registry.clone(), + manifest_reap_sql: manifest_reap_sql(&stub_registry), } } @@ -523,6 +608,21 @@ impl DedupService { pub async fn initialize(&self) -> Result<(), DomainError> { self.backend.initialize().await?; + // The reap statement is assembled from the registered reference + // sources, so it is not greppable in the source tree. Log it once so an + // operator can read — or paste into psql — exactly what GC will delete. + tracing::debug!( + target: "oxicloud::dedup", + sources = ?self + .reference_registry + .sources() + .iter() + .map(|s| s.source_name()) + .collect::>(), + "manifest reap statement:\n{}", + self.manifest_reap_sql, + ); + let blob_count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM storage.blobs") .fetch_one(self.pool.as_ref()) .await @@ -2558,10 +2658,18 @@ impl DedupService { // A manifest is collectible when: // • ref_count has been decremented to 0 by cleanup_if_orphaned // on the single-file-delete service path, OR - // • no `storage.files.blob_hash` references its file_hash + // • NO registered reference source references its file_hash // (covers bulk-delete paths: user cascade, empty_trash — // where the PG trigger only touches storage.blobs and the // per-file cleanup_if_orphaned call is skipped). + // + // The second clause used to name `storage.files` directly. That + // hardcoded "files is the only thing that can reference a manifest", + // so any new referring table (thumbnails via + // storage.content_derived_blobs, …) would see its manifests reaped + // on the next sweep despite a correct ref_count — the first clause + // is false, the second true, and the OR fires. It is now the union + // of every registered source; see docs/plan/derived-blobs.md. loop { // Keep the historically cheap DELETE-only shape for the dominant // no-work sweep. Embedding it in the delete/aggregate/update CTE @@ -2570,23 +2678,14 @@ impl DedupService { // update. From two onward, aggregate in-process and issue one UPDATE: // the measured crossover is already positive at two, while 500 and // 1,000 manifests improve by 60.03x and 51.16x respectively. - let batch: Vec<(String, Vec, i64)> = sqlx::query_as( - "DELETE FROM storage.chunk_manifests - WHERE ctid = ANY( - SELECT ctid FROM storage.chunk_manifests m - WHERE m.ref_count <= 0 - OR NOT EXISTS ( - SELECT 1 FROM storage.files f - WHERE f.blob_hash = m.file_hash - ) - LIMIT $1 - ) - RETURNING file_hash, chunk_hashes, total_size", - ) - .bind(BATCH_SIZE) - .fetch_all(self.maintenance_pool.as_ref()) - .await - .map_err(|e| DomainError::internal_error("Dedup", format!("GC manifests: {e}")))?; + // Assembled once at construction (see `manifest_reap_sql`), not + // per sweep: no string work in the hot path, a stable statement for + // prepared-statement caching, and a byte-for-byte golden test. + let batch: Vec<(String, Vec, i64)> = sqlx::query_as(&self.manifest_reap_sql) + .bind(BATCH_SIZE) + .fetch_all(self.maintenance_pool.as_ref()) + .await + .map_err(|e| DomainError::internal_error("Dedup", format!("GC manifests: {e}")))?; if batch.is_empty() { break; @@ -3267,6 +3366,41 @@ impl crate::infrastructure::scheduler::JobHandler for DedupService { #[cfg(test)] mod tests { use super::*; + + /// Golden test for the statement `garbage_collect` runs against production + /// data. It is assembled from the registered reference sources rather than + /// written as a literal, so this pins the whole thing byte-for-byte — the + /// point being that a reviewer reads the SQL *here* instead of mentally + /// evaluating the registry. + /// + /// If this fails after adding a source, read the diff carefully: the new + /// branch must appear inside the `NOT (...)` group, ORed with the others. + /// A branch landing outside that group inverts the predicate for every + /// other source and reaps live manifests. + #[tokio::test] + async fn manifest_reap_statement_is_stable() { + let sql = DedupService::new_stub().manifest_reap_sql; + let expected = r#"DELETE FROM storage.chunk_manifests + WHERE ctid = ANY( + SELECT ctid + FROM storage.chunk_manifests m + WHERE m.ref_count <= 0 + OR NOT (EXISTS (SELECT 1 FROM storage.files cnt_f WHERE cnt_f.blob_hash = m.file_hash)) + LIMIT $1 + ) + RETURNING file_hash, chunk_hashes, total_size"#; + assert_eq!(sql, expected, "reap statement changed:\n{sql}"); + } + + /// The reap predicate must never match a manifest that some source still + /// references. With an empty registry `NOT (...)` would have no operands, + /// so the builder refuses rather than emitting a statement that deletes + /// every manifest in the database. + #[test] + #[should_panic(expected = "no manifest-level blob reference source")] + fn empty_registry_refuses_to_build_reap_statement() { + let _ = manifest_reap_sql(&BlobReferenceRegistry::new()); + } use std::collections::HashSet; use tempfile::NamedTempFile;