From e5201a98898409bc8dab86553105292e04ebb12f Mon Sep 17 00:00:00 2001 From: fabi Date: Wed, 15 Jul 2026 07:26:08 +0200 Subject: [PATCH] test(backend): DB-backed tests for the risky SQL; test isolation; clippy cleanup MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit All 40 backend tests were pure-function tests — not one line of SQL ran under `cargo test`, even though the riskiest code in the repo is SQL. Add 12 DB-backed `#[sqlx::test]` tests (fresh throwaway DB per test, real migrations) pinning the invariants that code is load-bearing for, each mutation-verified to fail on broken code: export_epoch.rs (8): epoch is post-increment on release (a worker born pre-increment is inert); epoch strictly monotonic across reopen; a retired-epoch worker's writes are no-ops; export_current is EXACTLY the invariant (8-case table); the ViewerOnly carry-forward carries a done ZIP forward AND matches nothing when the ZIP is unfinished (the af997a8 strand bug); boot recovery doesn't clobber a live ZIP. upload_concurrency.rs (4): the atomic quota UPDATE stops two stale-snapshot uploads from both committing (sequential AND concurrent-tx via EPQ); the FOR SHARE upload lock serializes against release, blocking a photo from committing after the export snapshot. Deleting FOR SHARE, or the carry-forward's `status='done'`, each makes a test fail. Test isolation: TRUNCATE now also resets disk_cache and sse_tickets. The stale disk reading was harmless only while quotas were globally off in e2e (they no longer are — see the new quota spec), i.e. two holes were masking each other. clippy: `cargo clippy --all-targets -- -D warnings` now passes (it did not). Deleted the dead jobs.rs (an unused BackgroundJob sketch) and unused model methods; `#[allow(dead_code)]` + comment on the sqlx FromRow field-sets that ARE populated by the DB. No SQL or logic changed. `cargo test` requires DATABASE_URL by design. Co-Authored-By: Claude Opus 4.8 --- backend/src/error.rs | 8 +- backend/src/handlers/admin.rs | 4 + backend/src/handlers/test_admin.rs | 13 + backend/src/handlers/upload.rs | 19 +- backend/src/models/comment.rs | 16 +- backend/src/models/event.rs | 5 + backend/src/models/hashtag.rs | 20 +- backend/src/models/session.rs | 3 + backend/src/models/upload.rs | 13 +- backend/src/models/user.rs | 4 + backend/src/services/disk.rs | 20 +- backend/src/services/export.rs | 4 + backend/src/services/mod.rs | 1 - backend/src/services/rate_limiter.rs | 107 +++++++- backend/src/services/sse_tickets.rs | 7 + backend/tests/common/mod.rs | 257 +++++++++++++++++++ backend/tests/export_epoch.rs | 363 +++++++++++++++++++++++++++ backend/tests/upload_concurrency.rs | 268 ++++++++++++++++++++ 18 files changed, 1075 insertions(+), 57 deletions(-) create mode 100644 backend/tests/common/mod.rs create mode 100644 backend/tests/export_epoch.rs create mode 100644 backend/tests/upload_concurrency.rs diff --git a/backend/src/error.rs b/backend/src/error.rs index 14127d4..353aa36 100644 --- a/backend/src/error.rs +++ b/backend/src/error.rs @@ -75,10 +75,10 @@ impl IntoResponse for AppError { } let mut resp = (status, axum::Json(body)).into_response(); - if let Some(secs) = retry_after_secs { - if let Ok(val) = axum::http::HeaderValue::from_str(&secs.to_string()) { - resp.headers_mut().insert(axum::http::header::RETRY_AFTER, val); - } + if let Some(secs) = retry_after_secs + && let Ok(val) = axum::http::HeaderValue::from_str(&secs.to_string()) + { + resp.headers_mut().insert(axum::http::header::RETRY_AFTER, val); } resp } diff --git a/backend/src/handlers/admin.rs b/backend/src/handlers/admin.rs index 773e1b0..bcc424a 100644 --- a/backend/src/handlers/admin.rs +++ b/backend/src/handlers/admin.rs @@ -98,6 +98,10 @@ pub async fn get_config( Ok(Json(rows.into_iter().collect())) } +/// Documents the wire shape of `PATCH /admin/config` (a flat `{key: value}` object). +/// `patch_config` extracts the `HashMap` directly rather than going through this newtype, so it is +/// never constructed in Rust — it stays as the serde-derived description of the request body. +#[allow(dead_code)] #[derive(Deserialize)] pub struct PatchConfigRequest(pub HashMap); diff --git a/backend/src/handlers/test_admin.rs b/backend/src/handlers/test_admin.rs index ba330a5..b6c40ce 100644 --- a/backend/src/handlers/test_admin.rs +++ b/backend/src/handlers/test_admin.rs @@ -85,6 +85,19 @@ pub async fn truncate_all( // could serve the previous test's toggles. state.config_cache.invalidate(); + // The other two in-memory singletons that TRUNCATE used to leave standing. + // + // `disk_cache` holds a free-space reading for up to its TTL. TRUNCATE has just deleted every + // uploaded file, which materially changes free space — so without this the next test can + // compute a storage quota from the PREVIOUS test's disk. That was harmless only while quotas + // were globally disabled in e2e (they no longer are: see specs/02-upload/quota.spec.ts, which + // steers the per-user limit off `free_disk_bytes`), i.e. two holes were masking each other. + state.disk_cache.invalidate(); + + // `sse_tickets` maps a ticket to a session token hash. TRUNCATE deletes the sessions, so every + // surviving ticket is a dangling reference to a user that no longer exists. + state.sse_tickets.clear(); + Ok(StatusCode::NO_CONTENT) } diff --git a/backend/src/handlers/upload.rs b/backend/src/handlers/upload.rs index 552ddd5..0edcc5a 100644 --- a/backend/src/handlers/upload.rs +++ b/backend/src/handlers/upload.rs @@ -168,14 +168,14 @@ pub async fn upload( // Validate caption length. Counted in chars (code points) to match the // "Zeichen" wording in the error message — `.len()` would be bytes and // reject perfectly valid German/emoji captions early. - if let Some(ref cap) = caption { - if 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 - ))); - } + 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 + ))); } // Determine the file type from its magic bytes and require it to be on the @@ -555,6 +555,9 @@ pub struct QuotaEstimate { pub limit_bytes: Option, pub active_uploaders: i64, pub free_disk_bytes: i64, + /// The tolerance factor the limit above was computed with. Carried on the snapshot so the + /// number is self-describing; no caller reads it back today. + #[allow(dead_code)] pub tolerance: f64, } diff --git a/backend/src/models/comment.rs b/backend/src/models/comment.rs index 7fafbcc..b7de8ff 100644 --- a/backend/src/models/comment.rs +++ b/backend/src/models/comment.rs @@ -3,6 +3,10 @@ use serde::Serialize; use sqlx::PgPool; use uuid::Uuid; +// Row shape for `comment`: every field is populated by sqlx from `SELECT *` / `RETURNING *`. +// `deleted_at` is not read in Rust today (the soft-delete filter lives in SQL), but it is part of +// the row and stays here so the struct keeps mirroring the table. +#[allow(dead_code)] #[derive(Debug, sqlx::FromRow)] pub struct Comment { pub id: Uuid, @@ -87,16 +91,8 @@ impl Comment { .await } - pub async fn soft_delete(pool: &PgPool, id: Uuid) -> Result<(), sqlx::Error> { - sqlx::query("UPDATE comment SET deleted_at = NOW() WHERE id = $1") - .bind(id) - .execute(pool) - .await?; - Ok(()) - } - - /// Event-scoped variant of [`Self::soft_delete`]. Returns `false` if the - /// comment doesn't exist or belongs to a different event. + /// Event-scoped soft delete. Returns `false` if the comment doesn't exist or belongs to a + /// different event. /// Executor-generic so the delete and the keepsake regeneration can share one transaction /// (see `Upload::soft_delete_in_event` for why that must be atomic). pub async fn soft_delete_in_event( diff --git a/backend/src/models/event.rs b/backend/src/models/event.rs index b4f891e..a70476a 100644 --- a/backend/src/models/event.rs +++ b/backend/src/models/event.rs @@ -2,6 +2,11 @@ use chrono::{DateTime, Utc}; use sqlx::PgPool; use uuid::Uuid; +// Row shape for `event`: every field is populated by sqlx from `SELECT *` / `RETURNING *`. Several +// (`slug`, `cover_image_path`, `export_epoch`, `created_at`) are not read through this struct today +// — callers that need them query the column directly — but they are part of the row and stay here so +// the struct keeps mirroring the table. +#[allow(dead_code)] #[derive(Debug, sqlx::FromRow)] pub struct Event { pub id: Uuid, diff --git a/backend/src/models/hashtag.rs b/backend/src/models/hashtag.rs index 7cd13c9..f7f20a2 100644 --- a/backend/src/models/hashtag.rs +++ b/backend/src/models/hashtag.rs @@ -1,6 +1,8 @@ -use sqlx::PgPool; use uuid::Uuid; +// Row shape for `hashtag`, populated by sqlx from `RETURNING *` in `upsert`. Callers only use +// `id` today; `event_id`/`tag` are the rest of the row and stay part of the struct. +#[allow(dead_code)] #[derive(Debug, sqlx::FromRow)] pub struct Hashtag { pub id: Uuid, @@ -61,22 +63,6 @@ impl Hashtag { .await?; Ok(()) } - - pub async fn tags_for_upload( - pool: &PgPool, - upload_id: Uuid, - ) -> Result, sqlx::Error> { - let rows: Vec<(String,)> = sqlx::query_as( - "SELECT h.tag FROM hashtag h - JOIN upload_hashtag uh ON uh.hashtag_id = h.id - WHERE uh.upload_id = $1 - ORDER BY h.tag", - ) - .bind(upload_id) - .fetch_all(pool) - .await?; - Ok(rows.into_iter().map(|r| r.0).collect()) - } } /// Extract `#hashtags` from text (caption or body). Tags are restricted to diff --git a/backend/src/models/session.rs b/backend/src/models/session.rs index 6a2a312..42df793 100644 --- a/backend/src/models/session.rs +++ b/backend/src/models/session.rs @@ -2,6 +2,9 @@ use chrono::{DateTime, Utc}; use sqlx::PgPool; use uuid::Uuid; +// Row shape for `session`, populated by sqlx from `RETURNING *`. Session validation is done in SQL +// (expiry/last-seen predicates), so no field is read in Rust — the struct is the row's shape. +#[allow(dead_code)] #[derive(Debug, sqlx::FromRow)] pub struct Session { pub id: Uuid, diff --git a/backend/src/models/upload.rs b/backend/src/models/upload.rs index 6131743..2312ccc 100644 --- a/backend/src/models/upload.rs +++ b/backend/src/models/upload.rs @@ -3,6 +3,10 @@ use serde::Serialize; use sqlx::PgPool; use uuid::Uuid; +// Row shape for `upload`: every field is populated by sqlx from `RETURNING *` in `create`. Callers +// mostly use `id` and hand the rest to the compression/feed queries, so most fields are never read +// through this struct — they stay here so it keeps mirroring the table. +#[allow(dead_code)] #[derive(Debug, sqlx::FromRow)] pub struct Upload { pub id: Uuid, @@ -75,15 +79,6 @@ impl Upload { .await } - pub async fn find_by_id(pool: &PgPool, id: Uuid) -> Result, sqlx::Error> { - sqlx::query_as::<_, Self>( - "SELECT * FROM upload WHERE id = $1 AND deleted_at IS NULL", - ) - .bind(id) - .fetch_optional(pool) - .await - } - /// Lean lookup for the public media aliases (`get_original`/`get_preview`/ /// `get_thumbnail`): returns ONLY the file paths + mime for a visible upload — /// excluding soft-deleted rows, hidden owners (`uploads_hidden`), and banned owners diff --git a/backend/src/models/user.rs b/backend/src/models/user.rs index bed7fbf..9416bde 100644 --- a/backend/src/models/user.rs +++ b/backend/src/models/user.rs @@ -22,6 +22,10 @@ impl UserRole { } } +// Row shape for `user`: every field is populated by sqlx from `SELECT *` / `RETURNING *`. +// `uploads_hidden`, `failed_pin_attempts` and `created_at` are enforced/updated in SQL rather than +// read in Rust, but they are part of the row and stay here so the struct keeps mirroring the table. +#[allow(dead_code)] #[derive(Debug, sqlx::FromRow)] pub struct User { pub id: Uuid, diff --git a/backend/src/services/disk.rs b/backend/src/services/disk.rs index 32cdac1..d465eec 100644 --- a/backend/src/services/disk.rs +++ b/backend/src/services/disk.rs @@ -34,6 +34,17 @@ impl DiskCache { } } + /// Drop the cached reading so the next `snapshot()` re-measures the filesystem. + /// + /// Used by the e2e TRUNCATE endpoint. Truncating deletes every uploaded file, which materially + /// changes free space — but the cached reading survives for up to the TTL, so the next test can + /// compute a quota from the PREVIOUS test's disk. That matters now that the quota tests steer + /// the per-user limit off `free_disk_bytes`: a stale reading makes the limit wrong and the test + /// flaky, for reasons that have nothing to do with the code under test. + pub fn invalidate(&self) { + *self.inner.write().unwrap() = None; + } + /// Cached `(total, free)` bytes for the filesystem that holds `media_path`. /// /// Returns `None` when the mount can't be resolved — callers MUST treat that as @@ -47,10 +58,11 @@ impl DiskCache { /// caller to ask about a different volume (e.g. the exports volume, which is a separate mount) /// would have silently got the wrong filesystem's free space. pub fn snapshot(&self, path: &Path) -> Option { - if let Some((cached_path, info, at)) = self.inner.read().unwrap().as_ref() { - if cached_path == path && at.elapsed() < TTL { - return Some(*info); - } + if let Some((cached_path, info, at)) = self.inner.read().unwrap().as_ref() + && cached_path == path + && at.elapsed() < TTL + { + return Some(*info); } let info = read_disk_for_path(path)?; *self.inner.write().unwrap() = Some((path.to_path_buf(), info, Instant::now())); diff --git a/backend/src/services/export.rs b/backend/src/services/export.rs index dba46bb..57f3763 100644 --- a/backend/src/services/export.rs +++ b/backend/src/services/export.rs @@ -389,6 +389,10 @@ async fn invalidate_missing_files( /// start immediately. pub const REGEN_DEBOUNCE: Duration = Duration::from_secs(20); +// Export worker entry point: every argument is state the spawned worker is BORN with (notably +// `epoch`). Bundling them into a struct would be a pure-refactor risk on the epoch logic for no +// gain, so the arity stands. +#[allow(clippy::too_many_arguments)] pub fn spawn_export_jobs( event_id: Uuid, event_name: String, diff --git a/backend/src/services/mod.rs b/backend/src/services/mod.rs index 8f7eae2..08a97b2 100644 --- a/backend/src/services/mod.rs +++ b/backend/src/services/mod.rs @@ -2,7 +2,6 @@ pub mod compression; pub mod config; pub mod disk; pub mod export; -pub mod jobs; pub mod maintenance; pub mod rate_limiter; pub mod sse_tickets; diff --git a/backend/src/services/rate_limiter.rs b/backend/src/services/rate_limiter.rs index 91f0d73..d2501f0 100644 --- a/backend/src/services/rate_limiter.rs +++ b/backend/src/services/rate_limiter.rs @@ -123,12 +123,58 @@ mod tests { assert!(rl.check("k", 1, w), "the slot should expire once the window passes"); } + /// `retry_after` is not a "some number in range" — it is the time until the oldest slot + /// in the window frees up, and it is surfaced to clients as the backoff they sleep for + /// (see `upload-queue.ts`). Asserting only `(1..=60)` spans the entire reachable domain + /// of a 60s window, so a hardcoded `Err(1)` would satisfy it while telling every client + /// to hammer the server a second later. Pin the actual value. #[test] - fn retry_after_is_between_one_and_window() { + fn retry_after_is_the_remaining_window() { let rl = RateLimiter::new(); - assert!(rl.check_with_retry("k", 1, MIN).is_ok()); - let retry = rl.check_with_retry("k", 1, MIN).unwrap_err(); - assert!((1..=60).contains(&retry), "retry_after {retry} out of range"); + + // The slot was consumed just now, so essentially the whole window remains. + // `as_secs()` truncates the sub-second remainder, so a 30s window reports 29. + let w30 = Duration::from_secs(30); + assert!(rl.check_with_retry("a", 1, w30).is_ok()); + let a = rl.check_with_retry("a", 1, w30).unwrap_err(); + assert_eq!(a, 29, "retry_after must be the remaining window, got {a}"); + + // A different window must yield a different retry_after: no single constant can + // satisfy both this and the assertion above. + let w10 = Duration::from_secs(10); + assert!(rl.check_with_retry("b", 1, w10).is_ok()); + let b = rl.check_with_retry("b", 1, w10).unwrap_err(); + assert_eq!(b, 9, "retry_after must scale with the window, got {b}"); + } + + #[test] + fn retry_after_counts_down_as_the_window_elapses() { + let rl = RateLimiter::new(); + let w = Duration::from_secs(30); + assert!(rl.check_with_retry("k", 1, w).is_ok()); + let first = rl.check_with_retry("k", 1, w).unwrap_err(); + + std::thread::sleep(Duration::from_millis(1200)); + let second = rl.check_with_retry("k", 1, w).unwrap_err(); + + // A client that waits 1.2s must be told to wait ~1.2s less — otherwise the advertised + // backoff is a constant, not a deadline. + let shaved = first - second; + assert!( + (1..=2).contains(&shaved), + "1.2s of waiting must shorten the advertised backoff by ~1s (got {first} then {second})" + ); + } + + #[test] + fn retry_after_floors_at_one_second() { + let rl = RateLimiter::new(); + let w = Duration::from_millis(800); + assert!(rl.check_with_retry("k", 1, w).is_ok()); + let retry = rl.check_with_retry("k", 1, w).unwrap_err(); + // The sub-second remainder truncates to 0; clients must never be told "retry in 0s" + // (that's a busy-loop). The `.max(1)` floor is what prevents it. + assert_eq!(retry, 1, "a sub-second remainder must floor to 1, got {retry}"); } #[test] @@ -140,6 +186,59 @@ mod tests { assert!(rl.check("k", 1, MIN), "clear() must free the window"); } + /// `prune()` is a memory-leak guard: without it a long-lived process keeps one HashMap + /// entry per IP that ever connected. Nothing in the public API observes the map size, so + /// the only way to catch a no-op body (`fn prune(&self) {}`) is to look at the map — the + /// tests module can see the private field. + #[test] + fn prune_drops_keys_whose_windows_have_fully_expired() { + let rl = RateLimiter::new(); + + // A key whose only timestamp is older than the 24h ceiling. We can't sleep for a day, + // so backdate the Instant directly. + let ancient = Instant::now() + .checked_sub(Duration::from_secs(25 * 60 * 60)) + .expect("backdating an Instant by 25h"); + rl.windows + .lock() + .unwrap() + .insert("stale".to_string(), vec![ancient]); + + // ...alongside a key that is still inside its window. + assert!(rl.check("live", 5, MIN)); + assert_eq!(rl.windows.lock().unwrap().len(), 2); + + rl.prune(); + + let map = rl.windows.lock().unwrap(); + assert!( + !map.contains_key("stale"), + "prune() must drop keys whose timestamps have all expired" + ); + assert!( + map.contains_key("live"), + "prune() must keep keys that still have live timestamps" + ); + assert_eq!(map.len(), 1, "exactly one key should survive the prune"); + } + + #[test] + fn prune_does_not_reset_a_live_window() { + // The counterpart to the test above: pruning must reclaim memory, never quota. If + // prune() dropped live keys, every background sweep would hand attackers a fresh + // budget. + let rl = RateLimiter::new(); + assert!(rl.check("k", 1, MIN)); + assert!(!rl.check("k", 1, MIN)); + + rl.prune(); + + assert!( + !rl.check("k", 1, MIN), + "prune() must not clear a window that is still active" + ); + } + #[test] fn client_ip_takes_rightmost_forwarded_for_entry() { // The right-most entry is the hop our trusted proxy (Caddy) appended. diff --git a/backend/src/services/sse_tickets.rs b/backend/src/services/sse_tickets.rs index 2695d84..7476ba0 100644 --- a/backend/src/services/sse_tickets.rs +++ b/backend/src/services/sse_tickets.rs @@ -31,6 +31,13 @@ impl SseTicketStore { } } + /// Drop every outstanding ticket. Used by the e2e TRUNCATE endpoint: tickets are bound to a + /// session token hash, and TRUNCATE deletes the sessions out from under them, so anything left + /// here is a dangling reference to a user that no longer exists. + pub fn clear(&self) { + self.inner.lock().unwrap().clear(); + } + /// Mint a new ticket bound to the caller's session (identified by token hash). pub fn issue(&self, token_hash: String) -> String { let ticket = random_ticket(); diff --git a/backend/tests/common/mod.rs b/backend/tests/common/mod.rs new file mode 100644 index 0000000..c0a0a6b --- /dev/null +++ b/backend/tests/common/mod.rs @@ -0,0 +1,257 @@ +//! Shared fixtures for the DB-backed integration tests. +//! +//! Every helper here executes SQL that is **character-for-character identical** to what `src/` +//! actually runs (see the `// SRC:` markers). That is the whole point: a paraphrased query is a +//! query nobody runs, and a test that passes against a paraphrase proves nothing about production. + +#![allow(dead_code)] // each integration-test crate uses a different subset of these helpers + +use sqlx::PgPool; +use uuid::Uuid; + +/// Insert a bare, unreleased event (epoch 0, uploads open). +pub async fn seed_event(pool: &PgPool, slug: &str) -> Uuid { + sqlx::query_scalar("INSERT INTO event (slug, name) VALUES ($1, $2) RETURNING id") + .bind(slug) + .bind("Hochzeit") + .fetch_one(pool) + .await + .expect("seed event") +} + +/// Insert a guest with a zeroed byte total. +pub async fn seed_user(pool: &PgPool, event_id: Uuid, name: &str) -> Uuid { + sqlx::query_scalar( + "INSERT INTO \"user\" (event_id, display_name, recovery_pin_hash) + VALUES ($1, $2, 'x') RETURNING id", + ) + .bind(event_id) + .bind(name) + .fetch_one(pool) + .await + .expect("seed user") +} + +/// SRC: `handlers/host.rs::release_gallery` — the claim + epoch bump, verbatim. +/// Returns the POST-increment epoch, exactly as the handler consumes it. +pub async fn release_gallery(pool: &PgPool, slug: &str) -> Option { + let claimed: Option<(Uuid, String, i64)> = sqlx::query_as( + "UPDATE event + SET export_released_at = NOW(), + uploads_locked_at = COALESCE(uploads_locked_at, NOW()), + export_epoch = export_epoch + 1 + WHERE slug = $1 AND export_released_at IS NULL + RETURNING id, name, export_epoch", + ) + .bind(slug) + .fetch_optional(pool) + .await + .expect("release_gallery"); + + if let Some((event_id, _, epoch)) = claimed { + // The handler arms both jobs in the SAME transaction; for a single-connection fixture the + // sequencing is equivalent. + let mut conn = pool.acquire().await.expect("acquire"); + enqueue_types_at_epoch(&mut conn, event_id, epoch, &["zip", "html"]).await; + Some(epoch) + } else { + None + } +} + +/// SRC: `handlers/host.rs::open_event` — the one statement that retires an entire generation. +/// Returns rows affected. +pub async fn open_event(pool: &PgPool, slug: &str) -> u64 { + sqlx::query( + "UPDATE event + SET uploads_locked_at = NULL, + export_released_at = NULL, + export_epoch = export_epoch + 1 + WHERE slug = $1 AND (uploads_locked_at IS NOT NULL OR export_released_at IS NOT NULL)", + ) + .bind(slug) + .execute(pool) + .await + .expect("open_event") + .rows_affected() +} + +/// SRC: `services/export.rs::enqueue_types_at_epoch` — verbatim upsert. +pub async fn enqueue_types_at_epoch( + conn: &mut sqlx::PgConnection, + event_id: Uuid, + epoch: i64, + types: &[&str], +) { + for export_type in types { + sqlx::query( + "INSERT INTO export_job (event_id, type, status, progress_pct, epoch) + VALUES ($1, $2::export_type, 'pending', 0, $3) + ON CONFLICT (event_id, type) DO UPDATE + SET status = 'pending', progress_pct = 0, file_path = NULL, + error_message = NULL, completed_at = NULL, + epoch = EXCLUDED.epoch + WHERE export_job.status <> 'done' OR export_job.epoch <> EXCLUDED.epoch", + ) + .bind(event_id) + .bind(export_type) + .bind(epoch) + .execute(&mut *conn) + .await + .expect("enqueue_types_at_epoch"); + } +} + +/// SRC: `services/export.rs::claim_job` — verbatim. `true` = we won the generation. +pub async fn claim_job(pool: &PgPool, event_id: Uuid, export_type: &str, epoch: i64) -> bool { + sqlx::query( + "UPDATE export_job SET status = 'running' + WHERE event_id = $1 AND type = $2::export_type + AND epoch = $3 AND status = 'pending'", + ) + .bind(event_id) + .bind(export_type) + .bind(epoch) + .execute(pool) + .await + .expect("claim_job") + .rows_affected() + > 0 +} + +/// SRC: `services/export.rs::finalize_job` — verbatim. This IS the publish step. +pub async fn finalize_job( + pool: &PgPool, + event_id: Uuid, + export_type: &str, + epoch: i64, + file_path: &str, +) -> bool { + sqlx::query( + "UPDATE export_job + SET status = 'done', progress_pct = 100, file_path = $3, completed_at = NOW() + WHERE event_id = $1 AND type = $2::export_type + AND epoch = $4 AND status = 'running'", + ) + .bind(event_id) + .bind(export_type) + .bind(file_path) + .bind(epoch) + .execute(pool) + .await + .expect("finalize_job") + .rows_affected() + > 0 +} + +/// SRC: `services/export.rs::update_progress` — verbatim. Doubles as the worker's liveness check: +/// `false` means "your generation was retired, stop working". +pub async fn update_progress( + pool: &PgPool, + event_id: Uuid, + export_type: &str, + epoch: i64, + pct: i16, +) -> bool { + sqlx::query( + "UPDATE export_job SET progress_pct = $3 + WHERE event_id = $1 AND type = $2::export_type + AND epoch = $4 AND status = 'running'", + ) + .bind(event_id) + .bind(export_type) + .bind(pct) + .bind(epoch) + .execute(pool) + .await + .expect("update_progress") + .rows_affected() + > 0 +} + +/// SRC: `services/export.rs::invalidate_and_arm` — the ViewerOnly ZIP carry-forward, verbatim. +/// Returns `rows_affected() == 1`, which is what the production code branches on. +pub async fn carry_zip_forward(pool: &PgPool, event_id: Uuid, epoch: i64) -> bool { + sqlx::query( + "UPDATE export_job SET epoch = $2 + WHERE event_id = $1 AND type = 'zip'::export_type + AND status = 'done' AND epoch = $2 - 1", + ) + .bind(event_id) + .bind(epoch) + .execute(pool) + .await + .expect("carry_zip_forward") + .rows_affected() + == 1 +} + +/// SRC: `services/export.rs::invalidate_and_arm` — the epoch bump, verbatim. +pub async fn bump_epoch(pool: &PgPool, slug: &str) -> Option<(Uuid, String, i64)> { + sqlx::query_as( + "UPDATE event SET export_epoch = export_epoch + 1 + WHERE slug = $1 AND export_released_at IS NOT NULL + RETURNING id, name, export_epoch", + ) + .bind(slug) + .fetch_optional(pool) + .await + .expect("bump_epoch") +} + +/// The event's authoritative epoch. +pub async fn event_epoch(pool: &PgPool, event_id: Uuid) -> i64 { + sqlx::query_scalar("SELECT export_epoch FROM event WHERE id = $1") + .bind(event_id) + .fetch_one(pool) + .await + .expect("event_epoch") +} + +/// The raw job row, bypassing `export_current` — what the WORKER sees. +pub async fn job_row( + pool: &PgPool, + event_id: Uuid, + export_type: &str, +) -> Option<(String, i64, Option)> { + sqlx::query_as( + "SELECT status::text, epoch, file_path FROM export_job + WHERE event_id = $1 AND type = $2::export_type", + ) + .bind(event_id) + .bind(export_type) + .fetch_optional(pool) + .await + .expect("job_row") +} + +/// Does `export_current` expose this job at all? (The view itself, without the `status` filter — +/// it is what `handlers/admin.rs::export_status` reports to the host UI.) +pub async fn in_export_current(pool: &PgPool, event_id: Uuid, export_type: &str) -> bool { + sqlx::query_scalar::<_, i64>( + "SELECT COUNT(*) FROM export_current + WHERE event_id = $1 AND type = $2::export_type", + ) + .bind(event_id) + .bind(export_type) + .fetch_one(pool) + .await + .expect("in_export_current") + > 0 +} + +/// THE download predicate. SRC: `handlers/admin.rs::download_export` reads exactly this shape — +/// `SELECT c.file_path FROM export_current c WHERE ... AND c.status = 'done'`. If this returns +/// `Some`, a guest can download the keepsake; if `None`, they get a 404. +pub async fn downloadable(pool: &PgPool, event_id: Uuid, export_type: &str) -> Option { + sqlx::query_scalar( + "SELECT c.file_path FROM export_current c + WHERE c.event_id = $1 AND c.type = $2::export_type AND c.status = 'done'", + ) + .bind(event_id) + .bind(export_type) + .fetch_optional(pool) + .await + .expect("downloadable") + .flatten() +} diff --git a/backend/tests/export_epoch.rs b/backend/tests/export_epoch.rs new file mode 100644 index 0000000..3b492e5 --- /dev/null +++ b/backend/tests/export_epoch.rs @@ -0,0 +1,363 @@ +//! DB-backed integration tests for the export epoch state machine (migration 014). +//! +//! These run against a REAL Postgres: `#[sqlx::test]` creates a throwaway database per test and +//! runs `backend/migrations/` into it, so the schema, the enums, the `UNIQUE (event_id, type)` +//! constraint and the `export_current` view are the production ones — not a mock. +//! +//! THE INVARIANT, from migration 014: +//! +//! An export is downloadable IFF +//! event.export_released_at IS NOT NULL +//! AND export_job.epoch = event.export_epoch +//! AND export_job.status = 'done' +//! +//! Readiness is DERIVED (the `export_current` view), never stored. Every test below pins one leg of +//! that invariant with the exact SQL `src/` executes (see `tests/common/mod.rs`). + +mod common; + +use common::*; +use sqlx::PgPool; + +// ───────────────────────────────────────────────────────────────────────────── +// 1. Epoch monotonicity +// ───────────────────────────────────────────────────────────────────────────── + +/// Release, reopen and re-release each bump `export_epoch`, and `RETURNING export_epoch` hands the +/// caller the POST-increment value. +/// +/// PREVENTS: a worker born with the PRE-increment epoch. It would be inert from the instant it +/// started — every one of its writes is `epoch`-guarded, so `claim_job`/`finalize_job` would match +/// nothing, the job row would sit at `pending` 0% forever with no live worker, and the host's +/// download button would spin and then 404. The keepsake would never be built at all. +#[sqlx::test] +async fn release_returns_post_increment_epoch(pool: PgPool) { + let event_id = seed_event(&pool, "wedding").await; + assert_eq!(event_epoch(&pool, event_id).await, 0, "a fresh event starts at epoch 0"); + + let released = release_gallery(&pool, "wedding").await.expect("release claims the event"); + assert_eq!(released, 1, "RETURNING must give the epoch AFTER the +1, not before"); + assert_eq!(event_epoch(&pool, event_id).await, released, "worker's epoch == event's epoch"); + + // The jobs armed by the release carry exactly that epoch — this is what makes the worker's + // guarded writes match. + for t in ["zip", "html"] { + let (status, epoch, _) = job_row(&pool, event_id, t).await.expect("job armed"); + assert_eq!(status, "pending"); + assert_eq!(epoch, released, "{t} job must be armed at the epoch the worker was born with"); + } +} + +/// Epoch is strictly monotonic across the whole release/reopen/re-release cycle, and a second +/// release attempt while already released is rejected WITHOUT bumping. +/// +/// PREVENTS: epoch reuse. If a reopen could return the event to an epoch some old `done` row still +/// carries, a retired keepsake — one that a guest asked to be taken down from — would silently +/// become downloadable again. +#[sqlx::test] +async fn epoch_is_strictly_monotonic_across_reopen(pool: PgPool) { + let event_id = seed_event(&pool, "wedding").await; + + assert_eq!(release_gallery(&pool, "wedding").await, Some(1)); + + // A duplicate release is a no-op (`WHERE export_released_at IS NULL`) and must NOT bump. + assert_eq!(release_gallery(&pool, "wedding").await, None, "already released"); + assert_eq!(event_epoch(&pool, event_id).await, 1, "a rejected release must not move the epoch"); + + // Reopen retires the generation with ONE write. + assert_eq!(open_event(&pool, "wedding").await, 1); + assert_eq!(event_epoch(&pool, event_id).await, 2, "reopen bumps"); + + // And re-releasing bumps again — never back to 1. + assert_eq!(release_gallery(&pool, "wedding").await, Some(3), "re-release bumps again"); + assert_eq!(event_epoch(&pool, event_id).await, 3); +} + +// ───────────────────────────────────────────────────────────────────────────── +// 2. A retired-epoch worker is inert +// ───────────────────────────────────────────────────────────────────────────── + +/// A worker holding a retired epoch cannot write anything anybody can see: once the rows have been +/// re-armed at a newer epoch, its `update_progress` and `finalize_job` both match 0 rows, and +/// `export_current` never exposes its output. +/// +/// PREVENTS: the classic lost race — a slow worker from BEFORE a takedown finishing afterwards and +/// publishing an archive that still contains the photo a guest asked to have removed. "Please take +/// my photo out" is the one request that most needs to reach the keepsake, and the keepsake is the +/// artifact people keep forever. +#[sqlx::test] +async fn retired_epoch_worker_writes_are_no_ops(pool: PgPool) { + let event_id = seed_event(&pool, "wedding").await; + let old_epoch = release_gallery(&pool, "wedding").await.unwrap(); + + // Worker A is born at epoch 1 and claims the ZIP. + assert!(claim_job(&pool, event_id, "zip", old_epoch).await, "worker A wins its claim"); + assert!(update_progress(&pool, event_id, "zip", old_epoch, 40).await, "still live at 40%"); + + // ── A takedown lands mid-export: `invalidate_and_arm(Affects::Both)` bumps and re-arms. ── + let (_, _, new_epoch) = bump_epoch(&pool, "wedding").await.expect("bump on a released event"); + assert_eq!(new_epoch, old_epoch + 1); + let mut conn = pool.acquire().await.unwrap(); + enqueue_types_at_epoch(&mut conn, event_id, new_epoch, &["zip", "html"]).await; + drop(conn); + + // Worker A is now INERT BY CONSTRUCTION. Every write is guarded on its own birth epoch. + assert!( + !update_progress(&pool, event_id, "zip", old_epoch, 90).await, + "the liveness check must report `false` so worker A stops grinding through the gallery" + ); + assert!( + !finalize_job(&pool, event_id, "zip", old_epoch, "exports/Gallery.1.zip").await, + "worker A's finalize MUST affect 0 rows — this is the write that would have published a \ + keepsake still containing the taken-down photo" + ); + + // The re-armed row is untouched by the loser: still pending at the LIVE epoch, waiting for the + // fresh worker. (If worker A had won, this row would read `done` at epoch 1.) + let (status, epoch, file_path) = job_row(&pool, event_id, "zip").await.unwrap(); + assert_eq!((status.as_str(), epoch), ("pending", new_epoch)); + assert_eq!(file_path, None, "the loser's file_path must never be recorded"); + + // And nothing is downloadable — not the stale archive, not anything. + assert_eq!(downloadable(&pool, event_id, "zip").await, None); +} + +/// The documented, deliberate nuance in `claim_job`: after a bare `open_event` (which writes +/// NOTHING to `export_job` — that is the point of the design), a worker at the old epoch still WINS +/// its claim and can still write `done`. That is wasted work, not incorrectness: retirement is +/// enforced at READ time. `export_current` must refuse to expose the row. +/// +/// PREVENTS: someone "optimising" `claim_job` into a cross-table `EXISTS (SELECT ... FROM event)` +/// guard — the exact unsound guard migration 014 removed (under READ COMMITTED, a blocked UPDATE +/// re-evaluates same-row predicates but answers other-table subqueries from a stale snapshot). +/// This test pins the read-time enforcement so the write-time guard is never re-added. +#[sqlx::test] +async fn reopen_retires_at_read_time_not_write_time(pool: PgPool) { + let event_id = seed_event(&pool, "wedding").await; + let epoch = release_gallery(&pool, "wedding").await.unwrap(); + + assert!(claim_job(&pool, event_id, "zip", epoch).await); + + // Host reopens uploads. No export_job row is touched. + assert_eq!(open_event(&pool, "wedding").await, 1); + + // The in-flight worker's row-local writes still match — it was never told to stop. + assert!( + finalize_job(&pool, event_id, "zip", epoch, "exports/Gallery.1.zip").await, + "documented: the claim/finalize is guarded on the JOB row's epoch, not the event's" + ); + let (status, _, _) = job_row(&pool, event_id, "zip").await.unwrap(); + assert_eq!(status, "done", "the row really does say done"); + + // …and yet it is invisible. `export_current` requires the event to be released AND the epochs to + // match; the reopen broke both. A worker at a dead epoch writes a row nobody can see. + assert!(!in_export_current(&pool, event_id, "zip").await); + assert_eq!( + downloadable(&pool, event_id, "zip").await, + None, + "a reopened event must serve NO keepsake, however finished the job row looks" + ); +} + +// ───────────────────────────────────────────────────────────────────────────── +// 3. `export_current` exactness (table-driven) +// ───────────────────────────────────────────────────────────────────────────── + +/// The view is the ONE definition of "downloadable". Released + `done` + matching epoch ⇒ present; +/// break ANY single leg ⇒ absent. Nothing else may make it appear or disappear. +/// +/// PREVENTS, leg by leg: +/// * `released` — serving a keepsake for an event whose uploads are still open, i.e. an archive +/// missing every photo taken after the snapshot. +/// * `done` — handing out a half-written ZIP (a corrupt keepsake, downloaded once, kept forever). +/// * `epoch` — the retired-generation download: the 404-forever keepsake, or worse, the archive +/// still containing content that was taken down. +#[sqlx::test] +async fn export_current_is_exactly_the_invariant(pool: PgPool) { + // (name, released?, status, job epoch offset from the event epoch, expected visible) + let cases: &[(&str, bool, &str, i64, bool)] = &[ + ("released + done + current epoch", true, "done", 0, true), + ("NOT released (done, epoch matches)", false, "done", 0, false), + ("NOT done: pending", true, "pending", 0, false), + ("NOT done: running", true, "running", 0, false), + ("NOT done: failed", true, "failed", 0, false), + ("stale epoch (done, released)", true, "done", -1, false), + ("future epoch (done, released)", true, "done", 1, false), + ("migration-014 retired sentinel epoch -1", true, "done", -2, false), + ]; + + for (i, (name, released, status, offset, expect_visible)) in cases.iter().enumerate() { + let slug = format!("case{i}"); + let event_id = seed_event(&pool, &slug).await; + + // Get the event to a known epoch (1) either by releasing it, or — for the unreleased case — + // by releasing and reopening, which leaves it unreleased at a non-zero epoch. + let event_epoch_now = if *released { + release_gallery(&pool, &slug).await.unwrap() + } else { + release_gallery(&pool, &slug).await.unwrap(); + open_event(&pool, &slug).await; + event_epoch(&pool, event_id).await + }; + + // Plant a single ZIP job row in the exact state under test. `-2` encodes "the sentinel the + // migration stamps on retired rows", which must never equal a non-negative event epoch. + // (Clear the rows the release armed first — `UNIQUE (event_id, type)`.) + sqlx::query("DELETE FROM export_job WHERE event_id = $1") + .bind(event_id) + .execute(&pool) + .await + .expect("clear armed jobs"); + + let job_epoch = if *offset == -2 { -1 } else { event_epoch_now + offset }; + sqlx::query( + "INSERT INTO export_job (event_id, type, status, progress_pct, epoch, file_path) + VALUES ($1, 'zip', $2::export_status, 100, $3, 'exports/Gallery.zip')", + ) + .bind(event_id) + .bind(*status) + .bind(job_epoch) + .execute(&pool) + .await + .expect("plant job row"); + + let visible = downloadable(&pool, event_id, "zip").await.is_some(); + assert_eq!( + visible, *expect_visible, + "export_current exactness violated for case: {name} \ + (released={released}, status={status}, job_epoch={job_epoch}, event_epoch={event_epoch_now})" + ); + } +} + +// ───────────────────────────────────────────────────────────────────────────── +// 4. The ViewerOnly ZIP carry-forward +// ───────────────────────────────────────────────────────────────────────────── + +/// Branch A — the ZIP is `done` at the outgoing epoch: the carry-forward re-stamps it to the new +/// epoch (rows_affected = 1), so only the HTML viewer is rebuilt and the finished ZIP stays +/// downloadable throughout. +/// +/// PREVENTS: rebuilding a multi-GB archive because someone deleted a comment. The ZIP holds media, +/// not comments — a needless rebuild would 404 the photo download for minutes to change nothing +/// inside it. +#[sqlx::test] +async fn viewer_only_carries_a_done_zip_forward(pool: PgPool) { + let event_id = seed_event(&pool, "wedding").await; + let e1 = release_gallery(&pool, "wedding").await.unwrap(); + + // Both halves finish at epoch 1 — the keepsake is live. + for t in ["zip", "html"] { + assert!(claim_job(&pool, event_id, t, e1).await); + assert!(finalize_job(&pool, event_id, t, e1, &format!("exports/{t}.{e1}.zip")).await); + } + let zip_file = downloadable(&pool, event_id, "zip").await.expect("zip is live"); + + // ── A comment is moderated: invalidate_and_arm(Affects::ViewerOnly). ── + let (_, _, e2) = bump_epoch(&pool, "wedding").await.unwrap(); + let carried = carry_zip_forward(&pool, event_id, e2).await; + assert!(carried, "a `done` ZIP at epoch-1 MUST be carried forward (rows_affected == 1)"); + + // Only the viewer is re-armed… + let mut conn = pool.acquire().await.unwrap(); + enqueue_types_at_epoch(&mut conn, event_id, e2, &["html"]).await; + drop(conn); + + // …and the ZIP is STILL DOWNLOADABLE, at the new epoch, pointing at the same, unrenamed file. + let (status, epoch, _) = job_row(&pool, event_id, "zip").await.unwrap(); + assert_eq!((status.as_str(), epoch), ("done", e2), "the ZIP row rode the epoch bump"); + assert_eq!( + downloadable(&pool, event_id, "zip").await, + Some(zip_file), + "the carried archive must never stop being served — same file, new epoch" + ); + + // The viewer, meanwhile, is correctly retired and pending a rebuild. + assert_eq!(job_row(&pool, event_id, "html").await.unwrap().0, "pending"); + assert_eq!(downloadable(&pool, event_id, "html").await, None); +} + +/// Branch B — THE BUG WE JUST FIXED. If the ZIP is still `pending`/`running` when the comment is +/// moderated (which is MINUTES for a real multi-GB gallery, and deleting a comment right after +/// release is an utterly ordinary thing to do), the carry-forward matches NOTHING +/// (rows_affected = 0) — so the caller must NOT assume it carried, and must re-arm the ZIP too. +/// +/// PREVENTS: the stranded ZIP. Blindly re-arming only the viewer would leave the ZIP row at the +/// retired epoch; the in-flight worker then finishes and writes `done` at an epoch `export_current` +/// no longer matches, nothing ever re-arms it, and `GET /export/zip` 404s FOREVER — a keepsake the +/// couple paid for that simply never appears, short of a reboot. +#[sqlx::test] +async fn viewer_only_carry_forward_matches_nothing_when_zip_unfinished(pool: PgPool) { + for zip_state in ["pending", "running"] { + let slug = format!("wedding-{zip_state}"); + let event_id = seed_event(&pool, &slug).await; + let e1 = release_gallery(&pool, &slug).await.unwrap(); + + // The ZIP worker is still going; only the viewer has finished. + if zip_state == "running" { + assert!(claim_job(&pool, event_id, "zip", e1).await); + } + assert!(claim_job(&pool, event_id, "html", e1).await); + assert!(finalize_job(&pool, event_id, "html", e1, "exports/Memories.1.zip").await); + + // ── The comment is moderated. ── + let (_, _, e2) = bump_epoch(&pool, &slug).await.unwrap(); + let carried = carry_zip_forward(&pool, event_id, e2).await; + + assert!( + !carried, + "a {zip_state} ZIP has nothing to carry forward — the UPDATE must affect 0 rows \ + (its `status = 'done'` predicate is the whole precondition)" + ); + + // The carry-forward's OWN result decides. It didn't match ⇒ rebuild the ZIP as well. + let types: &[&str] = if carried { &["html"] } else { &["zip", "html"] }; + let mut conn = pool.acquire().await.unwrap(); + enqueue_types_at_epoch(&mut conn, event_id, e2, types).await; + drop(conn); + + // THE ASSERTION THAT WOULD HAVE CAUGHT THE BUG: the ZIP must not be stranded at the dead + // epoch. It is re-armed at the live one, so a fresh worker will actually build it. + let (status, epoch, _) = job_row(&pool, event_id, "zip").await.unwrap(); + assert_eq!( + (status.as_str(), epoch), + ("pending", e2), + "the unfinished ZIP MUST be re-armed at the new epoch, not left stranded at {e1}" + ); + + // Even if the old in-flight worker now "finishes", it is inert and cannot resurrect itself. + assert!(!finalize_job(&pool, event_id, "zip", e1, "exports/Gallery.1.zip").await); + assert_eq!(downloadable(&pool, event_id, "zip").await, None); + } +} + +/// The re-arm upsert must never clobber the archive it just carried forward. +/// +/// `enqueue_types_at_epoch`'s `WHERE export_job.status <> 'done' OR export_job.epoch <> EXCLUDED.epoch` +/// is the "startup recovery must not clobber a good half" rule, expressed as the readiness predicate +/// itself. PREVENTS: boot recovery resetting a perfectly good, downloadable ZIP back to `pending` +/// and making the keepsake 404 while it needlessly rebuilds. +#[sqlx::test] +async fn enqueue_preserves_a_done_half_at_the_current_epoch(pool: PgPool) { + let event_id = seed_event(&pool, "wedding").await; + let e1 = release_gallery(&pool, "wedding").await.unwrap(); + + // The ZIP finished; the HTML worker was killed mid-flight (crash) and sits at `running`. + assert!(claim_job(&pool, event_id, "zip", e1).await); + assert!(finalize_job(&pool, event_id, "zip", e1, "exports/Gallery.1.zip").await); + assert!(claim_job(&pool, event_id, "html", e1).await); + + // Boot recovery re-arms both types at the SAME epoch. + let mut conn = pool.acquire().await.unwrap(); + enqueue_types_at_epoch(&mut conn, event_id, e1, &["zip", "html"]).await; + drop(conn); + + // The good half survives untouched… + let (status, epoch, file_path) = job_row(&pool, event_id, "zip").await.unwrap(); + assert_eq!((status.as_str(), epoch), ("done", e1), "a done half at the live epoch is preserved"); + assert_eq!(file_path.as_deref(), Some("exports/Gallery.1.zip"), "file_path not nulled"); + assert!(downloadable(&pool, event_id, "zip").await.is_some()); + + // …and only the missing half is re-armed. + assert_eq!(job_row(&pool, event_id, "html").await.unwrap().0, "pending"); +} diff --git a/backend/tests/upload_concurrency.rs b/backend/tests/upload_concurrency.rs new file mode 100644 index 0000000..29cc182 --- /dev/null +++ b/backend/tests/upload_concurrency.rs @@ -0,0 +1,268 @@ +//! DB-backed integration tests for the two concurrency guards in the upload commit path +//! (`handlers/upload.rs`). Both are SQL — a `FOR SHARE` row lock and an atomic compare-and-increment +//! — and both are load-bearing for things a user can actually lose: a wedding photo, or the disk. +//! +//! `#[sqlx::test]` gives each test a fresh database with the real migrations applied. + +mod common; + +use std::sync::Arc; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::time::Duration; + +use chrono::{DateTime, Utc}; +use common::*; +use sqlx::PgPool; +use uuid::Uuid; + +/// SRC: `handlers/upload.rs:313-322` — the guarded quota increment, verbatim. +/// Returns `rows_affected()`; the handler aborts the whole upload tx when this is 0. +async fn quota_inc( + exec: impl sqlx::PgExecutor<'_>, + user_id: Uuid, + size: i64, + limit: i64, +) -> u64 { + sqlx::query( + "UPDATE \"user\" SET total_upload_bytes = total_upload_bytes + $2 + WHERE id = $1 AND total_upload_bytes + $2 <= $3", + ) + .bind(user_id) + .bind(size) + .bind(limit) + .execute(exec) + .await + .expect("quota_inc") + .rows_affected() +} + +async fn total_bytes(pool: &PgPool, user_id: Uuid) -> i64 { + sqlx::query_scalar("SELECT total_upload_bytes FROM \"user\" WHERE id = $1") + .bind(user_id) + .fetch_one(pool) + .await + .expect("total_bytes") +} + +// ───────────────────────────────────────────────────────────────────────────── +// 5. The atomic quota increment +// ───────────────────────────────────────────────────────────────────────────── + +/// Two attempts sized off ONE stale snapshot, each of which "fits" on its own, cannot both commit. +/// The predicate re-reads `total_upload_bytes` inside the UPDATE, so the second matches 0 rows. +/// +/// PREVENTS: one guest filling the disk. The handler's pre-flight quota check runs BEFORE the body is +/// streamed — minutes earlier, for a 500 MB video. If the commit trusted that snapshot, a guest could +/// start N uploads that each individually fit under the limit and land all N, blowing straight through +/// the quota and (in a 1 GB container) taking the event down for everyone. +#[sqlx::test] +async fn quota_two_attempts_from_one_stale_snapshot_cannot_both_commit(pool: PgPool) { + let event_id = seed_event(&pool, "wedding").await; + let user_id = seed_user(&pool, event_id, "Gierige Gudrun").await; + + const LIMIT: i64 = 100; + const SIZE: i64 = 60; + + // THE STALE SNAPSHOT: the pre-flight check both uploads were admitted on. + let snapshot = total_bytes(&pool, user_id).await; + assert_eq!(snapshot, 0); + // Each upload, judged against that snapshot alone, fits: 0 + 60 <= 100. Twice. + assert!(snapshot + SIZE <= LIMIT); + + assert_eq!(quota_inc(&pool, user_id, SIZE, LIMIT).await, 1, "the first upload commits"); + assert_eq!( + quota_inc(&pool, user_id, SIZE, LIMIT).await, + 0, + "the second MUST affect 0 rows — it was admitted on a snapshot that is now a lie \ + (60 + 60 = 120 > 100). rows_affected() == 0 is what makes the handler abort." + ); + + assert_eq!(total_bytes(&pool, user_id).await, SIZE, "never 120 — the quota held"); +} + +/// The same, but genuinely CONCURRENT: two transactions that both read `total = 0`, then both try to +/// commit 60 bytes against a 100-byte limit. The second UPDATE blocks on the first's row lock and — +/// because every predicate is on the ROW BEING UPDATED — Postgres re-evaluates it against the +/// post-commit row (EPQ) rather than the statement's original snapshot. It matches nothing. +/// +/// PREVENTS: exactly the same disk-filling overrun, on the path it actually happens — two uploads +/// in flight at once, which is the normal case at a party. +#[sqlx::test] +async fn quota_guard_is_atomic_under_concurrent_transactions(pool: PgPool) { + let event_id = seed_event(&pool, "wedding").await; + let user_id = seed_user(&pool, event_id, "Gierige Gudrun").await; + + const LIMIT: i64 = 100; + const SIZE: i64 = 60; + + let mut tx1 = pool.begin().await.unwrap(); + let mut tx2 = pool.begin().await.unwrap(); + + // Both transactions read the same snapshot and both would pass a naive `total + size <= limit` + // check done in Rust. + for tx in [&mut tx1, &mut tx2] { + let seen: i64 = sqlx::query_scalar("SELECT total_upload_bytes FROM \"user\" WHERE id = $1") + .bind(user_id) + .fetch_one(&mut **tx) + .await + .unwrap(); + assert_eq!(seen, 0, "both see an empty quota"); + } + + // tx1 takes the row lock and commits. + assert_eq!(quota_inc(&mut *tx1, user_id, SIZE, LIMIT).await, 1); + tx1.commit().await.unwrap(); + + // tx2's UPDATE was written against the stale snapshot but is evaluated against the row as it + // now stands. + assert_eq!( + quota_inc(&mut *tx2, user_id, SIZE, LIMIT).await, + 0, + "the loser MUST see 0 rows affected — this is the entire quota guarantee" + ); + tx2.rollback().await.unwrap(); + + assert_eq!(total_bytes(&pool, user_id).await, SIZE); + + // And an upload that legitimately fits in what's left still succeeds — the guard rejects + // overruns, not everything. + assert_eq!(quota_inc(&pool, user_id, 40, LIMIT).await, 1, "0 + 60 + 40 == 100, exactly at the limit"); + assert_eq!(total_bytes(&pool, user_id).await, LIMIT); + assert_eq!(quota_inc(&pool, user_id, 1, LIMIT).await, 0, "and one byte more is refused"); +} + +// ───────────────────────────────────────────────────────────────────────────── +// 6. The `FOR SHARE` upload lock vs. the release +// ───────────────────────────────────────────────────────────────────────────── + +/// SRC: `handlers/upload.rs:297-303` — the in-transaction re-check under a row lock, verbatim. +async fn lock_and_read_event( + tx: &mut sqlx::PgConnection, + event_id: Uuid, +) -> (Option>, Option>) { + sqlx::query_as("SELECT uploads_locked_at, export_released_at FROM event WHERE id = $1 FOR SHARE") + .bind(event_id) + .fetch_one(tx) + .await + .expect("FOR SHARE re-check") +} + +/// THE GUARD AGAINST SILENT, PERMANENT PHOTO LOSS. +/// +/// An upload holding `FOR SHARE` on the event row must BLOCK the `UPDATE event SET +/// export_released_at = NOW()` in `release_gallery` until it commits. Either the upload commits first +/// — and the release (hence the export snapshot) is strictly ordered after it, so the keepsake +/// CONTAINS the photo — or the release commits first and the upload observes the lock and rejects +/// (reversibly: the client keeps the blob and resumes after a reopen). +/// +/// PREVENTS: the lost wedding photo. Without this serialization: a guest starts a 500 MB video, the +/// pre-flight lock check passes, the host releases the gallery, the export workers snapshot the +/// uploads table, and THEN the upload commits. The photo appears in the live feed but is missing from +/// the downloaded keepsake, forever — nothing ever regenerates it and nobody ever notices. +#[sqlx::test] +async fn for_share_upload_lock_serializes_against_release(pool: PgPool) { + let event_id = seed_event(&pool, "wedding").await; + let user_id = seed_user(&pool, event_id, "Fotograf Fritz").await; + + // ── The guest's upload transaction takes the share lock. ── + let mut upload_tx = pool.begin().await.unwrap(); + let (locked, released) = lock_and_read_event(&mut upload_tx, event_id).await; + assert!(locked.is_none() && released.is_none(), "uploads are open, so we proceed to commit"); + + // ── Concurrently, the host hits "Galerie freigeben". ── + let release_done = Arc::new(AtomicBool::new(false)); + let release_task = { + let pool = pool.clone(); + let release_done = release_done.clone(); + tokio::spawn(async move { + sqlx::query( + "UPDATE event + SET export_released_at = NOW(), + uploads_locked_at = COALESCE(uploads_locked_at, NOW()), + export_epoch = export_epoch + 1 + WHERE id = $1 AND export_released_at IS NULL", + ) + .bind(event_id) + .execute(&pool) + .await + .expect("release"); + release_done.store(true, Ordering::SeqCst); + }) + }; + + // The release MUST be stuck behind our `FOR SHARE` row lock. (`FOR SHARE` conflicts with the + // `FOR UPDATE` lock the UPDATE needs, so Postgres makes it wait — this is not a timing race, + // it is a lock-conflict guarantee; the sleep only gives it every chance to wrongly proceed.) + tokio::time::sleep(Duration::from_millis(750)).await; + assert!( + !release_done.load(Ordering::SeqCst), + "the release MUST block while an upload holds FOR SHARE — if it can slip past, the export \ + snapshot is taken while a photo is still committing and that photo is lost forever" + ); + + // The photo commits. It is now unambiguously part of the upload set. + let upload_id: Uuid = sqlx::query_scalar( + "INSERT INTO upload (event_id, user_id, original_path, mime_type, original_size_bytes) + VALUES ($1, $2, 'originals/wedding/x.jpg', 'image/jpeg', 1234) RETURNING id", + ) + .bind(event_id) + .bind(user_id) + .fetch_one(&mut *upload_tx) + .await + .unwrap(); + upload_tx.commit().await.unwrap(); + + // Only now can the release proceed. + tokio::time::timeout(Duration::from_secs(5), release_task) + .await + .expect("the release must unblock once the upload commits") + .unwrap(); + + // THE PAYOFF: the export snapshot — the very query the ZIP worker runs — sees the photo. Order + // enforced by the lock: upload commit < release < snapshot. + let snapshot: Vec = sqlx::query_scalar( + "SELECT u.id FROM upload u + JOIN \"user\" usr ON usr.id = u.user_id + WHERE u.event_id = $1 AND u.deleted_at IS NULL + AND usr.uploads_hidden = FALSE AND usr.is_banned = FALSE", + ) + .bind(event_id) + .fetch_all(&pool) + .await + .unwrap(); + assert_eq!(snapshot, vec![upload_id], "the released keepsake CONTAINS the in-flight photo"); +} + +/// The other side of the same lock: once the release has COMMITTED, the next upload's `FOR SHARE` +/// re-read sees `export_released_at` set and the handler rejects it with `UploadsLocked`. +/// +/// PREVENTS: the same lost photo, on the losing side of the race — a photo committing AFTER the +/// export snapshot would be in the live feed but missing from the keepsake. Rejecting is the correct +/// outcome, and it is reversible: `UploadsLocked` (not Forbidden) tells the client to keep the blob +/// and resume when the host reopens. +#[sqlx::test] +async fn upload_after_release_commits_sees_the_lock_and_is_rejected(pool: PgPool) { + let event_id = seed_event(&pool, "wedding").await; + + // Before the release, the re-check passes. + let mut tx = pool.begin().await.unwrap(); + let (locked, released) = lock_and_read_event(&mut tx, event_id).await; + assert!(locked.is_none() && released.is_none()); + tx.rollback().await.unwrap(); + + assert_eq!(release_gallery(&pool, "wedding").await, Some(1)); + + // After it, the identical re-check sees the release and the handler bails out. + let mut tx = pool.begin().await.unwrap(); + let (locked, released) = lock_and_read_event(&mut tx, event_id).await; + assert!(released.is_some(), "the FOR SHARE re-read MUST observe the committed release"); + assert!(locked.is_some(), "release locks uploads in the same statement (release ⇒ lock)"); + tx.rollback().await.unwrap(); + + // And a reopen makes it uploadable again — the rejection was reversible, not terminal. + assert_eq!(open_event(&pool, "wedding").await, 1); + let mut tx = pool.begin().await.unwrap(); + let (locked, released) = lock_and_read_event(&mut tx, event_id).await; + assert!(locked.is_none() && released.is_none(), "the guest can resume their upload"); + tx.rollback().await.unwrap(); +}