Skip to main content

max / makenotwork

server: purge expired items and decrement storage in one tx The purge did the per-user storage_used decrement and the CASCADE DELETE as two separate pool calls. A crash between them left the items un-purged but un-re-measured, so the next tick decremented the same items again -> storage_used_bytes under-counted in the creator's favor (fuzz-2026-07-06 LOW; bounded, self-heals at the weekly drift recalc, but real). Fold the measure + decrement + delete into one transaction sharing a fixed NOW(), so a crash rolls the pair back and the next tick redoes it cleanly; a failed decrement now aborts the whole purge rather than deleting items without crediting storage. get_expired_deleted_item_storage_by_user / purge_expired_deleted_items take an executor so they can run on the shared tx (a &PgPool still satisfies it). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-07-07 15:29 UTC
Commit: 723d84a26edaa024dd00addcd1eec92e1c8f962c
Parent: 7b639a6
2 files changed, +45 insertions, -18 deletions
M server/src/db/items/mod.rs +6 -4
@@ -498,7 +498,9 @@ pub async fn get_expired_deleted_item_version_s3_keys(pool: &PgPool) -> Result<V
498 498 /// Sum total file sizes per user for items about to be purged, including version files.
499 499 /// Returns (user_id, total_bytes) pairs for storage decrement.
500 500 #[tracing::instrument(skip_all)]
501 - pub async fn get_expired_deleted_item_storage_by_user(pool: &PgPool) -> Result<Vec<(super::UserId, i64)>> {
501 + pub async fn get_expired_deleted_item_storage_by_user<'e>(
502 + executor: impl sqlx::PgExecutor<'e>,
503 + ) -> Result<Vec<(super::UserId, i64)>> {
502 504 let rows: Vec<(super::UserId, i64)> = sqlx::query_as(
503 505 r#"
504 506 SELECT p.user_id,
@@ -524,7 +526,7 @@ pub async fn get_expired_deleted_item_storage_by_user(pool: &PgPool) -> Result<V
524 526 GROUP BY p.user_id
525 527 "#,
526 528 )
527 - .fetch_all(pool)
529 + .fetch_all(executor)
528 530 .await?;
529 531
530 532 Ok(rows)
@@ -617,11 +619,11 @@ pub async fn get_project_storage_bytes(pool: &PgPool, project_id: super::Project
617 619
618 620 /// Permanently delete items that were soft-deleted more than 7 days ago.
619 621 #[tracing::instrument(skip_all)]
620 - pub async fn purge_expired_deleted_items(pool: &PgPool) -> Result<u64> {
622 + pub async fn purge_expired_deleted_items<'e>(executor: impl sqlx::PgExecutor<'e>) -> Result<u64> {
621 623 let result = sqlx::query(
622 624 "DELETE FROM items WHERE deleted_at IS NOT NULL AND deleted_at < NOW() - INTERVAL '7 days'",
623 625 )
624 - .execute(pool)
626 + .execute(executor)
625 627 .await?;
626 628
627 629 Ok(result.rows_affected())
M server/src/scheduler/cleanup.rs +39 -14
@@ -399,35 +399,60 @@ pub(super) async fn purge_expired_deleted_items(state: &AppState) {
399 399 tracing::info!(count = all_s3_keys.len(), "deleted S3 objects for purged items");
400 400 }
401 401
402 - // Decrement storage for each affected user before CASCADE delete
403 - match db::items::get_expired_deleted_item_storage_by_user(&state.db).await {
402 + // Decrement each affected user's storage AND purge the items in ONE
403 + // transaction. Previously these were two pool calls: a crash between them
404 + // left the items un-purged but never re-measured, so the next tick decremented
405 + // the SAME items again → `storage_used_bytes` under-counted (fuzz-2026-07-06
406 + // LOW). Folding them means a crash rolls both back and the next tick redoes the
407 + // pair cleanly. The measure and the DELETE share the transaction's fixed
408 + // `NOW()`, so both act on exactly the same expired set.
409 + let mut tx = match state.db.begin().await {
410 + Ok(tx) => tx,
411 + Err(e) => {
412 + tracing::error!(error = ?e, "failed to open tx for item purge; will retry next tick");
413 + return;
414 + }
415 + };
416 +
417 + match db::items::get_expired_deleted_item_storage_by_user(&mut *tx).await {
404 418 Ok(user_sizes) => {
405 419 for (user_id, total_bytes) in &user_sizes {
406 420 if *total_bytes > 0
407 - && let Err(e) = db::creator_tiers::decrement_storage_used(&state.db, *user_id, *total_bytes).await
421 + && let Err(e) =
422 + db::creator_tiers::decrement_storage_used(&mut *tx, *user_id, *total_bytes).await
408 423 {
424 + // A failed decrement poisons the tx; abort the whole purge so
425 + // items are never deleted without their storage being credited
426 + // back. The next tick retries the pair atomically.
409 427 tracing::warn!(user_id = %user_id, bytes = total_bytes, error = ?e,
410 - "failed to decrement storage for purged items");
428 + "failed to decrement storage for purged items; aborting purge tx");
429 + return;
411 430 }
412 431 }
413 432 }
414 433 Err(e) => {
415 - tracing::error!(error = ?e, "failed to query storage sizes for items pending purge");
434 + tracing::error!(error = ?e, "failed to query storage sizes for items pending purge; aborting purge tx");
435 + return;
416 436 }
417 437 }
418 438
419 - match db::items::purge_expired_deleted_items(&state.db).await {
420 - Ok(0) => {
421 - let _ = db::scheduler_jobs::record_job_run(&state.db, "soft_delete_purge", 0).await;
422 - }
423 - Ok(n) => {
424 - tracing::info!(deleted = n, "purged expired soft-deleted items");
425 - let _ = db::scheduler_jobs::record_job_run(&state.db, "soft_delete_purge", n as i64).await;
426 - }
439 + let purged = match db::items::purge_expired_deleted_items(&mut *tx).await {
440 + Ok(n) => n,
427 441 Err(e) => {
428 - tracing::error!(error = ?e, "failed to purge expired soft-deleted items");
442 + tracing::error!(error = ?e, "failed to purge expired soft-deleted items; aborting purge tx");
443 + return;
429 444 }
445 + };
446 +
447 + if let Err(e) = tx.commit().await {
448 + tracing::error!(error = ?e, "failed to commit item purge tx; will retry next tick");
449 + return;
450 + }
451 +
452 + if purged > 0 {
453 + tracing::info!(deleted = purged, "purged expired soft-deleted items");
430 454 }
455 + let _ = db::scheduler_jobs::record_job_run(&state.db, "soft_delete_purge", purged as i64).await;
431 456 }
432 457
433 458 /// Outcome of a guarded single-key orphan S3 delete.