fix(ops): reclaim abandoned upload temp files, and supervise the task that does

`stream_field_to_file` removes its `.tmp` on every error return, which covers
everything the handler can see. It cannot cover what actually happens at a party: the
client goes away — a phone sleeps, a guest walks out of range, the PWA is evicted
mid-video — and axum DROPS the handler future rather than returning an error, so no
cleanup runs at all. The shutdown backstop force-exits in-flight handlers for the same
net effect.

Nothing else reclaimed them. `cleanup_deleted_media` only visits rows with
`deleted_at`, and an abandoned upload never got a row; `export::sweep_orphan_temps` is
only ever pointed at the exports volume. `grep -rn read_dir src/` had three hits, all
in export.rs — the media tree was never read by anything.

So every abandonment stranded up to `max_video_size_mb` of unowned bytes permanently,
on the same 40 GB filesystem as `postgres_data`. Worse than a leak: the per-user quota
is computed from live free disk, so those bytes were also subtracted from what everyone
else was allowed to upload. An evening of flaky venue wifi could take the event down.

The threshold is on modification time, not creation time, which is what makes an hour
safe: a live upload is written to continuously so its mtime keeps advancing and it can
never age into the sweep no matter how slow the connection. The clock only starts once
the writer stops.

The periodic task is now supervised. It carries every piece of recurring hygiene in the
app — session pruning, media reclamation, this sweep, and the rate-limiter and
SSE-ticket maps — as a bare `tokio::spawn` with no retained handle, so a single panic
anywhere inside it stopped all five permanently and silently. No log line, no symptom
until the disk or a HashMap grew into one.

Tested: an hour-old temp is reclaimed, a temp still being written to is not (deleting
that one destroys a live upload), a committed `.jpg` is never touched, and a media tree
that does not exist yet is a silent no-op rather than an error logged 24 times a day.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

backend/src/services/maintenance.rs

@@ -132,20 +132,136 @@ pub fn spawn_periodic_tasks(
     sse_tickets: SseTicketStore,
     media_path: PathBuf,
 ) {
+    // 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
+    // inside it stopped all five permanently and silently — no log line, no symptom until the
+    // disk or a HashMap grew into one. The supervisor re-spawns and, just as importantly, says
+    // so; it can never spin hot because the inner loop only returns by dying.
     tokio::spawn(async move {
-        let mut tick = tokio::time::interval(Duration::from_secs(3600));
-        // Fire the first tick immediately, then hourly.
-        tick.tick().await;
         loop {
-            tick.tick().await;
-            cleanup_sessions(&pool).await;
-            cleanup_deleted_media(&pool, &media_path).await;
-            rate_limiter.prune();
-            sse_tickets.prune();
+            let inner = tokio::spawn(periodic_loop(
+                pool.clone(),
+                rate_limiter.clone(),
+                sse_tickets.clone(),
+                media_path.clone(),
+            ));
+            match inner.await {
+                Ok(()) => tracing::error!("periodic maintenance loop returned; restarting it"),
+                Err(e) => {
+                    tracing::error!(error = ?e, "periodic maintenance task died; restarting it")
+                }
+            }
+            tokio::time::sleep(Duration::from_secs(60)).await;
         }
     });
 }
 
