From c14ccd2df17fdfbcda10e6829c6a9f21e7efca35 Mon Sep 17 00:00:00 2001 From: fabi Date: Tue, 28 Jul 2026 20:40:15 +0200 Subject: [PATCH] fix(compression): reclaim failed originals instead of leaking them MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Round 1 stopped the compression worker deleting an upload's original on failure — a transient ENOSPC or a codec panic must never destroy the only copy of a photo a guest cannot retake. But it left `Upload::soft_delete`'s quota refund in place, so the bytes stayed on disk while the uploader was charged nothing for them. That is worse than it first looks. The row is soft-deleted, so the file is invisible and unowned; a guest hitting a reproducible codec failure can accumulate orphans indefinitely at zero personal cost. And `active_uploaders` counts only users with non-deleted uploads, so dropping out of that count RAISES everyone's per-user ceiling — the leak loosens the very quota meant to contain it. Keep the refund: the uploader didn't cause the failure and shouldn't silently lose quota to it. Bound the leak instead, with an hourly sweep alongside the existing session cleanup in `spawn_periodic_tasks`, reclaiming failed originals older than 14 days — comfortably longer than any single event, so an operator investigating a failed upload still has the file. The selection predicate is the entire safety argument, so it is deliberately narrow: `compression_status = 'failed'` AND soft-deleted AND past the window AND `original_path <> ''`. That is exactly the state the give-up path leaves behind, and it cannot reach a live upload, an owner-deleted one, or a failure still inside its recovery window. `original_path` is cleared after a successful reclaim, which makes the sweep idempotent — otherwise a row whose file is already gone is re-selected on every tick forever. The row itself is kept as the audit trail. Tests reproduce the selection verbatim (same pattern as upload_concurrency) and assert it against five near-misses that must survive, both sides of the retention boundary, and the idempotence property. Also fixes two comments in export.rs still claiming "the compression worker hard-deletes an original when its transcode fails" — no longer true, and the defensive handling they justify is now justified by this sweep and by ordinary deletes instead. Co-Authored-By: Claude Opus 5 (1M context) --- backend/src/main.rs | 1 + backend/src/services/export.rs | 15 +- backend/src/services/maintenance.rs | 92 ++++++++++- backend/tests/failed_original_sweep.rs | 207 +++++++++++++++++++++++++ 4 files changed, 305 insertions(+), 10 deletions(-) create mode 100644 backend/tests/failed_original_sweep.rs diff --git a/backend/src/main.rs b/backend/src/main.rs index 569ee17..bf5a3b9 100644 --- a/backend/src/main.rs +++ b/backend/src/main.rs @@ -70,6 +70,7 @@ async fn main() -> Result<()> { pool, state.rate_limiter.clone(), state.sse_tickets.clone(), + config.media_path.clone(), ); // Ensure media directories exist diff --git a/backend/src/services/export.rs b/backend/src/services/export.rs index 850658f..4622066 100644 --- a/backend/src/services/export.rs +++ b/backend/src/services/export.rs @@ -720,10 +720,11 @@ async fn run_html_export_inner( let src = media_path.join(&row.original_path); // Stat ONCE, up front, and skip this upload if the source is gone. The old code probed with // `exists()` here and then did `metadata(&src).await?` further down — a TOCTOU whose `?` - // aborted the ENTIRE keepsake if the file vanished in between. It genuinely can: the - // compression worker hard-deletes an original when its transcode fails, and it can still be - // running when the gallery is released. A missing source must degrade one entry, never the - // whole archive (which, once released, the host cannot rebuild without reopening uploads). + // aborted the ENTIRE keepsake if the file vanished in between. It can still happen: the + // compression worker no longer deletes originals on failure, but the hourly sweep reclaims + // them once past the retention window, and an owner or host delete can land mid-export. A + // missing source must degrade one entry, never the whole archive (which, once released, the + // host cannot rebuild without reopening uploads). let src_meta = match tokio::fs::metadata(&src).await { Ok(m) => m, Err(e) => { @@ -953,9 +954,9 @@ async fn run_html_export_inner( for (name, source) in &media_manifest { let path = source.path(); - // Open-first: a source that disappeared between the manifest being built and now (the - // compression worker deletes originals on transcode failure) must skip this entry, not - // fail the whole viewer. Opening collapses the check and the use into one operation. + // Open-first: a source that disappeared between the manifest being built and now (a + // delete, or the hourly sweep reclaiming a long-failed original) must skip this entry, + // not fail the whole viewer. Opening collapses the check and the use into one operation. let src_file = match tokio::fs::File::open(path).await { Ok(f) => f, Err(e) => { diff --git a/backend/src/services/maintenance.rs b/backend/src/services/maintenance.rs index 0460d59..2cd07eb 100644 --- a/backend/src/services/maintenance.rs +++ b/backend/src/services/maintenance.rs @@ -9,10 +9,12 @@ //! users staring at a spinner. Resetting them on startup recovers gracefully. //! //! 2. **Periodic tasks** — pruning that should happen "every hour" rather than per -//! request: expired sessions (otherwise the table grows unboundedly), and the +//! request: expired sessions (otherwise the table grows unboundedly), the //! rate-limiter's in-memory windows (so keys for IPs that left long ago don't -//! accumulate). +//! accumulate), and the originals of uploads whose compression permanently failed +//! (which are deliberately retained for a recovery window, then reclaimed). +use std::path::PathBuf; use std::time::Duration; use sqlx::PgPool; @@ -20,6 +22,21 @@ use sqlx::PgPool; use crate::services::rate_limiter::RateLimiter; use crate::services::sse_tickets::SseTicketStore; +/// How long a permanently-failed upload's original is kept on disk before it is +/// reclaimed. +/// +/// The compression worker stops deleting originals on failure — a transient error must +/// never destroy the guest's only copy of a photo they can't retake. But the row is +/// soft-deleted and the uploader's quota IS refunded, so without a sweep those bytes are +/// invisible, unowned, and free: a reproducible codec failure lets one guest accumulate +/// orphans at no personal cost, and because `active_uploaders` counts only users with +/// non-deleted uploads, dropping out of that count actually RAISES everyone's per-user +/// ceiling while the disk gets fuller. +/// +/// Two weeks is comfortably longer than any single event, so an operator investigating a +/// failed upload still has the file, while the leak stays bounded. +const FAILED_ORIGINAL_RETENTION_DAYS: i64 = 14; + /// Reset rows left in flight by a previous crashed instance. Run once on startup, /// before the HTTP server starts taking requests, so users never observe the /// half-state. @@ -87,7 +104,12 @@ pub async fn startup_recovery(pool: &PgPool) { /// - drops expired SSE tickets (30s TTL but the map keeps the slot until pruned) /// /// Cadence is 1h — fine for both jobs at our scale. -pub fn spawn_periodic_tasks(pool: PgPool, rate_limiter: RateLimiter, sse_tickets: SseTicketStore) { +pub fn spawn_periodic_tasks( + pool: PgPool, + rate_limiter: RateLimiter, + sse_tickets: SseTicketStore, + media_path: PathBuf, +) { tokio::spawn(async move { let mut tick = tokio::time::interval(Duration::from_secs(3600)); // Fire the first tick immediately, then hourly. @@ -95,12 +117,76 @@ pub fn spawn_periodic_tasks(pool: PgPool, rate_limiter: RateLimiter, sse_tickets loop { tick.tick().await; cleanup_sessions(&pool).await; + cleanup_failed_originals(&pool, &media_path).await; rate_limiter.prune(); sse_tickets.prune(); } }); } +/// Reclaim the originals of uploads whose compression permanently failed, once they are +/// past [`FAILED_ORIGINAL_RETENTION_DAYS`]. +/// +/// Deliberately narrow. It only touches rows that are BOTH `compression_status = 'failed'` +/// AND soft-deleted — i.e. the exact state the compression worker's give-up path leaves +/// behind — so it can never reach a live upload or one whose preview works. `original_path` +/// is cleared in the same pass, which makes the sweep idempotent and stops a later run +/// re-reporting a file that is already gone. The row itself is kept: it is the audit trail +/// for the failure, and it costs a few hundred bytes. +async fn cleanup_failed_originals(pool: &PgPool, media_path: &std::path::Path) { + let rows = sqlx::query_as::<_, (uuid::Uuid, String)>( + "SELECT id, original_path FROM upload + WHERE compression_status = 'failed' + AND deleted_at IS NOT NULL + AND deleted_at < NOW() - ($1 || ' days')::interval + AND original_path <> ''", + ) + .bind(FAILED_ORIGINAL_RETENTION_DAYS.to_string()) + .fetch_all(pool) + .await; + + let rows = match rows { + Ok(r) => r, + Err(e) => { + tracing::warn!(error = ?e, "failed-original sweep query failed"); + return; + } + }; + if rows.is_empty() { + return; + } + + let mut reclaimed = 0usize; + for (id, original_path) in rows { + let absolute = media_path.join(&original_path); + match tokio::fs::remove_file(&absolute).await { + Ok(()) => reclaimed += 1, + // Already gone (manual cleanup, restored backup) — still clear the column so + // the row stops being re-selected every hour. + Err(e) if e.kind() == std::io::ErrorKind::NotFound => {} + Err(e) => { + tracing::warn!(error = ?e, %id, path = %absolute.display(), + "could not reclaim failed original; leaving the row for the next sweep"); + continue; + } + } + if let Err(e) = sqlx::query("UPDATE upload SET original_path = '' WHERE id = $1") + .bind(id) + .execute(pool) + .await + { + tracing::warn!(error = ?e, %id, "reclaimed the file but could not clear original_path"); + } + } + + if reclaimed > 0 { + tracing::info!( + "reclaimed {reclaimed} original(s) from uploads that failed compression more than \ + {FAILED_ORIGINAL_RETENTION_DAYS} days ago" + ); + } +} + async fn cleanup_sessions(pool: &PgPool) { match sqlx::query("DELETE FROM session WHERE expires_at < NOW() - INTERVAL '1 day'") .execute(pool) diff --git a/backend/tests/failed_original_sweep.rs b/backend/tests/failed_original_sweep.rs new file mode 100644 index 0000000..a1743d9 --- /dev/null +++ b/backend/tests/failed_original_sweep.rs @@ -0,0 +1,207 @@ +//! DB-backed tests for the failed-original sweep (`services/maintenance.rs`). +//! +//! Context: the compression worker deliberately no longer deletes an upload's original when +//! its transcode fails — a transient ENOSPC or a codec panic must never destroy the only +//! copy of a photo a guest cannot retake. But the row is soft-deleted and the uploader's +//! quota IS refunded, so those bytes become invisible, unowned and free. A guest hitting a +//! reproducible codec failure could accumulate orphans at no personal cost, and since +//! `active_uploaders` counts only users with non-deleted uploads, dropping out of that count +//! actually RAISES everyone's per-user ceiling while the disk fills. +//! +//! The sweep reclaims them after a retention window. Its selection predicate is the whole +//! safety argument — it must reach the give-up path's leftovers and nothing else — so that +//! is what these tests pin, following the same "reproduce the SQL verbatim" pattern as +//! `upload_concurrency.rs`. +//! +//! `#[sqlx::test]` gives each test a fresh database with the real migrations applied. + +mod common; + +use common::*; +use sqlx::PgPool; +use uuid::Uuid; + +/// SRC: `services/maintenance.rs::cleanup_failed_originals` — the selection, verbatim. +async fn sweep_selects(pool: &PgPool, retention_days: i64) -> Vec { + sqlx::query_as::<_, (Uuid, String)>( + "SELECT id, original_path FROM upload + WHERE compression_status = 'failed' + AND deleted_at IS NOT NULL + AND deleted_at < NOW() - ($1 || ' days')::interval + AND original_path <> ''", + ) + .bind(retention_days.to_string()) + .fetch_all(pool) + .await + .expect("sweep query") + .into_iter() + .map(|(id, _)| id) + .collect() +} + +#[allow(clippy::too_many_arguments)] +async fn seed_upload( + pool: &PgPool, + event_id: Uuid, + user_id: Uuid, + status: &str, + deleted_days_ago: Option, + original_path: &str, +) -> Uuid { + let id: Uuid = sqlx::query_scalar( + "INSERT INTO upload (event_id, user_id, original_path, mime_type, original_size_bytes, + compression_status, deleted_at) + VALUES ($1, $2, $3, 'image/jpeg', 1000, $4, + CASE WHEN $5::bigint IS NULL THEN NULL + ELSE NOW() - ($5::text || ' days')::interval END) + RETURNING id", + ) + .bind(event_id) + .bind(user_id) + .bind(original_path) + .bind(status) + .bind(deleted_days_ago) + .fetch_one(pool) + .await + .expect("seed upload"); + id +} + +#[sqlx::test] +async fn sweeps_only_long_failed_soft_deleted_uploads(pool: PgPool) { + let event_id = seed_event(&pool, "sweep-event").await; + let user_id = seed_user(&pool, event_id, "Sweeper").await; + + // The one and only thing the sweep may touch: the exact state the compression worker's + // give-up path leaves behind, aged past the window. + let target = seed_upload( + &pool, + event_id, + user_id, + "failed", + Some(30), + "originals/e/target.jpg", + ) + .await; + + // Everything below is a near-miss that must survive. + // A healthy live upload — the catastrophic case if the predicate were ever loosened. + let live = seed_upload(&pool, event_id, user_id, "done", None, "originals/e/live.jpg").await; + // Failed but still inside the retention window: the recovery window is the entire point + // of keeping the file, so reclaiming it early would defeat the fix it protects. + let recent = seed_upload( + &pool, + event_id, + user_id, + "failed", + Some(1), + "originals/e/recent.jpg", + ) + .await; + // Failed but NOT soft-deleted — not the give-up path; something else set this status. + let failed_live = seed_upload( + &pool, + event_id, + user_id, + "failed", + None, + "originals/e/failed-live.jpg", + ) + .await; + // Soft-deleted by the OWNER, compression fine. Its file is already gone; this row must + // never be re-processed. + let owner_deleted = seed_upload( + &pool, + event_id, + user_id, + "done", + Some(30), + "originals/e/owner.jpg", + ) + .await; + // Already swept: `original_path` cleared. Re-selecting it every hour would log a + // phantom reclaim forever. + let already_swept = seed_upload(&pool, event_id, user_id, "failed", Some(30), "").await; + + let selected = sweep_selects(&pool, 14).await; + + assert_eq!( + selected, + vec![target], + "the sweep must select exactly the aged give-up-path leftovers" + ); + for (id, what) in [ + (live, "a live upload"), + (recent, "a failure still inside the retention window"), + (failed_live, "a failed but not soft-deleted upload"), + (owner_deleted, "an owner-deleted upload"), + (already_swept, "an already-swept row"), + ] { + assert!(!selected.contains(&id), "the sweep must not touch {what}"); + } +} + +#[sqlx::test] +async fn retention_window_is_honoured_at_the_boundary(pool: PgPool) { + let event_id = seed_event(&pool, "sweep-boundary").await; + let user_id = seed_user(&pool, event_id, "Boundary").await; + + let inside = seed_upload( + &pool, + event_id, + user_id, + "failed", + Some(13), + "originals/e/inside.jpg", + ) + .await; + let outside = seed_upload( + &pool, + event_id, + user_id, + "failed", + Some(15), + "originals/e/outside.jpg", + ) + .await; + + let selected = sweep_selects(&pool, 14).await; + assert!( + selected.contains(&outside), + "15 days old must be past a 14-day window" + ); + assert!( + !selected.contains(&inside), + "13 days old must still be retained" + ); +} + +#[sqlx::test] +async fn clearing_original_path_makes_the_sweep_idempotent(pool: PgPool) { + // The sweep clears `original_path` after reclaiming the file. Without that, a row whose + // file is already gone is re-selected on every hourly tick forever. + let event_id = seed_event(&pool, "sweep-idempotent").await; + let user_id = seed_user(&pool, event_id, "Idem").await; + let id = seed_upload( + &pool, + event_id, + user_id, + "failed", + Some(30), + "originals/e/once.jpg", + ) + .await; + + assert_eq!(sweep_selects(&pool, 14).await, vec![id]); + + sqlx::query("UPDATE upload SET original_path = '' WHERE id = $1") + .bind(id) + .execute(&pool) + .await + .expect("clear path"); + + assert!( + sweep_selects(&pool, 14).await.is_empty(), + "a swept row must not come back" + ); +}