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

perf(db): statement-level tree-ETag triggers; batched trash purge

Browse Source 
Replace the per-row tree_modified_at bump triggers with AFTER ... FOR
EACH STATEMENT triggers using transition tables: each DML statement now
pays one bump covering the distinct ancestor chains of all affected
rows (locked in id order so concurrent bumps over overlapping chains
cannot deadlock), instead of one chain UPDATE per affected row.
Value-based change detection replaces UPDATE OF column lists (PG
forbids those with transition tables), so EXIF media_sort_date syncs
and no-op updates no longer bump at all, and file moves now invalidate
the source chain as well as the destination.

Also fix a latent bug surfaced while testing this: the descendant
path/lpath cascade (trg_folders_cascade_path) was declared AFTER
UPDATE OF path, lpath, but rename/move statements SET name/parent_id,
and BEFORE-trigger rewrites do not count for UPDATE OF — the cascade
never fired, leaving every descendant folder with a stale path/lpath
after any rename or move. The migration canonically repairs existing
trees and re-creates the cascade on the columns the app actually
writes.

delete_expired_bulk now deletes in LIMIT-ed batches (1000 files / 100
folders per round, each its own implicit transaction) ordered by
trashed_at and served by new partial indexes over trashed rows, so
retention purges no longer hold one unbounded transaction.

Verified against a local PG16 cluster: all migrations apply from
scratch, the new one re-runs idempotently, and a 16-case behavioral
battery passes (chain bumps for insert/update/delete/move/trash, EXIF
invisibility, no-op invisibility, cascade repair, FK-cascade depth
guards, batched purge shape and partial-index plan).

https://claude.ai/code/session_01QxwJDHqQhbMkHK333QtMme
This commit is contained in:
Claude
2026-06-10 13:22:27 +00:00
parent c3b853abd2
commit 4200209d4a
2 changed files with 379 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/infrastructure/repositories/pg/trash_db_repository.rs
+61 -31
View File
@@ -47,6 +47,40 @@ impl TrashDbRepository {
}
}
/// Runs a LIMIT-ed DELETE statement repeatedly until a round affects
/// fewer rows than `batch_size`, yielding to the runtime between rounds.
///
/// `sql` must bind `$1` = cutoff timestamp and `$2` = batch size; the
/// candidate sub-select is served by the `idx_*_trash_expiry` partial
/// indexes. Each round is its own implicit transaction, so row locks,
/// WAL volume and the statement-trigger transition tables stay bounded
/// no matter how many items expired. Partial progress is fine — the
/// next retention sweep continues where this one stopped.
async fn delete_expired_batch_loop(
&self,
sql: &'static str,
cutoff: DateTime<Utc>,
batch_size: i64,
) -> Result<u64> {
let mut total: u64 = 0;
loop {
let affected = sqlx::query(sql)
.bind(cutoff)
.bind(batch_size)
.execute(self.pool.as_ref())
.await
.map_err(|e| {
DomainError::internal_error("TrashDb", format!("bulk delete batch: {e}"))
})?
.rows_affected();
total += affected;
if affected < batch_size as u64 {
return Ok(total);
}
tokio::task::yield_now().await;
}
}
/// Convert a trash_items view row into a TrashedItem entity.
fn row_to_trashed_item(
&self,
@@ -180,40 +214,36 @@ impl TrashRepository for TrashDbRepository {
async fn delete_expired_bulk(&self) -> Result<(u64, u64)> {
let cutoff = Utc::now() - chrono::Duration::days(self.retention_days);
let mut tx = self
.pool
.begin()
.await
.map_err(|e| DomainError::internal_error("TrashDb", format!("begin tx: {e}")))?;
// 1. Bulk-delete expired trashed files.
// 1. Bulk-delete expired trashed files in batches.
// 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)
.execute(&mut *tx)
.await
.map_err(|e| {
DomainError::internal_error("TrashDb", format!("bulk delete files: {e}"))
})?
.rows_affected();
let files_deleted = self
.delete_expired_batch_loop(
"DELETE FROM storage.files
WHERE id IN (SELECT id FROM storage.files
WHERE is_trashed = TRUE AND trashed_at < $1
ORDER BY trashed_at
LIMIT $2)",
cutoff,
1_000,
)
.await?;
// 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}")))?;
// 2. Bulk-delete expired trashed folders in batches.
// FK ON DELETE CASCADE handles descendant folders and their
// files, so each row can fan out to an entire subtree — hence
// the smaller batch size.
let folders_deleted = self
.delete_expired_batch_loop(
"DELETE FROM storage.folders
WHERE id IN (SELECT id FROM storage.folders
WHERE is_trashed = TRUE AND trashed_at < $1
ORDER BY trashed_at
LIMIT $2)",
cutoff,
100,
)
.await?;
Ok((files_deleted, folders_deleted))
}