fix(upload): a client vanishing mid-commit lost the photo's preview forever

The compression job is handed to the worker by a bare `tokio::spawn` that runs
AFTER `tx.commit()`, and the commit is a suspension point. If the guest walks
out of range inside it, Postgres applies the COMMIT and axum drops the future
before the spawn is reached: the row lands durably at
`compression_status = 'pending'` with no job behind it.

Nothing looked at it again. `startup_recovery` rescues only `'processing'`, and
`backfill_stale_derivatives` — whose predicate WOULD match — runs once at boot
and never on a timer. The photo keeps its feed entry, never gets a derivative,
forces every viewer to pull the full original instead, and is skipped by the
diashow for the rest of the event. Silent, and permanent.

This is the same hazard `TempFileGuard` already covers for the file, reached one
line later, so it gets the same answer: an `EnqueueGuard` armed before the commit
and stood down only once the job is with the worker. `Drop` runs on cancellation,
which closes the window — and it covers the indeterminate-commit error path too,
which returns Err on a row that may well be live. Queueing a job for a row that
did not commit is harmless: the status update matches zero rows and the task
retires.

A guard cannot survive a SIGKILL, so `requeue_stuck_pending` sweeps for the same
state at boot and every ten minutes. The grace window is what makes it safe to
re-enter the live path: a task flips the row to `'processing'` as its FIRST act
after taking a permit, so a row still `'pending'` ten minutes on is lost rather
than merely queued. The event simulation peaked at 120 queued with a p99 of 97s
while being fed thirty times a real event's arrival rate, so the margin is wide.

Hit 1 upload in 932 during that run.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

backend/src/handlers/upload.rs

@@ -172,6 +172,74 @@ impl Drop for TempFileGuard {
     }
 }
 
+/// Guarantees the compression job is queued even if this request never finishes.
+///
+/// The enqueue is a bare `tokio::spawn` that happens AFTER `tx.commit()`, and the commit is a
+/// suspension point. If the guest walks out of range inside it, Postgres applies the COMMIT and
+/// axum drops this future before the spawn is reached: the row lands durably at
+/// `compression_status = 'pending'` with no job behind it, and nothing in the app ever looks at
+/// it again — `startup_recovery` only rescues `'processing'`, and the derivative backfill runs
+/// once at boot. The photo then shows in the feed with no preview, forces every viewer to pull
+/// the full original, and is skipped by the diashow forever.
+///
+/// This is the same hazard `TempFileGuard` covers for the file, reached one line later, and it
+/// gets the same answer: `Drop` runs on cancellation, so arming before the commit and disarming
+/// after the spawn closes the window. It also covers the indeterminate-commit error path, which
+/// returns `Err` on a row that may well be live.
+///
+/// Enqueuing a job for a row that did NOT commit is harmless: `set_compression_status` matches
+/// zero rows and `begin_derivative_attempt` returns `None`, so the task retires immediately.
+/// A guard is still not sufficient on its own — nothing in-process survives a SIGKILL — which is
+/// why `CompressionWorker::requeue_stuck_pending` sweeps for the same state on a timer.
+struct EnqueueGuard {
+    /// `None` once the job has actually been handed to the worker.
+    armed: Option<(crate::services::compression::CompressionWorker, Uuid, String, String)>,
+}
+
+impl EnqueueGuard {
+    fn new(
+        worker: crate::services::compression::CompressionWorker,
+        upload_id: Uuid,
+        original_path: String,
+        mime_type: String,
+    ) -> Self {
+        Self {
+            armed: Some((worker, upload_id, original_path, mime_type)),
+        }
+    }
+
+    /// The job is queued; stand down.
+    fn disarm(&mut self) {
+        self.armed = None;
+    }
+}
+
+impl Drop for EnqueueGuard {
+    fn drop(&mut self) {
+        let Some((worker, upload_id, path, mime)) = self.armed.take() else {
+            return;
+        };
+        // `process` calls `tokio::spawn`, which panics without a runtime. During a normal
+        // cancellation we are still on the runtime that dropped us; during shutdown we may not
+        // be. Log loudly rather than panicking in a destructor — the timed sweeper is the
+        // backstop for exactly this case.
+        if tokio::runtime::Handle::try_current().is_err() {
+            tracing::error!(
+                %upload_id,
+                "upload committed but the compression job could not be queued (no runtime in \
+                 Drop); requeue_stuck_pending will pick it up"
+            );
+            return;
+        }
+        tracing::warn!(
+            %upload_id,
+            "request ended before the compression job was queued — queueing it from the drop \
+             guard (client most likely disconnected during COMMIT)"
+        );
+        worker.process(upload_id, path, mime);
+    }
+}
+
 /// Allowlist of accepted media types, keyed by the MIME that `infer` derives from
 /// the file's magic bytes. The detected MIME (not the client-declared one) is what
 /// we trust, store, and hand to the compression pipeline — so a text-based payload