+/// The actual hygiene loop. Never returns in normal operation — see the supervisor above.
+async fn periodic_loop(
+    pool: PgPool,
+    rate_limiter: RateLimiter,
+    sse_tickets: SseTicketStore,
+    media_path: PathBuf,
+) {
+    // A crash left whatever the previous process was mid-upload behind, and the first periodic
+    // tick is an hour away — sweep once up front so a restart is also a cleanup.
+    sweep_orphan_upload_temps(&media_path).await;
+
+    let mut tick = tokio::time::interval(Duration::from_secs(3600));
+    // Fire the first tick immediately, then hourly.
+    tick.tick().await;
+    loop {
+        tick.tick().await;
+        cleanup_sessions(&pool).await;
+        cleanup_deleted_media(&pool, &media_path).await;
+        sweep_orphan_upload_temps(&media_path).await;
+        rate_limiter.prune();
+        sse_tickets.prune();
+    }
+}
+
+/// How long an upload's `.tmp` file must have been untouched before it is treated as abandoned.
+///
+/// This is an age on the MODIFICATION time, not on creation, and that is what makes an hour
+/// safe rather than reckless: a live upload is being written to continuously, so its mtime keeps
+/// advancing and it can never age into the sweep no matter how slow the connection. The clock
+/// only starts once the writer stops — i.e. once the upload is genuinely dead.
+const ORPHAN_TEMP_MAX_AGE: Duration = Duration::from_secs(3600);
+
+/// Reclaim `.tmp` files left in the media tree by uploads that never finished.
+///
+/// `stream_field_to_file` removes its temp file on every error return, which covers everything
+/// the handler can see. It cannot cover the case that actually happens at a party: the client
+/// simply goes away — a phone sleeps, a guest walks out of range, the PWA is evicted mid-video —
+/// and axum DROPS the handler future rather than returning an error, so no cleanup code runs at
+/// all. The shutdown backstop force-exits in-flight handlers for the same net effect.
+///
+/// Nothing else reclaims these. `cleanup_deleted_media` only visits rows with `deleted_at`, and
+/// an abandoned upload never got a row; `export::sweep_orphan_temps` is only ever pointed at the
+/// exports volume. So before this, every abandonment stranded up to `max_video_size_mb` of
+/// unowned bytes permanently — and worse than merely leaking, they were subtracted from what
+/// everyone else could upload, because the per-user quota is computed from live free disk
+/// (`compute_storage_quota`). On a 40 GB disk shared with `postgres_data`, an evening of flaky
+/// venue wifi could take the event down.
+async fn sweep_orphan_upload_temps(media_path: &std::path::Path) {
+    let originals = media_path.join("originals");
+    let mut event_dirs = match tokio::fs::read_dir(&originals).await {
+        Ok(d) => d,
+        // Absent before the first upload — not a problem worth logging every hour.
+        Err(e) if e.kind() == std::io::ErrorKind::NotFound => return,
+        Err(e) => {
+            tracing::warn!(error = ?e, path = %originals.display(), "orphan temp sweep: unreadable");
+            return;
+        }
+    };
+
+    let mut reclaimed = 0usize;
+    let mut bytes = 0u64;
+    while let Ok(Some(event_dir)) = event_dirs.next_entry().await {
+        let Ok(mut files) = tokio::fs::read_dir(event_dir.path()).await else {
+            continue;
+        };
+        while let Ok(Some(file)) = files.next_entry().await {
+            let path = file.path();
+            if path.extension().is_none_or(|e| e != "tmp") {
+                continue;
+            }
+            let Ok(meta) = file.metadata().await else {
+                continue;
+            };
+            // No mtime (or a clock that moved backwards) means we cannot show the file is
+            // abandoned, and deleting a live upload is far worse than leaking one temp file.
+            let abandoned = meta
+                .modified()
+                .ok()
+                .and_then(|m| m.elapsed().ok())
+                .is_some_and(|age| age >= ORPHAN_TEMP_MAX_AGE);
+            if !abandoned {
+                continue;
+            }
+            match tokio::fs::remove_file(&path).await {
+                Ok(()) => {
+                    reclaimed += 1;
+                    bytes += meta.len();
+                }
+                Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
+                Err(e) => {
+                    tracing::warn!(error = ?e, path = %path.display(),
+                        "orphan temp sweep: could not reclaim")
+                }
+            }
+        }
+    }
+
+    if reclaimed > 0 {
+        tracing::info!(
+            "orphan temp sweep: reclaimed {reclaimed} abandoned upload temp file(s), {} MiB",
+            bytes / (1024 * 1024)
+        );
+    }
+}
+
 /// Reclaim the media of soft-deleted uploads once they are past their retention window.
 ///
 /// ONLY ever touches rows with `deleted_at IS NOT NULL`, so it can never reach a live upload. Two
@@ -262,3 +378,72 @@ async fn cleanup_sessions(pool: &PgPool) {
         Err(e) => tracing::warn!("session cleanup failed: {e:#}"),
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    /// Build `<root>/originals/<event>/<name>` with `len` bytes, optionally back-dating its mtime
+    /// by `age`. Back-dating is the only way to test the sweep without sleeping through an hour.
+    fn temp_file(root: &std::path::Path, name: &str, len: usize, age: Option<Duration>) {
+        let dir = root.join("originals").join("wedding");
+        std::fs::create_dir_all(&dir).expect("create dir");
+        let path = dir.join(name);
+        let f = std::fs::File::create(&path).expect("create file");
+        std::io::Write::write_all(&mut &f, &vec![0u8; len]).expect("write");
+        if let Some(age) = age {
+            let when = std::time::SystemTime::now() - age;
+            f.set_modified(when).expect("set mtime");
+        }
+    }
+
+    fn exists(root: &std::path::Path, name: &str) -> bool {
+        root.join("originals").join("wedding").join(name).exists()
+    }
+
+    /// The two halves of the guarantee in one pass: an abandoned temp is reclaimed, and a temp
+    /// that is still being written to is NOT — the second matters more, because deleting a live
+    /// upload's temp file would corrupt a photo that was about to succeed.
+    #[tokio::test]
+    async fn the_sweep_reclaims_abandoned_temps_and_spares_live_ones() {
+        let root = std::env::temp_dir().join(format!("es-sweep-{}", uuid::Uuid::new_v4()));
+
+        // Abandoned: the writer died over an hour ago and nothing has touched it since.
+        temp_file(
+            &root,
+            "dead.tmp",
+            2048,
+            Some(ORPHAN_TEMP_MAX_AGE + Duration::from_secs(60)),
+        );
+        // Live: an upload in progress keeps advancing its mtime, so it always looks young —
+        // this is why the threshold is on modification time and not on creation time.
+        temp_file(&root, "inflight.tmp", 2048, None);
+        // A committed original. The sweep must only ever consider `.tmp`.
+        temp_file(&root, "keeper.jpg", 2048, Some(Duration::from_secs(86_400)));
+
+        sweep_orphan_upload_temps(&root).await;
+
+        assert!(
+            !exists(&root, "dead.tmp"),
+            "an abandoned temp must be reclaimed"
+        );
+        assert!(
+            exists(&root, "inflight.tmp"),
+            "a temp still being written to must survive — deleting it destroys a live upload"
+        );
+        assert!(
+            exists(&root, "keeper.jpg"),
+            "the sweep must never touch a committed original"
+        );
+
+        let _ = std::fs::remove_dir_all(&root);
+    }
+
+    /// Runs on every boot and every hour, so a media tree that does not exist yet (before the
+    /// first upload) must be a silent no-op rather than an error logged 24 times a day.
+    #[tokio::test]
+    async fn a_missing_media_tree_is_not_an_error() {
+        let root = std::env::temp_dir().join(format!("es-sweep-absent-{}", uuid::Uuid::new_v4()));
+        sweep_orphan_upload_temps(&root).await; // must simply return
+    }
+}