Merge branch 'fix/compression-orphan-quota'

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

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
+        // aborted the ENTIRE keepsake if the file vanished in between. It can still happen: the
-        // compression worker hard-deletes an original when its transcode fails, and it can still be
+        // compression worker no longer deletes originals on failure, but the hourly sweep reclaims
-        // running when the gallery is released. A missing source must degrade one entry, never the
+        // them once past the retention window, and an owner or host delete can land mid-export. A
-        // whole archive (which, once released, the host cannot rebuild without reopening uploads).
+        // 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
+            // Open-first: a source that disappeared between the manifest being built and now (a
-            // compression worker deletes originals on transcode failure) must skip this entry, not
+            // delete, or the hourly sweep reclaiming a long-failed original) must skip this entry,
-            // fail the whole viewer. Opening collapses the check and the use into one operation.
+            // 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) => {

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)

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<Uuid> {
+    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<i64>,
+    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"
+    );
+}