@@ -719,7 +787,9 @@ pub async fn upload(
     // Quota accounting, the upload row, and its hashtag links must be atomic: a
     // crash between the bytes increment and the insert would permanently charge
     // bytes with no row to reclaim them (silent quota erosion / spurious lockout).
-    let tx_result: Result<Upload, AppError> = async {
+    // Carries the enqueue guard out with the row: it is armed inside the block (before the
+    // commit) and can only be stood down after `process()` runs, which happens out here.
+    let tx_result: Result<(Upload, EnqueueGuard), AppError> = async {
         let mut tx = state.pool.begin().await?;
 
         // RE-CHECK THE LOCK, UNDER A ROW LOCK, INSIDE THE COMMIT TX.
@@ -836,6 +906,15 @@ pub async fn upload(
         // the failure to the recoverable side: if we are dropped mid-commit the bytes leak,
         // and leaked bytes under a final name are exactly what the orphan sweeper reclaims.
         // A committed row whose file we deleted is unrecoverable. Prefer the leak.
+        // Armed BEFORE the commit, for the mirror-image of the reason the file guard is
+        // disarmed before it: from here on, a cancellation can leave a live row behind, and a
+        // live row with no compression job is silent, permanent loss. See `EnqueueGuard`.
+        let enqueue_guard = EnqueueGuard::new(
+            state.compression.clone(),
+            upload.id,
+            relative_path.clone(),
+            mime.clone(),
+        );
         file_guard.disarm();
         if let Err(e) = tx.commit().await {
             // Deliberately do NOT re-arm the guard here.
@@ -861,7 +940,7 @@ pub async fn upload(
             );
             return Err(e.into());
         }
-        Ok(upload)
+        Ok((upload, enqueue_guard))
     }
     .await;
 
@@ -873,8 +952,8 @@ pub async fn upload(
     //
     // The successful-commit case disarmed the guard inside the block, immediately before
     // `tx.commit()` — see the comment there for why it cannot be done out here.
-    let upload = match tx_result {
-        Ok(u) => u,
+    let (upload, mut enqueue_guard) = match tx_result {
+        Ok(v) => v,
         // The concurrent duplicate resolved inside the transaction. The winner's row is committed;
         // answer with it so both retries of the same photo get the same successful reply. The
         // loser's bytes are reclaimed by the guard when this return drops it.
@@ -917,10 +996,12 @@ pub async fn upload(
         Err(e) => return Err(e),
     };
 
-    // Spawn compression task
+    // Spawn compression task. The guard above has covered this call since before the commit;
+    // stand it down only once the job is genuinely with the worker.
     state
         .compression
         .process(upload.id, relative_path, mime.clone());
+    enqueue_guard.disarm();
 
     // Broadcast SSE event
     let dto = UploadDto {

backend/src/main.rs

@@ -127,6 +127,7 @@ async fn main() -> Result<()> {
         state.rate_limiter.clone(),
         state.sse_tickets.clone(),
         config.media_path.clone(),
+        state.compression.clone(),
     );
 
     let api = Router::new()

backend/src/services/compression.rs

@@ -411,6 +411,66 @@ impl CompressionWorker {
     /// `derivative_attempts` stops a fatal row being replayed on every boot, `BACKFILL_BATCH`
     /// stops one start queueing unbounded work, and the whole thing runs as ONE task walking
     /// the rows sequentially rather than N tasks racing for the same semaphore.
+    /// How long an upload may sit at `'pending'` before the sweeper assumes its job was lost.
+    ///
+    /// Must comfortably exceed the deepest real queue wait, or the sweeper re-enqueues photos
+    /// that are merely waiting their turn. The live path makes that easy to bound: a task's
+    /// FIRST action after taking a permit is to flip the row to `'processing'`, so a row still
+    /// `'pending'` after this long is either behind that much work or genuinely lost. The
+    /// event simulation peaked at 120 queued with a p99 of 97s while being fed thirty times a
+    /// real event's arrival rate, so ten minutes is a wide margin rather than a tight one.
+    const PENDING_GRACE: chrono::Duration = chrono::Duration::minutes(10);
+
+    /// Re-queue uploads whose compression job was lost between the commit and the spawn.
+    ///
+    /// `EnqueueGuard` closes that window for a cancelled request, but nothing in-process
+    /// survives a SIGKILL or an OOM kill, and `startup_recovery` only rescues `'processing'`
+    /// — so a row orphaned at `'pending'` had no path back at all. Without this, the failure is
+    /// silent and permanent: the photo keeps its feed entry, never gets a derivative, and is
+    /// dropped from the diashow for the rest of the event.
+    ///
+    /// Re-entering the live `process` path (rather than regenerating derivatives inline, as
+    /// `backfill_stale_derivatives` does) is deliberate: it reuses the status transitions, the
+    /// retry ladder and the SSE broadcast, so a rescued photo behaves exactly like one that was
+    /// never lost.
+    pub async fn requeue_stuck_pending(&self) {
+        let cutoff = chrono::Utc::now() - Self::PENDING_GRACE;
+        let rows = sqlx::query_as::<_, (Uuid, String, String)>(
+            "SELECT id, original_path, mime_type FROM upload
+             WHERE deleted_at IS NULL
+               AND compression_status = 'pending'
+               AND original_path <> ''
+               AND created_at < $1
+               AND derivative_attempts < $2
+             ORDER BY created_at
+             LIMIT $3",
+        )
+        .bind(cutoff)
+        .bind(Self::MAX_DERIVATIVE_ATTEMPTS)
+        .bind(Self::BACKFILL_BATCH)
+        .fetch_all(&self.pool)
+        .await;
+        let rows = match rows {
+            Ok(r) => r,
+            Err(e) => {
+                tracing::warn!(error = ?e, "stuck-pending sweep query failed");
+                return;
+            }
+        };
+        if rows.is_empty() {
+            return;
+        }
+        // WARN, not INFO: reaching this means the guard did not run, which is worth noticing
+        // even though the photo is being rescued.
+        tracing::warn!(
+            count = rows.len(),
+            "found upload(s) stuck at 'pending' with no compression job; re-queueing"
+        );
+        for (id, original_path, mime_type) in rows {
+            self.process(id, original_path, mime_type);
+        }
+    }
+
     pub async fn backfill_stale_derivatives(&self) {
         // `original_path IS NOT NULL` was dead — the column is NOT NULL. What actually needs
         // excluding is the blanked path `cleanup_deleted_media` leaves behind.

backend/src/services/maintenance.rs

@@ -144,7 +144,36 @@ pub fn spawn_periodic_tasks(
     rate_limiter: RateLimiter,
     sse_tickets: SseTicketStore,
     media_path: PathBuf,
+    compression: crate::services::compression::CompressionWorker,
 ) {
+    // Rescuing a lost compression job gets its OWN loop rather than a line in the hourly one.
+    // Cadence is the whole point: an upload stuck at 'pending' is a photo missing from the
+    // diashow and showing without a preview, so an hour of that during a five-hour party is a
+    // guest-visible hole. Ten minutes bounds it while still sitting far above the deepest
+    // observed queue wait (see `PENDING_GRACE`). Supervised for the same reason as the loop
+    // below: silently dying is how this class of safety net stops existing.
+    tokio::spawn(async move {
+        loop {
+            let worker = compression.clone();
+            let inner = tokio::spawn(async move {
+                // `interval` fires its first tick immediately, and that is deliberate here:
+                // a restart is exactly when stranded rows exist (the drop guard cannot
+                // survive a SIGKILL), and `startup_recovery` rescues only 'processing'. The
+                // grace window makes the boot pass safe — nothing uploaded in the last ten
+                // minutes can match, so this can never race a job that is merely queued.
+                let mut tick = tokio::time::interval(Duration::from_secs(600));
+                loop {
+                    tick.tick().await;
+                    worker.requeue_stuck_pending().await;
+                }
+            });
+            match inner.await {
+                Ok(()) => tracing::error!("stuck-pending sweeper returned; restarting it"),
+                Err(e) => tracing::error!(error = ?e, "stuck-pending sweeper died; restarting it"),
+            }
+            tokio::time::sleep(Duration::from_secs(60)).await;
+        }
+    });
     // Supervised, because this one task carries EVERY piece of recurring hygiene in the app:
     // session pruning, media reclamation, the orphan-temp sweep, and the rate-limiter and
     // SSE-ticket maps. As a bare `tokio::spawn` with no retained handle, a single panic anywhere