diff --git a/backend/src/handlers/upload.rs b/backend/src/handlers/upload.rs index 0422e12..35f82bb 100644 --- a/backend/src/handlers/upload.rs +++ b/backend/src/handlers/upload.rs @@ -17,6 +17,62 @@ use crate::state::AppState; const MAX_CAPTION_LENGTH: usize = 2000; +/// Owns the bytes an in-flight upload has written to disk, and deletes them unless the +/// request reaches the point where a database row takes ownership. +/// +/// Reclaim used to be a dozen explicit `remove_file` calls on the handler's return paths. +/// That covers every way the handler can FINISH, and none of the ways it can simply STOP: +/// when a client disconnects mid-body — a phone leaving wifi, iOS killing a backgrounded +/// PWA, the user hitting back — axum drops the handler future at a `.await` inside +/// `field.chunk()`, and no return path runs at all. The partial file then survives forever: +/// it has no upload row, so `cleanup_deleted_media` (which is row-driven) can never see it, +/// and no sweeper existed for the originals directory. Those bytes are also invisible to the +/// quota while still consuming the free disk that `quota_limit_bytes` divides among guests. +/// +/// A drop guard is the only construct that survives cancellation, because dropping the future +/// is exactly what runs it. +struct TempFileGuard { + /// `None` once disarmed — a row now owns these bytes. + path: Option, +} + +impl TempFileGuard { + fn new(path: std::path::PathBuf) -> Self { + Self { path: Some(path) } + } + + /// Follow the bytes to their new location after a rename. + /// + /// NOT `disarm`. Between the rename and the commit the file exists under its FINAL name + /// with still no row pointing at it, so that window needs guarding just as much as the + /// `.tmp` did — arguably more, since a leftover final-named original looks legitimate. + fn retarget(&mut self, path: std::path::PathBuf) { + self.path = Some(path); + } + + /// Hand ownership to the committed row. Only correct after `tx.commit()` succeeds. + fn disarm(&mut self) { + self.path = None; + } +} + +impl Drop for TempFileGuard { + fn drop(&mut self) { + let Some(path) = self.path.take() else { + return; + }; + // std::fs, not tokio::fs: `Drop` cannot await, and a runtime-dependent unlink is not + // guaranteed a live runtime here (shutdown drops in-flight tasks). + match std::fs::remove_file(&path) { + Ok(()) => tracing::debug!(path = %path.display(), "reclaimed an abandoned upload"), + Err(e) if e.kind() == std::io::ErrorKind::NotFound => {} + Err(e) => { + tracing::warn!(error = ?e, path = %path.display(), "failed to reclaim an abandoned upload") + } + } + } +} + /// 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 @@ -107,13 +163,18 @@ pub async fn upload( .media_path .join(format!("originals/{event_slug}")); let temp_abs = originals_dir.join(format!("{upload_id}.tmp")); + // Armed before anything can create the file, so there is no window in which bytes exist + // unowned. From here on, EVERY exit — return, `?`, panic, or the future being dropped + // mid-body by a client disconnect — reclaims them, and the explicit `remove_file` calls + // that used to be sprinkled over the return paths are gone. One owner, one rule. + let mut file_guard = TempFileGuard::new(temp_abs.clone()); let mut streamed: Option<(i64, Vec)> = None; // (size, head bytes for sniffing) let mut caption: Option = None; let mut hashtags_csv: Option = None; - // Wrap the multipart read so any error after the temp file is created still cleans - // it up (a mid-stream parse failure must not leave a stray `.tmp` on disk). + // The multipart read is wrapped so the field loop can use `?` freely; reclaiming the temp + // file on failure is `file_guard`'s job, not this block's. let parse_result: Result<(), AppError> = async { while let Some(field) = multipart .next_field() @@ -165,13 +226,10 @@ pub async fn upload( } .await; - if let Err(e) = parse_result { - let _ = tokio::fs::remove_file(&temp_abs).await; - return Err(e); - } + parse_result?; - // From here on the temp file may exist; every validation failure removes it before - // returning so a rejected upload never leaves bytes behind. + // From here on the temp file may exist. Every exit reclaims it via `file_guard` — see + // TempFileGuard for why the explicit per-branch cleanup this replaced was not enough. let (size, head) = match streamed { Some(s) => s, None => return Err(AppError::BadRequest("Keine Datei hochgeladen.".into())), @@ -183,7 +241,6 @@ pub async fn upload( if let Some(ref cap) = caption && cap.chars().count() > MAX_CAPTION_LENGTH { - let _ = tokio::fs::remove_file(&temp_abs).await; return Err(AppError::BadRequest(format!( "Beschreibung ist zu lang. Maximum: {} Zeichen.", MAX_CAPTION_LENGTH @@ -198,7 +255,6 @@ pub async fn upload( let kind = match infer::get(&head) { Some(k) => k, None => { - let _ = tokio::fs::remove_file(&temp_abs).await; return Err(AppError::BadRequest( "Dateityp nicht erkannt oder nicht unterstützt.".into(), )); @@ -211,7 +267,6 @@ pub async fn upload( { Some(v) => v, None => { - let _ = tokio::fs::remove_file(&temp_abs).await; return Err(AppError::BadRequest(format!( "Dateityp wird nicht unterstützt: {}.", kind.mime_type() @@ -226,7 +281,6 @@ pub async fn upload( max_image_mb * 1024 * 1024 }; if size > max_bytes { - let _ = tokio::fs::remove_file(&temp_abs).await; return Err(AppError::BadRequest(format!( "Datei ist zu groß. Maximum: {} MB.", max_bytes / (1024 * 1024) @@ -245,7 +299,6 @@ pub async fn upload( %mime, megapixels = ?mp, "rejecting an image that exceeds the decode budget at admission" ); - let _ = tokio::fs::remove_file(&temp_abs).await; let detail = mp.map_or(String::new(), |mp| format!(" (ca. {mp:.0} Megapixel)")); return Err(AppError::BadRequest(format!( "Bild hat zu viele Bildpunkte{detail} und kann nicht verarbeitet werden. \ @@ -271,7 +324,6 @@ pub async fn upload( quota_limit = Some(limit); let prospective_total = user.total_upload_bytes.saturating_add(size); if prospective_total > limit { - let _ = tokio::fs::remove_file(&temp_abs).await; return Err(AppError::QuotaExceeded( "Du hast dein Upload-Limit für dieses Event erreicht.".into(), )); @@ -286,6 +338,11 @@ pub async fn upload( tokio::fs::rename(&temp_abs, &absolute_path) .await .map_err(|e| AppError::Internal(e.into()))?; + // THERE MUST BE NO `.await` BETWEEN THE RENAME AND THIS LINE. Both statements resolve on + // the same poll, so the future cannot be dropped between them and the guard is never + // pointing at a path that no longer holds the bytes. If the rename fails the guard still + // owns `temp_abs`, which is why retargeting comes after it rather than before. + file_guard.retarget(absolute_path.clone()); // Process hashtags from caption and explicit CSV let mut tags: Vec = Vec::new(); @@ -386,15 +443,10 @@ pub async fn upload( } .await; - // The file is already on disk at `absolute_path`. If the transaction failed, no DB - // row will ever reference it, so remove it now rather than orphan bytes on disk. - let upload = match tx_result { - Ok(u) => u, - Err(e) => { - let _ = tokio::fs::remove_file(&absolute_path).await; - return Err(e); - } - }; + // The committed row now references these bytes — hand ownership over. Anything other than + // a successful commit leaves the guard armed, so the file is reclaimed on the way out. + let upload = tx_result?; + file_guard.disarm(); // Spawn compression task state @@ -1078,4 +1130,54 @@ mod tests { fn full_tolerance_is_identity_for_a_single_uploader() { assert_eq!(quota_limit_bytes(500, 1.0, 1), 500); } + + /// The guard is the only reclaim mechanism that survives a client disconnect, so its three + /// states have to be exactly right — a wrong `disarm` leaks bytes forever, a wrong `Drop` + /// deletes a committed guest's photo. + mod temp_file_guard { + use super::super::TempFileGuard; + + fn scratch(name: &str) -> std::path::PathBuf { + let dir = std::env::temp_dir().join(format!("es-guard-{}", std::process::id())); + std::fs::create_dir_all(&dir).unwrap(); + let p = dir.join(name); + std::fs::write(&p, b"bytes").unwrap(); + p + } + + #[test] + fn dropping_an_armed_guard_reclaims_the_file() { + let p = scratch("armed.tmp"); + drop(TempFileGuard::new(p.clone())); + assert!(!p.exists(), "an abandoned upload must not survive the request"); + } + + #[test] + fn a_disarmed_guard_leaves_the_file_alone() { + let p = scratch("committed.jpg"); + let mut g = TempFileGuard::new(p.clone()); + g.disarm(); + drop(g); + assert!(p.exists(), "a committed upload's bytes must never be deleted"); + } + + #[test] + fn retarget_follows_the_rename_and_forgets_the_old_path() { + let old = scratch("old.tmp"); + let new = scratch("new.jpg"); + let mut g = TempFileGuard::new(old.clone()); + // The rename already moved the bytes; only the new path is at risk now. + std::fs::remove_file(&old).unwrap(); + g.retarget(new.clone()); + drop(g); + assert!(!new.exists(), "the final-named original is orphaned too until the row commits"); + } + + #[test] + fn a_guard_whose_file_is_already_gone_is_harmless() { + let p = scratch("vanished.tmp"); + std::fs::remove_file(&p).unwrap(); + drop(TempFileGuard::new(p)); // must not panic + } + } } diff --git a/backend/src/services/maintenance.rs b/backend/src/services/maintenance.rs index b59b046..c202630 100644 --- a/backend/src/services/maintenance.rs +++ b/backend/src/services/maintenance.rs @@ -120,6 +120,19 @@ pub async fn startup_recovery(pool: &PgPool) { } } +/// How long a file in `originals/` may exist without a database row before it is treated as +/// abandoned. +/// +/// This window is the ONLY thing making the sweep safe, because the upload handler renames the +/// temp file into its final path BEFORE committing the row: for a short moment a perfectly +/// healthy upload legitimately looks exactly like an orphan. Six hours is far beyond any live +/// request (a 576 MiB body over a bad venue uplink is minutes, and the request itself is bounded +/// by the reverse proxy) while still reclaiming the leak inside a single event. +/// +/// DO NOT SHORTEN THIS to make a test faster — a value below the longest possible in-flight +/// upload deletes photos out from under the request that is committing them. +const ORPHAN_UPLOAD_RETENTION_HOURS: u64 = 6; + /// Spawns a background task that periodically: /// - deletes session rows whose `expires_at` is more than a day in the past /// - prunes the in-memory rate-limiter HashMap of empty windows @@ -140,6 +153,7 @@ pub fn spawn_periodic_tasks( tick.tick().await; cleanup_sessions(&pool).await; cleanup_deleted_media(&pool, &media_path).await; + sweep_orphan_originals(&pool, &media_path).await; rate_limiter.prune(); sse_tickets.prune(); } @@ -262,3 +276,123 @@ async fn cleanup_sessions(pool: &PgPool) { Err(e) => tracing::warn!("session cleanup failed: {e:#}"), } } + +/// Reclaim files in `originals/` that no upload row references. +/// +/// The backstop behind [`TempFileGuard`](crate::handlers::upload). The guard covers the +/// process that is running; this covers the process that was killed — a SIGKILL, an OOM, or a +/// power cut leaves whatever bytes had been written with no `Drop` to reclaim them, and those +/// files are then permanently invisible: they have no row, so `cleanup_deleted_media` (which is +/// row-driven) can never see them, and they are not counted against any quota while still +/// consuming the free disk that `compute_storage_quota` divides among guests. On a single box +/// where all three volumes share a filesystem, that ends with Postgres unable to write WAL. +/// +/// Two classes: +/// - `*.tmp` — an upload that never got as far as being renamed. Always safe past the window. +/// - everything else — a final-named original whose commit never happened. +async fn sweep_orphan_originals(pool: &PgPool, media_path: &std::path::Path) { + let originals = media_path.join("originals"); + let cutoff = Duration::from_secs(ORPHAN_UPLOAD_RETENTION_HOURS * 3600); + + // originals/{event_slug}/{uuid}.{ext} — one level of per-event directories. + let mut event_dirs = match tokio::fs::read_dir(&originals).await { + Ok(rd) => rd, + // Nothing uploaded yet; the directory is created lazily by the upload handler. + Err(_) => return, + }; + + let mut candidates: Vec<(String, std::path::PathBuf)> = Vec::new(); + let mut temps_removed = 0u32; + + while let Ok(Some(event_dir)) = event_dirs.next_entry().await { + if !event_dir + .file_type() + .await + .map(|t| t.is_dir()) + .unwrap_or(false) + { + continue; + } + let slug = event_dir.file_name().to_string_lossy().to_string(); + let Ok(mut files) = tokio::fs::read_dir(event_dir.path()).await else { + continue; + }; + while let Ok(Some(entry)) = files.next_entry().await { + let Ok(meta) = entry.metadata().await else { + continue; + }; + if !meta.is_file() { + continue; + } + // Too young to judge: an upload committing RIGHT NOW is indistinguishable from an + // orphan, because the rename precedes the commit. + let recent = meta + .modified() + .ok() + .and_then(|m| m.elapsed().ok()) + .is_none_or(|age| age < cutoff); + if recent { + continue; + } + + let name = entry.file_name().to_string_lossy().to_string(); + if name.ends_with(".tmp") { + // A `.tmp` never has a row by construction — no DB check needed. + if tokio::fs::remove_file(entry.path()).await.is_ok() { + temps_removed += 1; + } + continue; + } + candidates.push((format!("originals/{slug}/{name}"), entry.path())); + } + } + + if temps_removed > 0 { + tracing::warn!( + "reclaimed {temps_removed} abandoned upload temp file(s) older than \ + {ORPHAN_UPLOAD_RETENTION_HOURS}h" + ); + } + if candidates.is_empty() { + return; + } + + // One query per batch, not one per file: a backlog of thousands of orphans must not turn + // into thousands of round trips on an hourly timer. + let mut orphans_removed = 0u32; + for chunk in candidates.chunks(500) { + let paths: Vec = chunk.iter().map(|(rel, _)| rel.clone()).collect(); + // NO `deleted_at IS NULL` FILTER HERE. A soft-deleted row still points at its file + // during its retention window, and reclaiming that file is `cleanup_deleted_media`'s + // job — filtering here would race the two sweeps and destroy the exact files the + // recovery window exists to preserve. + let unreferenced: Result, _> = sqlx::query_as( + "SELECT p FROM unnest($1::text[]) AS p + WHERE NOT EXISTS (SELECT 1 FROM upload u WHERE u.original_path = p)", + ) + .bind(&paths) + .fetch_all(pool) + .await; + let unreferenced = match unreferenced { + Ok(rows) => rows, + Err(e) => { + tracing::warn!(error = ?e, "orphan-original sweep query failed"); + return; + } + }; + for (rel,) in unreferenced { + if let Some((_, abs)) = chunk.iter().find(|(r, _)| *r == rel) + && tokio::fs::remove_file(abs).await.is_ok() + { + orphans_removed += 1; + } + } + } + + if orphans_removed > 0 { + tracing::warn!( + "reclaimed {orphans_removed} original(s) with no upload row, older than \ + {ORPHAN_UPLOAD_RETENTION_HOURS}h" + ); + } +}