diff --git a/backend/migrations/014_export_epoch.down.sql b/backend/migrations/014_export_epoch.down.sql new file mode 100644 index 0000000..e94705b --- /dev/null +++ b/backend/migrations/014_export_epoch.down.sql @@ -0,0 +1,28 @@ +-- Restore the stored ready flags and the per-job counter. +DROP VIEW IF EXISTS export_current; + +ALTER TABLE event ADD COLUMN export_zip_ready BOOLEAN NOT NULL DEFAULT FALSE; +ALTER TABLE event ADD COLUMN export_html_ready BOOLEAN NOT NULL DEFAULT FALSE; + +-- Re-derive the flags from what the epoch model considers downloadable, so the old code sees +-- exactly the state it would have written itself. +UPDATE event e + SET export_zip_ready = EXISTS ( + SELECT 1 FROM export_job j + WHERE j.event_id = e.id AND j.type = 'zip' + AND j.status = 'done' AND j.epoch = e.export_epoch + ), + export_html_ready = EXISTS ( + SELECT 1 FROM export_job j + WHERE j.event_id = e.id AND j.type = 'html' + AND j.status = 'done' AND j.epoch = e.export_epoch + ) + WHERE e.export_released_at IS NOT NULL; + +ALTER TABLE export_job RENAME COLUMN epoch TO release_seq; + +-- The old code treats release_seq as a non-negative per-job counter; the -1 retirement sentinel +-- has no meaning there, so floor it back to 0. +UPDATE export_job SET release_seq = 0 WHERE release_seq < 0; + +ALTER TABLE event DROP COLUMN export_epoch; diff --git a/backend/migrations/014_export_epoch.up.sql b/backend/migrations/014_export_epoch.up.sql new file mode 100644 index 0000000..6faac8d --- /dev/null +++ b/backend/migrations/014_export_epoch.up.sql @@ -0,0 +1,79 @@ +-- Export generation, unified. +-- +-- Before this migration, "which generation of the keepsake is current?" had no single answer. +-- It was assembled at runtime from three separately-written pieces of state across two tables: +-- * event.export_released_at — a release marker with NO identity (you cannot ask *which* release) +-- * export_job.release_seq — a per-row counter, in a different table, bumped in a different statement +-- * event.export_{zip,html}_ready — a CACHED COPY of the derivation of the other two +-- Keeping those three in agreement took ~10 hand-written guards, and every guard was a repair of the +-- same missing invariant. Three consecutive review rounds each found another path that slipped through. +-- +-- This replaces all of it with ONE authority: +-- +-- event.export_epoch — a monotonic counter bumped in the SAME UPDATE as any change to +-- export_released_at (release and reopen are its only writers). +-- export_job.epoch — a COPY of event.export_epoch taken when the job was enqueued. +-- NEVER independently incremented. +-- +-- The single invariant, which replaces the entire proof: +-- +-- 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 therefore DERIVED, never stored — so it cannot drift, and no worker can set it. +-- A worker holding a dead epoch is INERT BY CONSTRUCTION: anything it writes is invisible to every +-- reader, with no guard involved. Losing a race can no longer corrupt state; it can only waste work. +-- +-- Crucially, epoch equality is checked on the SAME ROW the worker updates (export_job), never via a +-- cross-table EXISTS. That matters: under READ COMMITTED, when an UPDATE blocks on a row lock and the +-- blocker commits, Postgres re-evaluates the WHERE against the *updated target row* but answers +-- subqueries on OTHER tables from the statement's ORIGINAL snapshot. The old cross-row guard +-- (`EXISTS (SELECT 1 FROM event WHERE ... export_released_at IS NOT NULL)`) was unsound for exactly +-- that reason — it could admit a claim against an already-reopened event while RETURNING the +-- post-bump seq. A same-row `epoch = $n` predicate is re-evaluated correctly by EPQ. + +ALTER TABLE event ADD COLUMN export_epoch BIGINT NOT NULL DEFAULT 0; + +-- A currently-released event's live generation becomes epoch 1. +UPDATE event SET export_epoch = 1 WHERE export_released_at IS NOT NULL; + +-- The per-job counter becomes a plain copy of the event epoch it was enqueued for. +ALTER TABLE export_job RENAME COLUMN release_seq TO epoch; + +-- Carry the OLD notion of "downloadable" across exactly: a job the old model considered ready +-- (released + done + its ready flag) adopts the event's live epoch. Everything else is retired to +-- -1, which can never equal a non-negative event.export_epoch, so it is inert forever. +UPDATE export_job j + SET epoch = CASE + WHEN e.export_released_at IS NOT NULL + AND j.status = 'done' + AND ((j.type = 'zip' AND e.export_zip_ready) + OR (j.type = 'html' AND e.export_html_ready)) + THEN e.export_epoch + ELSE -1 + END + FROM event e + WHERE e.id = j.event_id; + +-- The duplicated derivation. Gone — along with the whole class of bugs where a stale worker +-- resurrected it, or where it disagreed with the job row it was supposed to summarise. +ALTER TABLE event DROP COLUMN export_zip_ready; +ALTER TABLE event DROP COLUMN export_html_ready; + +-- The ONE definition of "the current generation". Every reader goes through this, so readiness and +-- the file path are resolved in a SINGLE read — closing the download-path TOCTOU where the ready +-- flag was checked in one query and file_path fetched in another (a reopen between the two served a +-- keepsake that should have 404'd). +CREATE VIEW export_current AS +SELECT j.event_id, + j.type, + j.status, + j.progress_pct, + j.file_path, + j.epoch + FROM export_job j + JOIN event e ON e.id = j.event_id + WHERE e.export_released_at IS NOT NULL -- implied by the epoch match; kept as an assertion + AND j.epoch = e.export_epoch; diff --git a/backend/src/handlers/admin.rs b/backend/src/handlers/admin.rs index d38c63e..fe17f34 100644 --- a/backend/src/handlers/admin.rs +++ b/backend/src/handlers/admin.rs @@ -288,32 +288,28 @@ pub async fn download_zip( authenticate_download_ticket(&state, &q.ticket).await?; enforce_export_rate(&state, &headers).await?; - let event = crate::models::event::Event::find_by_slug(&state.pool, &state.config.event_slug) - .await? - .ok_or_else(|| AppError::NotFound("Event nicht gefunden.".into()))?; - - if !event.export_zip_ready { - return Err(AppError::NotFound( - "Der ZIP-Export ist noch nicht verfügbar.".into(), - )); - } - - let path = resolve_export_file(&state, "zip").await?; + let path = resolve_export_file(&state, "zip", "Der ZIP-Export ist noch nicht verfügbar.").await?; serve_file(path, "Gallery.zip", "application/zip").await } -/// Resolve the on-disk path of the CURRENT export generation from `export_job.file_path` -/// rather than a fixed filename. Exports are written to per-generation paths -/// (`Gallery..zip`) so a superseded worker can't stomp a fresh keepsake (H1); the -/// download must therefore follow whatever the current 'done' row points at, not guess. +/// Resolve the on-disk path of the CURRENT export generation — readiness check and path lookup in +/// ONE read, through the `export_current` view (migration 014). +/// +/// This used to be two statements: the caller checked `event.export_zip_ready`, then this fetched +/// `export_job.file_path`. A reopen landing between them served a keepsake that should have 404'd — +/// the same stale-keepsake class, leaking through the read path, and it existed only because +/// readiness was a stored copy rather than a derivation. `export_current` gives both facts from one +/// consistent snapshot: it yields a row only when the event is released AND the job is `done` at the +/// event's current epoch, so "is it ready" and "which file" can no longer disagree. async fn resolve_export_file( state: &AppState, export_type: &str, + not_ready_msg: &str, ) -> Result { let file_path: Option<(Option,)> = sqlx::query_as( - "SELECT file_path FROM export_job ej - JOIN event e ON e.id = ej.event_id - WHERE e.slug = $1 AND ej.type = $2::export_type AND ej.status = 'done'", + "SELECT c.file_path FROM export_current c + JOIN event e ON e.id = c.event_id + WHERE e.slug = $1 AND c.type = $2::export_type AND c.status = 'done'", ) .bind(&state.config.event_slug) .bind(export_type) @@ -321,11 +317,12 @@ async fn resolve_export_file( .await .map_err(|e| AppError::Internal(e.into()))?; + let Some((Some(rel),)) = file_path else { + return Err(AppError::NotFound(not_ready_msg.into())); + }; + // `file_path` is stored as `exports/`; the base dir is already `export_path`, so // join only the file name (defends against any absolute/`..` content too). - let rel = file_path - .and_then(|(p,)| p) - .ok_or_else(|| AppError::NotFound("Exportdatei nicht gefunden.".into()))?; let name = std::path::Path::new(&rel) .file_name() .ok_or_else(|| AppError::NotFound("Exportdatei nicht gefunden.".into()))?; @@ -344,17 +341,8 @@ pub async fn download_html( authenticate_download_ticket(&state, &q.ticket).await?; enforce_export_rate(&state, &headers).await?; - let event = crate::models::event::Event::find_by_slug(&state.pool, &state.config.event_slug) - .await? - .ok_or_else(|| AppError::NotFound("Event nicht gefunden.".into()))?; - - if !event.export_html_ready { - return Err(AppError::NotFound( - "Der HTML-Export ist noch nicht verfügbar.".into(), - )); - } - - let path = resolve_export_file(&state, "html").await?; + let path = + resolve_export_file(&state, "html", "Der HTML-Export ist noch nicht verfügbar.").await?; serve_file(path, "Memories.zip", "application/zip").await } @@ -400,10 +388,16 @@ pub async fn export_status( let released = event.export_released_at.is_some(); + // Only report jobs at the CURRENT epoch. A row left behind by a retired generation (a worker + // that was superseded mid-run) is meaningless — surfacing its frozen `running`/77% would show + // the host a progress bar that never moves for a keepsake nobody is building. A retired row + // reads as "locked" (no current job), which is exactly what it is. let jobs: Vec<(String, String, i16)> = sqlx::query_as( - "SELECT type::text, status::text, progress_pct FROM export_job WHERE event_id = $1", + "SELECT type::text, status::text, progress_pct FROM export_job + WHERE event_id = $1 AND epoch = $2", ) .bind(event.id) + .bind(event.export_epoch) .fetch_all(&state.pool) .await?; diff --git a/backend/src/handlers/host.rs b/backend/src/handlers/host.rs index 93ed3e7..0f22d2b 100644 --- a/backend/src/handlers/host.rs +++ b/backend/src/handlers/host.rs @@ -435,6 +435,48 @@ pub async fn dismiss_pin_reset_request( Ok(StatusCode::NO_CONTENT) } +/// Content changed AFTER the gallery was released — regenerate the keepsake. +/// +/// Deleting a photo used to remove it from the live feed but leave it in the already-generated +/// archive FOREVER: the old `if ready { continue }` skip guaranteed the export was never rebuilt. +/// For a takedown ("please remove my photo") that is the one place it most needs to disappear. +/// +/// Bumping the epoch (while staying released) retires the current generation, so the stale archive +/// stops being downloadable the instant the delete commits, and a fresh worker rebuilds it without +/// the removed content. The download 404s in the meantime, which is the correct answer — serving +/// the old archive would serve the deleted photo. +async fn regenerate_keepsake_if_released(state: &AppState) -> Result<(), AppError> { + let mut tx = state.pool.begin().await?; + + let bumped: 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(&state.config.event_slug) + .fetch_optional(&mut *tx) + .await?; + + // Not released → nothing to regenerate; the export will be built fresh at release time. + let Some((event_id, event_name, epoch)) = bumped else { + return Ok(()); + }; + + crate::services::export::enqueue_jobs_at_epoch(&mut tx, event_id, epoch).await?; + tx.commit().await?; + + crate::services::export::spawn_export_jobs( + event_id, + event_name, + epoch, + state.pool.clone(), + state.config.media_path.clone(), + state.config.export_path.clone(), + state.sse_tx.clone(), + ); + Ok(()) +} + pub async fn host_delete_upload( State(state): State, RequireHost(auth): RequireHost, @@ -454,6 +496,9 @@ pub async fn host_delete_upload( serde_json::json!({ "upload_id": upload.id }).to_string(), )); + // A released keepsake must not keep serving a photo the host just took down. + regenerate_keepsake_if_released(&state).await?; + tracing::info!( actor_user_id = %auth.user_id, event_id = %auth.event_id, @@ -478,6 +523,8 @@ pub async fn host_delete_comment( "comment-deleted", serde_json::json!({ "comment_id": comment_id }).to_string(), )); + // The HTML viewer embeds comments — rebuild it so a deleted comment doesn't live on there. + regenerate_keepsake_if_released(&state).await?; tracing::info!( actor_user_id = %auth.user_id, event_id = %auth.event_id, @@ -511,57 +558,27 @@ pub async fn open_event( State(state): State, RequireHost(_auth): RequireHost, ) -> Result { - // Reopening also invalidates any prior release: the keepsake was snapshotted at - // release time, so allowing new uploads afterwards would silently diverge the live - // feed from the frozen export. Clearing `export_released_at` (and the readiness - // flags) lets the host re-release later to regenerate a correct, complete keepsake. + // Reopening invalidates any prior release: the keepsake was snapshotted at release time, so + // allowing new uploads afterwards would silently diverge the live feed from the frozen export. // - // BOTH statements run in ONE transaction. They must be atomic: the first makes - // `release_gallery` possible again (it gates on `export_released_at IS NULL`), so if a - // concurrent re-release could slip between them it would enqueue + spawn a FRESH worker at - // seq N+1, and our second statement would then bump that brand-new generation to N+2 — - // orphaning the live worker (its seq-guarded finalize/fail/flip all match nothing), stranding - // the row at `running` with no owner, and leaving the host unable to retry ("bereits - // freigegeben"). Inside a transaction, our row lock on `event` serializes `release_gallery`'s - // claim behind our commit, so it only ever sees the fully-reopened, fully-bumped state. - let mut tx = state.pool.begin().await?; - + // ONE statement retires the entire export generation. Bumping `export_epoch` in the same write + // that clears `export_released_at` instantly invalidates every in-flight worker, every `done` + // row and all readiness — because readiness is DERIVED from this epoch (migration 014), not + // stored. There is no export_job row to touch and nothing to keep in sync, so this needs no + // transaction. Any worker still streaming holds the old epoch and is now inert: its + // epoch-guarded finalize matches nothing, and it discards its own output. let result = sqlx::query( "UPDATE event - SET uploads_locked_at = NULL, + SET uploads_locked_at = NULL, export_released_at = NULL, - export_zip_ready = FALSE, - export_html_ready = FALSE + export_epoch = export_epoch + 1 WHERE slug = $1 AND (uploads_locked_at IS NOT NULL OR export_released_at IS NOT NULL)", ) .bind(&state.config.event_slug) - .execute(&mut *tx) + .execute(&state.pool) .await?; - let reopened = result.rows_affected() > 0; - - if reopened { - // A reopen invalidates any released keepsake, so make it a *supersession point* for the - // export pipeline: bump every export_job generation for this event. An in-flight worker - // — or one that has already `finalize_job`'d but not yet flipped its ready flag — captured - // the pre-reopen `release_seq`, so its guarded finalize/ready-flip now match nothing and - // become no-ops. Without this, the `export_released_at IS NOT NULL` flip guard alone leaves - // a sub-window open: a re-release re-arms `export_released_at` *before* it bumps the seq, and - // a stale worker's flip landing in that gap satisfies both guards and resurrects a - // pre-reopen keepsake (silent data loss). Bumping here retires the old seq for good. - sqlx::query( - "UPDATE export_job SET release_seq = release_seq + 1 - FROM event e - WHERE e.id = export_job.event_id AND e.slug = $1", - ) - .bind(&state.config.event_slug) - .execute(&mut *tx) - .await?; - } - - tx.commit().await?; - - if reopened { + if result.rows_affected() > 0 { let _ = state.sse_tx.send(SseEvent::new("event-opened", "{}")); } @@ -572,24 +589,37 @@ pub async fn release_gallery( State(state): State, RequireHost(_auth): RequireHost, ) -> Result { - // Atomic claim: the conditional UPDATE is the sole gate, so two concurrent - // release calls can't both pass a check-then-set and double-enqueue exports. - // rows_affected == 0 means someone already released. + // The release claim, the epoch bump, the upload lock and BOTH job rows are written in ONE + // transaction. Two reasons, both of which were live bugs: // - // Releasing also locks uploads in the same statement (release ⇒ lock). Otherwise a - // guest whose offline upload reconnects *after* the export snapshot would land in the - // live feed but not in the downloaded keepsake — a silent, non-regenerable data loss. - // `COALESCE` preserves an earlier explicit lock time rather than overwriting it. - let result = sqlx::query( + // 1. Cancellation. This handler used to commit `export_released_at` and only THEN await the + // enqueue. Axum drops the handler future when the client disconnects (closed tab, proxy + // timeout), which left the event released and uploads locked with ZERO export_job rows and + // no workers: downloads 404 forever and the host cannot retry, because release_gallery + // rejects an already-released event ("bereits freigegeben"). Only a restart escaped it. + // Now nothing is committed until every row is written, and the workers are spawned AFTER + // the commit (a detached `tokio::spawn` survives cancellation). + // 2. Atomicity vs. a concurrent reopen. Bumping the epoch in the same statement that sets + // `export_released_at` means no worker and no reader can ever observe "released again but + // the generation hasn't moved on yet" — the window every previous fix kept leaving open. + // + // Release also locks uploads in the same statement (release ⇒ lock), so the export snapshot is + // taken against a frozen upload set. `COALESCE` preserves an earlier explicit lock time. + let mut tx = state.pool.begin().await?; + + let claimed: Option<(Uuid, String, i64)> = sqlx::query_as( "UPDATE event SET export_released_at = NOW(), - uploads_locked_at = COALESCE(uploads_locked_at, NOW()) - WHERE slug = $1 AND export_released_at IS NULL", + 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(&state.config.event_slug) - .execute(&state.pool) + .fetch_optional(&mut *tx) .await?; - if result.rows_affected() == 0 { + + let Some((event_id, event_name, epoch)) = claimed else { // Distinguish "no such event" from "already released" for a clean error. let exists = Event::find_by_slug(&state.pool, &state.config.event_slug) .await? @@ -599,28 +629,29 @@ pub async fn release_gallery( } else { AppError::NotFound("Event nicht gefunden.".into()) }); - } + }; - // Release locked uploads too — tell any open composer to flip to the locked UI live - // rather than discovering it via a rejected upload. + // Arm both types at THIS epoch. No "skip the type that's already ready" check any more: the + // epoch bump above retired every prior generation, so there is nothing to preserve and nothing + // to be fooled by. (That skip was how a stale ready flag used to suppress regeneration.) + crate::services::export::enqueue_jobs_at_epoch(&mut tx, event_id, epoch).await?; + + tx.commit().await?; + + // Release locks uploads too — tell any open composer to flip to the locked UI live rather than + // discovering it via a rejected upload. let _ = state.sse_tx.send(SseEvent::new("event-closed", "{}")); - // We won the claim — load the event for its id/name to enqueue export jobs. - let event = Event::find_by_slug(&state.pool, &state.config.event_slug) - .await? - .ok_or_else(|| AppError::NotFound("Event nicht gefunden.".into()))?; - - // Enqueue + spawn via the shared path so a re-release (after reopen) regenerates - // cleanly and startup recovery uses identical logic. - crate::services::export::enqueue_and_spawn_exports( - event.id, - event.name, + // Detached — survives this handler being cancelled. + crate::services::export::spawn_export_jobs( + event_id, + event_name, + epoch, state.pool.clone(), state.config.media_path.clone(), state.config.export_path.clone(), state.sse_tx.clone(), - ) - .await?; + ); Ok(StatusCode::NO_CONTENT) } diff --git a/backend/src/handlers/upload.rs b/backend/src/handlers/upload.rs index 34081d7..2b93c9f 100644 --- a/backend/src/handlers/upload.rs +++ b/backend/src/handlers/upload.rs @@ -3,6 +3,7 @@ use std::time::Duration; use axum::extract::{Multipart, Path, State}; use axum::http::StatusCode; use axum::Json; +use chrono::{DateTime, Utc}; use serde::Deserialize; use uuid::Uuid; @@ -274,6 +275,36 @@ pub async fn upload( // bytes with no row to reclaim them (silent quota erosion / spurious lockout). let tx_result: Result = async { let mut tx = state.pool.begin().await?; + + // RE-CHECK THE LOCK, UNDER A ROW LOCK, INSIDE THE COMMIT TX. + // + // The pre-flight check at the top of this handler ran BEFORE we streamed the body — which + // for a 500 MB video is minutes. Trusting it here is a TOCTOU that silently loses photos + // from the keepsake, and it is the real cause of the "stale keepsake" bug that survived + // three rounds of fixes inside the export state machine: + // + // 1. guest starts a big upload; the lock check passes (event open) + // 2. host releases the gallery → uploads lock, export workers snapshot the uploads table + // 3. this upload commits AFTER that snapshot → it shows up in the live feed but is + // MISSING from the downloaded keepsake, permanently (nothing ever regenerates it) + // + // `FOR SHARE` conflicts with the `UPDATE event` in `release_gallery`, which serializes us + // against it. Either we take the lock first — and release (hence the export snapshot) is + // strictly ordered after our commit, so the snapshot CONTAINS this upload — or release + // commits first and we observe the lock here and reject. Either way the keepsake is + // complete. `UploadsLocked` (not Forbidden) is reversible: the client keeps the blob and + // resumes it when the host reopens. + let (locked_at, released_at): (Option>, Option>) = + sqlx::query_as( + "SELECT uploads_locked_at, export_released_at FROM event WHERE id = $1 FOR SHARE", + ) + .bind(auth.event_id) + .fetch_one(&mut *tx) + .await?; + if locked_at.is_some() || released_at.is_some() { + return Err(AppError::UploadsLocked("Uploads sind gesperrt.".into())); + } + // Increment the user's byte total. When a quota is in force, guard it atomically // (`total + size <= limit`) so two concurrent uploads can't both slip past the // stale pre-check — the loser's UPDATE matches 0 rows and we abort with the same diff --git a/backend/src/models/event.rs b/backend/src/models/event.rs index f739769..b4f891e 100644 --- a/backend/src/models/event.rs +++ b/backend/src/models/event.rs @@ -11,8 +11,11 @@ pub struct Event { pub is_active: bool, pub uploads_locked_at: Option>, pub export_released_at: Option>, - pub export_zip_ready: bool, - pub export_html_ready: bool, + /// Monotonic generation counter for the keepsake. Bumped in the SAME UPDATE as any change to + /// `export_released_at` (release and reopen are its only writers). An export is downloadable + /// iff a `done` `export_job` row carries this exact epoch — readiness is derived from that, + /// never stored, so it cannot drift and no worker can resurrect it. See migration 014. + pub export_epoch: i64, pub created_at: DateTime, } diff --git a/backend/src/services/export.rs b/backend/src/services/export.rs index a49d6c4..7bc0c6d 100644 --- a/backend/src/services/export.rs +++ b/backend/src/services/export.rs @@ -9,6 +9,7 @@ use include_dir::{include_dir, Dir}; use serde::Serialize; use sqlx::PgPool; use tokio::sync::broadcast; +use tokio::io::AsyncWriteExt; use tokio_util::compat::TokioAsyncReadCompatExt; use uuid::Uuid; @@ -82,79 +83,59 @@ struct ViewerMedia { // ── Entry point ────────────────────────────────────────────────────────────── -/// (Re)enqueue the export jobs for an event and spawn the workers. Safe to call on a -/// fresh release *and* on a re-release: the `export_job` rows are reset to a clean -/// `pending` state (clearing any prior `failed`/`done` from an earlier run) and the -/// event's `export_*_ready` flags are cleared so downloads reflect the regenerating -/// export rather than the stale one. This is the single path used by `release_gallery` -/// and by startup export recovery, so the two can't drift. -pub async fn enqueue_and_spawn_exports( +/// Arm both export jobs at `epoch`, unless a type is ALREADY complete at that same epoch. +/// +/// Takes an executor rather than a pool so `release_gallery` can run it inside the same +/// transaction that bumps the epoch — the release must be all-or-nothing. +/// +/// The `WHERE` on the upsert is the old `if ready { continue }` skip, expressed correctly. Its +/// purpose was always "startup recovery must not clobber a good half", and that intent is fine — +/// the bug was that it keyed off a SEPARATELY STORED ready flag that a stale worker could set. +/// Here the condition IS the readiness predicate itself (`done` at the current epoch), so it cannot +/// disagree with reality. On a release it is always true (the epoch just moved, so no row can be +/// done at the new epoch) and both types regenerate; on recovery it preserves a genuinely finished +/// half and re-arms only what is missing. +pub async fn enqueue_jobs_at_epoch( + conn: &mut sqlx::PgConnection, event_id: Uuid, - event_name: String, - pool: PgPool, - media_path: PathBuf, - export_path: PathBuf, - sse_tx: broadcast::Sender, + epoch: i64, ) -> Result<()> { - // Only (re)generate a type that isn't ALREADY complete. A (re)release always arrives - // with both ready flags false (a fresh release starts false; `open_event` clears them - // before a re-release), so both regenerate as intended. But startup recovery of a - // half-finished export (e.g. ZIP done+ready, HTML crashed) must leave the good half - // alone — resetting it would 404 an already-downloadable keepsake until it needlessly - // regenerates. Skip the ready ones; their still-`done` rows keep their `file_path`, and - // the worker we spawn for them below simply bails its claim (no `pending` row to grab). - let (zip_ready, html_ready): (bool, bool) = sqlx::query_as( - "SELECT export_zip_ready, export_html_ready FROM event WHERE id = $1", - ) - .bind(event_id) - .fetch_one(&pool) - .await?; - - for (export_type, ready) in [("zip", zip_ready), ("html", html_ready)] { - if ready { - continue; - } - // Reset the row to a clean 'pending' and BUMP `release_seq` — unconditionally, even - // if a worker from a prior release is still 'running'. That worker captured the old - // seq at claim time; bumping it here supersedes its generation, so when it finishes - // its guarded finalize (`WHERE release_seq = `) matches nothing and it - // discards its now-stale output instead of overwriting a fresh keepsake. The worker - // we spawn below then claims this new generation and regenerates from a current - // snapshot. Per-seq temp/final paths (see `run_zip_export`/`run_html_export`) keep - // the old and new workers from racing the same file on disk. + for export_type in ["zip", "html"] { sqlx::query( - "INSERT INTO export_job (event_id, type, status, progress_pct, release_seq) - VALUES ($1, $2::export_type, 'pending', 0, 1) + "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, - release_seq = export_job.release_seq + 1", + epoch = EXCLUDED.epoch + WHERE export_job.status <> 'done' OR export_job.epoch <> EXCLUDED.epoch", ) .bind(event_id) .bind(export_type) - .execute(&pool) + .bind(epoch) + .execute(&mut *conn) .await?; } - - spawn_export_jobs(event_id, event_name, pool, media_path, export_path, sse_tx); Ok(()) } -/// Startup export recovery: re-spawn exports for any released event whose keepsake is -/// not fully ready (a crash mid-export left `export_released_at` set but the ZIP/HTML -/// jobs `failed`). Without this, `release_gallery` would reject a retry with "bereits -/// freigegeben" and downloads would 404 forever. Runs once at boot, after `AppState` -/// exists (it needs the media/export paths + SSE sender). +/// Startup export recovery: re-arm exports for any released event whose keepsake isn't fully +/// present at the current epoch (a crash mid-export, or a `done` row whose file has since gone +/// missing). Without this, `release_gallery` would reject a retry with "bereits freigegeben" and +/// downloads would 404 forever. Runs once at boot, after `AppState` exists. +/// +/// Unlike the old version, this VERIFIES THE FILE IS ACTUALLY ON DISK. Recovery used to be purely +/// flag-driven, so a `done` row whose archive had been lost (a wiped/restored volume, or a +/// truncated write) was a permanent dead end: the download 404'd, recovery skipped the event +/// because it looked complete, and the only escape was manual DB surgery. pub async fn recover_exports( pool: PgPool, media_path: PathBuf, export_path: PathBuf, sse_tx: broadcast::Sender, ) { - let rows = match sqlx::query_as::<_, (Uuid, String)>( - "SELECT id, name FROM event - WHERE export_released_at IS NOT NULL - AND NOT (export_zip_ready AND export_html_ready)", + let rows = match sqlx::query_as::<_, (Uuid, String, i64)>( + "SELECT id, name, export_epoch FROM event WHERE export_released_at IS NOT NULL", ) .fetch_all(&pool) .await @@ -165,26 +146,101 @@ pub async fn recover_exports( return; } }; - for (event_id, event_name) in rows { - tracing::warn!("export recovery: re-spawning export jobs for event {event_id}"); - if let Err(e) = enqueue_and_spawn_exports( + + for (event_id, event_name, epoch) in rows { + // Retire any `done` row at the current epoch whose file is missing or empty, so the + // upsert below re-arms it instead of leaving an undownloadable "ready" keepsake. + if let Err(e) = invalidate_missing_files(&pool, &export_path, event_id, epoch).await { + tracing::error!("export recovery: file verification failed for {event_id}: {e:#}"); + } + + let complete: bool = sqlx::query_scalar( + "SELECT COUNT(*) = 2 FROM export_job + WHERE event_id = $1 AND epoch = $2 AND status = 'done'", + ) + .bind(event_id) + .bind(epoch) + .fetch_one(&pool) + .await + .unwrap_or(false); + if complete { + continue; + } + + tracing::warn!("export recovery: re-arming export jobs for event {event_id} @ epoch {epoch}"); + let mut conn = match pool.acquire().await { + Ok(c) => c, + Err(e) => { + tracing::error!("export recovery: cannot acquire connection for {event_id}: {e:#}"); + continue; + } + }; + if let Err(e) = enqueue_jobs_at_epoch(&mut conn, event_id, epoch).await { + tracing::error!("export recovery: failed to re-arm for event {event_id}: {e:#}"); + continue; + } + spawn_export_jobs( event_id, event_name, + epoch, pool.clone(), media_path.clone(), export_path.clone(), sse_tx.clone(), - ) - .await - { - tracing::error!("export recovery: failed to re-spawn for event {event_id}: {e:#}"); + ); + } +} + +/// Reset any `done` job at the current epoch whose archive is absent or zero-length. Guards +/// against DB/disk divergence (lost volume, truncated write) that recovery would otherwise +/// mistake for a finished keepsake. +async fn invalidate_missing_files( + pool: &PgPool, + export_path: &Path, + event_id: Uuid, + epoch: i64, +) -> Result<()> { + let done: Vec<(String, Option)> = sqlx::query_as( + "SELECT type::text, file_path FROM export_job + WHERE event_id = $1 AND epoch = $2 AND status = 'done'", + ) + .bind(event_id) + .bind(epoch) + .fetch_all(pool) + .await?; + + for (export_type, file_path) in done { + let intact = match file_path.as_deref().and_then(|p| Path::new(p).file_name()) { + Some(name) => tokio::fs::metadata(export_path.join(name)) + .await + .map(|m| m.is_file() && m.len() > 0) + .unwrap_or(false), + None => false, + }; + if !intact { + tracing::warn!( + "export recovery: {export_type} export for event {event_id} is marked done but its \ + file is missing/empty — re-arming it" + ); + sqlx::query( + "UPDATE export_job SET status = 'failed', file_path = NULL, + error_message = 'Exportdatei fehlt — wird neu erzeugt' + WHERE event_id = $1 AND type = $2::export_type AND epoch = $3", + ) + .bind(event_id) + .bind(&export_type) + .bind(epoch) + .execute(pool) + .await?; } } + Ok(()) } pub fn spawn_export_jobs( event_id: Uuid, event_name: String, + epoch: i64, pool: PgPool, media_path: PathBuf, export_path: PathBuf, @@ -197,19 +253,23 @@ pub fn spawn_export_jobs( let event_name2 = event_name.clone(); tokio::spawn(async move { - // Failure marking happens INSIDE the worker (guarded by the claimed `release_seq`), - // so a superseded worker's error can't clobber the fresh generation's row. - if let Err(e) = run_zip_export(event_id, &pool, &media_path, &export_path, &sse_tx).await { - tracing::error!("ZIP export failed for event {event_id}: {e:#}"); + // The worker is BORN with its epoch — it never learns it from the DB. A worker whose epoch + // has been retired is inert by construction: every write it makes is `epoch`-guarded on the + // row it is updating, so it simply matches nothing. No cross-table guard, no race. + if let Err(e) = run_zip_export(event_id, epoch, &pool, &media_path, &export_path, &sse_tx).await { + tracing::error!("ZIP export failed for event {event_id} @ epoch {epoch}: {e:#}"); + mark_failed(&pool, event_id, "zip", epoch, &e.to_string()).await; } maybe_broadcast_complete(&pool, event_id, &sse_tx).await; }); tokio::spawn(async move { if let Err(e) = - run_html_export(event_id, &event_name2, &pool2, &media_path2, &export_path2, &sse_tx2).await + run_html_export(event_id, epoch, &event_name2, &pool2, &media_path2, &export_path2, &sse_tx2) + .await { - tracing::error!("HTML export failed for event {event_id}: {e:#}"); + tracing::error!("HTML export failed for event {event_id} @ epoch {epoch}: {e:#}"); + mark_failed(&pool2, event_id, "html", epoch, &e.to_string()).await; } maybe_broadcast_complete(&pool2, event_id, &sse_tx2).await; }); @@ -219,28 +279,39 @@ pub fn spawn_export_jobs( async fn run_zip_export( event_id: Uuid, + epoch: i64, pool: &PgPool, media_path: &Path, export_path: &Path, sse_tx: &broadcast::Sender, ) -> Result<()> { - let seq = match claim_job(pool, event_id, "zip").await { - Some(seq) => seq, - // Another worker already owns this ZIP export — bail rather than race the temp file. - None => return Ok(()), - }; + if !claim_job(pool, event_id, "zip", epoch).await? { + // Either another worker owns this generation, or our epoch has been retired by a + // reopen/re-release. Both mean: not ours. Bail without touching anything. + return Ok(()); + } - // Run the body under the captured `seq`; on error mark THIS generation failed (a no-op - // if a re-release already superseded us). - let res = run_zip_export_inner(seq, event_id, pool, media_path, export_path, sse_tx).await; - if let Err(e) = &res { - mark_failed(pool, event_id, "zip", seq, &e.to_string()).await; + // On error, mark THIS generation failed — a no-op if we've since been superseded (the + // caller in `spawn_export_jobs` does it, epoch-guarded). Temp artifacts are cleaned up + // here so a failing export can't leak them (which is what fills the disk in the first place). + let res = run_zip_export_inner(epoch, event_id, pool, media_path, export_path, sse_tx).await; + if res.is_err() { + let _ = tokio::fs::remove_file(export_path.join(gen_name(event_id, "Gallery", epoch, ".tmp"))) + .await; } res } +/// Per-generation, per-EVENT artifact name. The event id matters: all events share one exports +/// volume, and a name keyed only by generation would let two events collide on the same path (and +/// let one event's prune delete another's live keepsake). The viewer temp dir was already +/// event-scoped; the archives were not. +fn gen_name(event_id: Uuid, prefix: &str, epoch: i64, suffix: &str) -> String { + format!("{prefix}.{event_id}.{epoch}{suffix}") +} + async fn run_zip_export_inner( - seq: i64, + epoch: i64, event_id: Uuid, pool: &PgPool, media_path: &Path, @@ -254,12 +325,10 @@ async fn run_zip_export_inner( let exports_dir = export_path.to_path_buf(); tokio::fs::create_dir_all(&exports_dir).await?; - // Per-generation paths: a superseded worker (older `seq`) and the fresh worker never - // share a file on disk, so neither can truncate or interleave the other's bytes. The - // final name embeds the seq too; the download handler serves whatever `file_path` the - // current 'done' row points at, so an orphaned older generation is never served. - let tmp_path = exports_dir.join(format!("Gallery.zip.{seq}.tmp")); - let out_name = format!("Gallery.{seq}.zip"); + // Per-generation paths: a superseded worker (older epoch) and the fresh worker never share a + // file on disk, so neither can truncate or interleave the other's bytes. + let tmp_path = exports_dir.join(gen_name(event_id, "Gallery", epoch, ".tmp")); + let out_name = gen_name(event_id, "Gallery", epoch, ".zip"); let out_path = exports_dir.join(&out_name); { @@ -268,9 +337,6 @@ async fn run_zip_export_inner( for (i, row) in uploads.iter().enumerate() { let src = media_path.join(&row.original_path); - if !src.exists() { - continue; - } let ext = ext_from_path(&row.original_path); let date = row.created_at.format("%Y-%m-%d_%H-%M").to_string(); let name_safe = sanitize_name(&row.uploader_name); @@ -278,66 +344,64 @@ async fn run_zip_export_inner( let entry_name = format!("{folder}/{date}_{name_safe}_{}.{ext}", row.id); let builder = ZipEntryBuilder::new(entry_name.into(), Compression::Stored); - let mut entry = zip.write_entry_stream(builder).await?; - let mut f = tokio::fs::File::open(&src).await?.compat(); + // Open BEFORE writing the entry header. A missing source is skipped (the media file + // was deleted, or its processing failed) — but it must be skipped without aborting the + // whole export. The old code did `if !src.exists() { continue }` and then `open(..)?`, + // a TOCTOU: a file vanishing in between turned a tolerated gap into a hard error that + // failed the entire keepsake, permanently (the event is already released, so the host + // cannot retry). Opening first collapses the check and the use into one operation. + let src_file = match tokio::fs::File::open(&src).await { + Ok(f) => f, + Err(e) => { + tracing::warn!( + "ZIP export: skipping upload {} — cannot read {}: {e}", + row.id, + src.display() + ); + continue; + } + }; + + let mut entry = zip.write_entry_stream(builder).await?; + let mut f = src_file.compat(); fcopy(&mut f, &mut entry).await?; entry.close().await?; let pct = ((i + 1) as f32 / total * 100.0) as i16; - update_progress(pool, event_id, "zip", seq, pct.min(99)).await; + update_progress(pool, event_id, "zip", epoch, pct.min(99)).await; } - zip.close().await?; + // FLUSH AND FSYNC BEFORE THE RENAME. + // + // `async_zip`'s `close()` writes the central directory and then hands back the inner + // writer — it never flushes it. And `tokio::fs::File` is write-behind: dropping it awaits + // nothing and SILENTLY DISCARDS any error from the last write. So on a full disk the final + // chunk (the one carrying the end-of-central-directory record) could fail, the error would + // vanish, and we would rename a truncated archive into place, mark it done, and then prune + // the last good generation. `sync_all` both surfaces that error and makes the bytes durable + // before the DB is told the archive exists. + let mut file = zip.close().await?.into_inner(); + file.flush().await?; + file.sync_all().await?; } tokio::fs::rename(&tmp_path, &out_path).await?; - // Guarded finalize: commit ONLY if our generation is still current. If a reopen→ - // re-release bumped `release_seq` while we were streaming, we lost — discard our (now - // stale) archive and let the fresh worker own the keepsake. This is the fix for the - // stale-keepsake data loss: a superseded worker can no longer flip `export_zip_ready`. - if !finalize_job(pool, event_id, "zip", seq, &format!("exports/{out_name}")).await { + // Commit ONLY if our generation is still current. `finalize_job` is guarded on `epoch` — a + // predicate on the very row it updates, so Postgres re-evaluates it correctly even when the + // statement blocks behind a concurrent reopen. If we lost, our archive is stale: discard it. + // + // Note what ISN'T here any more: the ready-flag flip. Readiness is derived from + // (released AND job.epoch = event.epoch AND status = 'done'), so writing `done` at a live epoch + // IS the publish, atomically. A worker at a dead epoch simply writes a row nobody can see. + if !finalize_job(pool, event_id, "zip", epoch, &format!("exports/{out_name}")).await { let _ = tokio::fs::remove_file(&out_path).await; - tracing::info!("ZIP export for event {event_id} superseded by a newer release; discarded"); + tracing::info!("ZIP export for event {event_id} superseded (epoch {epoch} retired); discarded"); return Ok(()); } - // Flip the ready flag only while our generation is still current AND the event is still - // released. Two layers guard this against a reopen resurrecting a pre-reopen keepsake: - // 1. `release_seq = $2` — a reopen bumps every export_job's seq (see `open_event`), so a - // stale worker holding the old seq matches nothing here. - // 2. `export_released_at IS NOT NULL` — belt-and-suspenders in case any path clears the - // release without bumping the seq. - // If this flip is a no-op (0 rows), a reopen/supersession has landed: the keepsake isn't - // downloadable and a fresh generation owns cleanup + the completion signal, so we must NOT - // advertise 100% or prune on this superseded generation's behalf. - let flipped = sqlx::query( - "UPDATE event SET export_zip_ready = TRUE - WHERE id = $1 - AND export_released_at IS NOT NULL - AND EXISTS (SELECT 1 FROM export_job - WHERE event_id = $1 AND type = 'zip'::export_type - AND release_seq = $2 AND status = 'done')", - ) - .bind(event_id) - .bind(seq) - .execute(pool) - .await?; - - if flipped.rows_affected() == 0 { - // Superseded between finalize and here. Discard our now-stale archive (mirroring the - // finalize-lost branch above) instead of leaving it to linger: if the host reopens and - // never re-releases, no future generation would ever come along to prune it. We still - // sweep strictly-older generations — that's safe from any worker, and keeps a reopen that - // is never followed by a re-release from stranding the whole export dir on disk. - let _ = tokio::fs::remove_file(&out_path).await; - prune_stale_export_files(&exports_dir, "Gallery", event_id, seq).await; - tracing::info!("ZIP export for event {event_id} superseded before ready-flip; discarded"); - return Ok(()); - } - - prune_stale_export_files(&exports_dir, "Gallery", event_id, seq).await; + prune_stale_export_files(&exports_dir, "Gallery", event_id, epoch).await; let _ = sse_tx.send(SseEvent { event_type: "export-progress".to_string(), @@ -369,28 +433,33 @@ impl MediaSource { async fn run_html_export( event_id: Uuid, + epoch: i64, event_name: &str, pool: &PgPool, media_path: &Path, export_path: &Path, sse_tx: &broadcast::Sender, ) -> Result<()> { - let seq = match claim_job(pool, event_id, "html").await { - Some(seq) => seq, - // Another worker already owns this HTML export — bail rather than race the temp dir. - None => return Ok(()), - }; + if !claim_job(pool, event_id, "html", epoch).await? { + // Another worker owns this generation, or our epoch has been retired. Not ours. + return Ok(()); + } let res = - run_html_export_inner(seq, event_id, event_name, pool, media_path, export_path, sse_tx).await; - if let Err(e) = &res { - mark_failed(pool, event_id, "html", seq, &e.to_string()).await; + run_html_export_inner(epoch, event_id, event_name, pool, media_path, export_path, sse_tx).await; + if res.is_err() { + // Clean up this generation's temp artifacts so a failing export can't leak them (the leak + // is what fills the disk, which is what corrupts the next archive). + let _ = + tokio::fs::remove_file(export_path.join(gen_name(event_id, "Memories", epoch, ".tmp"))).await; + let _ = tokio::fs::remove_dir_all(export_path.join(format!("viewer_tmp_{event_id}_{epoch}"))) + .await; } res } async fn run_html_export_inner( - seq: i64, + epoch: i64, event_id: Uuid, event_name: &str, pool: &PgPool, @@ -404,14 +473,14 @@ async fn run_html_export_inner( let hashtags_per_upload = query_hashtags(pool, event_id).await?; let total = uploads.len().max(1) as f32; - update_progress(pool, event_id, "html", seq, 5).await; + update_progress(pool, event_id, "html", epoch, 5).await; // Written OUTSIDE media_path: the public /media ServeDir must never reach these. let exports_dir = export_path.to_path_buf(); tokio::fs::create_dir_all(&exports_dir).await?; // 2. Create temp directory for media processing (per-generation — see run_zip_export). - let tmp_dir = exports_dir.join(format!("viewer_tmp_{event_id}_{seq}")); + let tmp_dir = exports_dir.join(format!("viewer_tmp_{event_id}_{epoch}")); let media_tmp = tmp_dir.join("media"); tokio::fs::create_dir_all(&media_tmp).await?; @@ -573,7 +642,7 @@ async fn run_html_export_inner( }); let pct = 10 + ((i + 1) as f32 / total * 60.0) as i16; - update_progress(pool, event_id, "html", seq, pct.min(69)).await; + update_progress(pool, event_id, "html", epoch, pct.min(69)).await; } // 4. Build data.json @@ -587,11 +656,11 @@ async fn run_html_export_inner( let data_json = serde_json::to_string_pretty(&viewer_data).context("failed to serialize data.json")?; - update_progress(pool, event_id, "html", seq, 72).await; + update_progress(pool, event_id, "html", epoch, 72).await; // 5. Create ZIP (per-generation paths — see run_zip_export) - let tmp_path = exports_dir.join(format!("Memories.zip.{seq}.tmp")); - let out_name = format!("Memories.{seq}.zip"); + let tmp_path = exports_dir.join(gen_name(event_id, "Memories", epoch, ".tmp")); + let out_name = gen_name(event_id, "Memories", epoch, ".zip"); let out_path = exports_dir.join(&out_name); { @@ -601,7 +670,7 @@ async fn run_html_export_inner( // Write embedded viewer assets (index.html, _app/*, etc.) write_dir_to_zip(&VIEWER_DIR, &mut zip).await?; - update_progress(pool, event_id, "html", seq, 75).await; + update_progress(pool, event_id, "html", epoch, 75).await; // Write data.json { @@ -621,7 +690,7 @@ async fn run_html_export_inner( entry.close().await?; } - update_progress(pool, event_id, "html", seq, 78).await; + update_progress(pool, event_id, "html", epoch, 78).await; // Write media files from the manifest built in step 3. Thumbnails and derived // image variants stream from temp; video and small-image full variants stream @@ -645,10 +714,14 @@ async fn run_html_export_inner( files_written += 1; let pct = 78 + (files_written as f32 / file_total * 20.0) as i16; - update_progress(pool, event_id, "html", seq, pct.min(98)).await; + update_progress(pool, event_id, "html", epoch, pct.min(98)).await; } - zip.close().await?; + // Flush + fsync before the rename — see run_zip_export for why dropping the file here + // would silently discard a failed final write (and hand us a truncated keepsake). + let mut file = zip.close().await?.into_inner(); + file.flush().await?; + file.sync_all().await?; } // 6. Finalise @@ -657,38 +730,16 @@ async fn run_html_export_inner( // Clean up temp directory let _ = tokio::fs::remove_dir_all(&tmp_dir).await; - // Guarded finalize — discard if a re-release superseded us mid-export (see run_zip_export). - if !finalize_job(pool, event_id, "html", seq, &format!("exports/{out_name}")).await { + // Epoch-guarded finalize — writing `done` at a live epoch IS the publish (readiness is derived + // from it), so there is no separate ready flag to flip. If our epoch was retired, we lost: + // discard the stale archive. + if !finalize_job(pool, event_id, "html", epoch, &format!("exports/{out_name}")).await { let _ = tokio::fs::remove_file(&out_path).await; - tracing::info!("HTML export for event {event_id} superseded by a newer release; discarded"); + tracing::info!("HTML export for event {event_id} superseded (epoch {epoch} retired); discarded"); return Ok(()); } - // Same two-layer guard as the ZIP flip (see run_zip_export): the reopen seq-bump plus the - // `export_released_at IS NOT NULL` anchor keep a superseded worker from resurrecting - // `export_html_ready` on a pre-reopen snapshot. A no-op flip → superseded → don't advertise/prune. - let flipped = sqlx::query( - "UPDATE event SET export_html_ready = TRUE - WHERE id = $1 - AND export_released_at IS NOT NULL - AND EXISTS (SELECT 1 FROM export_job - WHERE event_id = $1 AND type = 'html'::export_type - AND release_seq = $2 AND status = 'done')", - ) - .bind(event_id) - .bind(seq) - .execute(pool) - .await?; - - if flipped.rows_affected() == 0 { - // Superseded — discard our stale archive, still sweep older generations (see run_zip_export). - let _ = tokio::fs::remove_file(&out_path).await; - prune_stale_export_files(&exports_dir, "Memories", event_id, seq).await; - tracing::info!("HTML export for event {event_id} superseded before ready-flip; discarded"); - return Ok(()); - } - - prune_stale_export_files(&exports_dir, "Memories", event_id, seq).await; + prune_stale_export_files(&exports_dir, "Memories", event_id, epoch).await; let _ = sse_tx.send(SseEvent { event_type: "export-progress".to_string(), @@ -748,66 +799,59 @@ async fn query_hashtags(pool: &PgPool, event_id: Uuid) -> Result Option { - sqlx::query_as::<_, (i64,)>( +/// Every predicate here is on the row being updated — no cross-table `EXISTS`. That is deliberate +/// and load-bearing. Under READ COMMITTED, when an UPDATE blocks on a row lock and the blocker +/// commits, Postgres re-evaluates the WHERE against the *updated target row* but answers subqueries +/// against OTHER tables from the statement's ORIGINAL snapshot. The previous guard +/// (`EXISTS (SELECT 1 FROM event WHERE ... export_released_at IS NOT NULL)`) was therefore unsound: +/// a claim blocking behind a concurrent reopen could see the pre-reopen event snapshot, pass the +/// check, and return the post-bump seq — handing the worker a LIVE generation on an event that was +/// already reopened. A same-row `epoch = $3` predicate is re-evaluated correctly by EPQ, so a +/// retired epoch can never be claimed. +/// +/// Errors are distinguished from a lost claim: silently treating a pool timeout as "someone else +/// owns it" left the row `pending` at 0% with no live worker and no error — a spinner forever. +async fn claim_job(pool: &PgPool, event_id: Uuid, export_type: &str, epoch: i64) -> Result { + let r = sqlx::query( "UPDATE export_job SET status = 'running' - WHERE event_id = $1 AND type = $2::export_type AND status = 'pending' - AND EXISTS (SELECT 1 FROM event - WHERE id = $1 AND export_released_at IS NOT NULL) - RETURNING release_seq", + WHERE event_id = $1 AND type = $2::export_type + AND epoch = $3 AND status = 'pending'", ) .bind(event_id) .bind(export_type) - .fetch_optional(pool) + .bind(epoch) + .execute(pool) .await - .ok() - .flatten() - .map(|(seq,)| seq) + .context("claiming export job")?; + Ok(r.rows_affected() > 0) } -/// Guarded finalize: mark this export `done` and record its file path ONLY if our -/// generation is still current (`release_seq = seq` and still `running`). Returns `true` if -/// we won the finalize, `false` if a re-release superseded us mid-export (in which case the -/// caller must discard its output — the fresh worker owns the keepsake now). Atomic, so it -/// can't race a concurrent re-release. +/// Guarded finalize: mark this export `done` and record its file path ONLY if our generation is +/// still current (`epoch` matches and we still hold it `running`). +/// +/// This IS the publish step. Readiness is derived from `(released AND epoch = event.export_epoch +/// AND status = 'done')`, so a single row write makes the keepsake downloadable — there is no +/// second flag to flip and therefore no window between "done" and "visible". A worker whose epoch +/// was retired matches nothing here and must discard its output. async fn finalize_job( pool: &PgPool, event_id: Uuid, export_type: &str, - seq: i64, + 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 release_seq = $4 AND status = 'running'", + AND epoch = $4 AND status = 'running'", ) .bind(event_id) .bind(export_type) .bind(file_path) - .bind(seq) + .bind(epoch) .execute(pool) .await .map(|r| r.rows_affected() > 0) @@ -831,8 +875,10 @@ fn parse_gen_seq(name: &str, prefix: &str, suffix: &str) -> Option { /// again (their `file_path` was nulled by the re-release that superseded them). Tolerates /// races and IO errors — purely disk hygiene. async fn prune_stale_export_files(exports_dir: &Path, prefix: &str, event_id: Uuid, keep_seq: i64) { - let final_prefix = format!("{prefix}."); // Gallery. / Memories. - let temp_prefix = format!("{prefix}.zip."); // Gallery.zip. / Memories.zip. + // EVERY shape is event-scoped. All events share one exports volume, so a name keyed only by + // generation would let event A's prune delete event B's live keepsake (and let two events + // collide on the same archive path). `viewer_tmp_` was already scoped; the archives were not. + let final_prefix = format!("{prefix}.{event_id}."); // Gallery.. / Memories.. let viewer_prefix = format!("viewer_tmp_{event_id}_"); let mut rd = match tokio::fs::read_dir(exports_dir).await { Ok(rd) => rd, @@ -841,18 +887,18 @@ async fn prune_stale_export_files(exports_dir: &Path, prefix: &str, event_id: Uu while let Ok(Some(entry)) = rd.next_entry().await { let name = entry.file_name(); let name = name.to_string_lossy(); - // Try each artifact shape; a strictly-older seq in any of them marks it for deletion. - // Check the temp shape before the final shape: `Gallery.zip..tmp` also starts with - // `Gallery.` but isn't a `.zip`, so its final-shape parse returns None anyway. - let seq = parse_gen_seq(&name, &temp_prefix, ".tmp") - .or_else(|| parse_gen_seq(&name, &final_prefix, ".zip")) - .or_else(|| { - if prefix == "Memories" { - parse_gen_seq(&name, &viewer_prefix, "") - } else { - None - } - }); + // Try each artifact shape; a strictly-older generation in any of them marks it for + // deletion. Only the FINAL archive and the viewer staging dir are swept here — never a + // `.tmp`, which may belong to a superseded worker that is still streaming into it. Deleting + // that out from under it made its rename fail with a confusing hard error; it cleans up its + // own temp on the way out now. + let seq = parse_gen_seq(&name, &final_prefix, ".zip").or_else(|| { + if prefix == "Memories" { + parse_gen_seq(&name, &viewer_prefix, "") + } else { + None + } + }); if seq.is_some_and(|n| n < keep_seq) { let path = entry.path(); let is_dir = entry.file_type().await.map(|t| t.is_dir()).unwrap_or(false); @@ -865,55 +911,59 @@ async fn prune_stale_export_files(exports_dir: &Path, prefix: &str, event_id: Uu } } -/// Mark this worker's export `failed` — but ONLY for the generation it was running -/// (`release_seq = seq` and still `running`). Without the seq guard, a superseded worker -/// erroring out after a re-release would clobber the FRESH worker's `running`/`pending` row -/// to `failed`, stranding the new keepsake. A superseded worker's failure is a no-op here. -async fn mark_failed(pool: &PgPool, event_id: Uuid, export_type: &str, seq: i64, msg: &str) { +/// Mark this worker's export `failed` — ONLY for the generation it was running (`epoch` matches and +/// it still holds it `running`). A superseded worker's failure is a no-op, so it can't clobber the +/// fresh generation's row and strand the new keepsake. +async fn mark_failed(pool: &PgPool, event_id: Uuid, export_type: &str, epoch: i64, msg: &str) { let _ = sqlx::query( "UPDATE export_job SET status = 'failed', error_message = $3 WHERE event_id = $1 AND type = $2::export_type - AND release_seq = $4 AND status = 'running'", + AND epoch = $4 AND status = 'running'", ) .bind(event_id) .bind(export_type) .bind(msg) - .bind(seq) + .bind(epoch) .execute(pool) .await; } -/// Update the progress bar for THIS generation only (`release_seq = seq`). Seq-guarded so a -/// superseded worker still streaming can't overwrite the fresh generation's progress and make -/// the bar jump backwards. Cosmetic, but keeps the displayed percent monotonic per release. -async fn update_progress(pool: &PgPool, event_id: Uuid, export_type: &str, seq: i64, pct: i16) { +/// Update the progress bar for THIS generation only. Epoch-guarded (and status-guarded) so a +/// superseded worker still streaming can't overwrite the fresh generation's progress or scribble on +/// a row the startup sweep already marked failed. +async fn update_progress(pool: &PgPool, event_id: Uuid, export_type: &str, epoch: i64, pct: i16) { let _ = sqlx::query( "UPDATE export_job SET progress_pct = $3 - WHERE event_id = $1 AND type = $2::export_type AND release_seq = $4", + WHERE event_id = $1 AND type = $2::export_type + AND epoch = $4 AND status = 'running'", ) .bind(event_id) .bind(export_type) .bind(pct) - .bind(seq) + .bind(epoch) .execute(pool) .await; } +/// Broadcast `export-available` once BOTH halves are downloadable. Reads the derived predicate +/// (a `done` row at the event's current epoch, on a released event) rather than a stored flag, so +/// it can never advertise a keepsake that a reopen has already invalidated. async fn maybe_broadcast_complete( pool: &PgPool, event_id: Uuid, sse_tx: &broadcast::Sender, ) { - let row: Option<(bool, bool)> = sqlx::query_as( - "SELECT export_zip_ready, export_html_ready FROM event WHERE id = $1", + let complete: bool = sqlx::query_scalar( + "SELECT COUNT(*) = 2 FROM export_current + WHERE event_id = $1 AND status = 'done'", ) .bind(event_id) - .fetch_optional(pool) + .fetch_one(pool) .await - .unwrap_or(None); + .unwrap_or(false); - if let Some((zip_ready, html_ready)) = row { - if zip_ready && html_ready { + { + if complete { let _ = sse_tx.send(SseEvent { event_type: "export-available".to_string(), data: serde_json::json!({ "types": ["zip", "html"] }).to_string(), diff --git a/backend/src/services/maintenance.rs b/backend/src/services/maintenance.rs index 7651302..2088169 100644 --- a/backend/src/services/maintenance.rs +++ b/backend/src/services/maintenance.rs @@ -49,6 +49,16 @@ pub async fn startup_recovery(pool: &PgPool) { // rejects an already-released event), so `export::recover_exports` re-spawns these // failed-but-released jobs from `main` once `AppState` exists (it needs the media // paths + SSE sender this fn doesn't have). + // + // SINGLE-INSTANCE ASSUMPTION: this sweep is unscoped — it reaps every `running` row, not just + // this process's. That is correct for the supported deployment (one app container; see + // docker-compose.yml), where no other process can own a job at boot. If EventSnap is ever run + // with more than one replica, a booting replica would mark a peer's live workers `failed`. + // The epoch model CONTAINS that (the victim's epoch-guarded finalize simply no-ops, and + // recovery re-arms the job at the SAME epoch, so no keepsake is corrupted — the work is just + // redone), but it would waste an export. Fixing it properly means an owner/lease/heartbeat on + // export_job; that machinery is not worth it for a single-container app, so this is a + // documented constraint rather than a hidden one. match sqlx::query( "UPDATE export_job SET status = 'failed', diff --git a/docs/USER_JOURNEYS.md b/docs/USER_JOURNEYS.md index a80887c..e1deed7 100644 --- a/docs/USER_JOURNEYS.md +++ b/docs/USER_JOURNEYS.md @@ -193,7 +193,18 @@ the Host can clean up later). ## 12. Releasing the export and downloading 1. Host (or Admin) taps **Galerie freigeben** in the dashboard. -2. Server sets `event.export_released_at` and enqueues two background jobs. +2. Server sets `event.export_released_at`, locks uploads, **bumps `event.export_epoch`**, and + enqueues two background jobs — all in ONE transaction (workers spawn after it commits, so a + client disconnecting mid-request can never leave the event released with no export to build). + + The **epoch** is the whole generation model (migration 014). It is bumped in the same UPDATE as + any change to `export_released_at` — release and reopen are its only writers — and each + `export_job` row carries a copy of the epoch it was enqueued for. An export is downloadable + **iff** `released AND job.epoch = event.export_epoch AND job.status = 'done'`. Readiness is + therefore *derived*, never stored: it cannot drift, and a worker whose epoch has been retired + (by a reopen, a re-release, or a takedown) is inert — anything it writes is simply invisible. + A reopen retires the current keepsake instantly, which is why a reopened event serves no export + until the host releases again. 3. ZIP job: streams `Gallery.zip` (`Photos/` + `Videos/`, full-quality originals) directly to disk via `async-zip`. Progress updates via `export-progress` SSE. 4. HTML-viewer job: copies the pre-built viewer assets from @@ -201,6 +212,11 @@ the Host can clean up later). `include_dir!`), generates `data.json` from the database, processes `_thumb`/`_full` variants for each upload, and assembles `Memories.zip`. 5. Both jobs complete → server broadcasts `export-available` SSE. + **Takedowns:** if a host deletes an upload (or a comment) while the gallery is released, the + epoch is bumped and the keepsake is REGENERATED without it. Otherwise a photo removed on request + would live on forever in the already-generated archive — the one place it most needs to be gone. + The download 404s for the few seconds it takes to rebuild, which is the correct answer: serving + the old archive would serve the deleted photo. 6. Any user opens `/export`: - Before release: friendly "Export not yet available" banner. - During generation: progress bars per artifact. diff --git a/e2e/fixtures/db.ts b/e2e/fixtures/db.ts index bdbb9b2..f881632 100644 --- a/e2e/fixtures/db.ts +++ b/e2e/fixtures/db.ts @@ -69,26 +69,51 @@ export const db = { }, /** - * Flip the `export_zip_ready` gate directly. The download handler serves bytes - * only when this boolean is true AND the file exists on disk, so setting it true - * without a file lets tests exercise the "ready but file missing" 404 branch. + * Make an export "ready" (or not) in the epoch model. There is no `export_zip_ready` column any + * more — readiness is DERIVED (`released AND job.epoch = event.export_epoch AND status='done'`), + * so a job is ready exactly when its row carries the event's live epoch. To make a `done` job NOT + * ready we retire it to a dead epoch (-1), which is what a reopen effectively does. + * + * `file_path` is deliberately left NULL, so a "ready" job with no file on disk still exercises + * the download's missing-file 404 branch. */ async setExportZipReady(slug: string, ready: boolean) { await withClient((c) => - c.query(`UPDATE event SET export_zip_ready = $2 WHERE slug = $1`, [slug, ready]) + c.query( + `UPDATE export_job ej + SET epoch = CASE WHEN $2 THEN e.export_epoch ELSE -1 END + FROM event e + WHERE e.id = ej.event_id AND e.slug = $1 AND ej.type = 'zip'`, + [slug, ready] + ) ); }, - /** Insert a pre-baked export job row to skip the (slow) real compression path. */ + /** + * Insert a pre-baked export job row to skip the (slow) real compression path. Stamped with the + * event's CURRENT epoch so it counts as the live generation. + */ async fakeExportJob(eventSlug: string, type: 'zip' | 'html', status: 'pending' | 'running' | 'done') { await withClient(async (c) => { - const ev = await c.query<{ id: string }>(`SELECT id FROM event WHERE slug = $1`, [eventSlug]); + const ev = await c.query<{ id: string; export_epoch: string }>( + `SELECT id, export_epoch FROM event WHERE slug = $1`, + [eventSlug] + ); if (ev.rows.length === 0) throw new Error(`No event with slug ${eventSlug}`); await c.query( - `INSERT INTO export_job (event_id, type, status, progress_pct, completed_at) - VALUES ($1, $2::export_type, $3::export_status, $4, $5) - ON CONFLICT (event_id, type) DO UPDATE SET status = EXCLUDED.status, progress_pct = EXCLUDED.progress_pct`, - [ev.rows[0].id, type, status, status === 'done' ? 100 : 0, status === 'done' ? new Date() : null] + `INSERT INTO export_job (event_id, type, status, progress_pct, completed_at, epoch) + VALUES ($1, $2::export_type, $3::export_status, $4, $5, $6) + ON CONFLICT (event_id, type) DO UPDATE + SET status = EXCLUDED.status, progress_pct = EXCLUDED.progress_pct, + epoch = EXCLUDED.epoch`, + [ + ev.rows[0].id, + type, + status, + status === 'done' ? 100 : 0, + status === 'done' ? new Date() : null, + ev.rows[0].export_epoch, + ] ); }); }, diff --git a/e2e/specs/06-export/export.spec.ts b/e2e/specs/06-export/export.spec.ts index fa17887..48d8a1d 100644 --- a/e2e/specs/06-export/export.spec.ts +++ b/e2e/specs/06-export/export.spec.ts @@ -54,27 +54,32 @@ test.describe('Export — release and download', () => { return (await res.json()).ticket; } - test('ZIP download 404s when the export is not yet marked ready', async ({ guest, db }) => { + test('ZIP download 404s for a `done` job at a RETIRED epoch', async ({ guest, db }) => { const g = await guest('NotReady'); - // Released flag set, but export_zip_ready is still false → must refuse, never serve. + // The event is released and the zip job says `done` — but the job carries a dead epoch, which + // is exactly the state a reopen (or a superseded worker) leaves behind. Readiness is derived + // from `job.epoch = event.export_epoch`, so this archive is NOT current and must never be + // served. A 200 here would be the stale-keepsake bug leaking through the read path. await db.setExportReleased(SLUG, true); await db.fakeExportJob(SLUG, 'zip', 'done'); + await db.setExportZipReady(SLUG, false); // retire the job to a dead epoch const ticket = await mintTicket(g.jwt); const res = await fetch(base + '/api/v1/export/zip?ticket=' + encodeURIComponent(ticket)); - // Pinned to 404 (not [404,200]): a 200 here would mean serving an export that was - // never released for download — a data-exposure regression. This hits the - // `!export_zip_ready` guard. expect(res.status).toBe(404); }); - test('ZIP download 404s when marked ready but the file is missing on disk', async ({ guest, db }) => { + test('ZIP download 404s when the job is current but the file is missing on disk', async ({ + guest, + db, + }) => { const g = await guest('ReadyNoFile'); - // Released AND ready, but no Gallery.zip on disk (we never ran a real export) → - // the handler must 404 on the missing-file check, not 200/500 or serve a stale file. + // Released, and the job is `done` at the LIVE epoch — but no archive was ever written (we + // never ran a real export). The handler must 404 on the missing file, not 200/500 or serve + // something stale. await db.setExportReleased(SLUG, true); - await db.setExportZipReady(SLUG, true); await db.fakeExportJob(SLUG, 'zip', 'done'); + await db.setExportZipReady(SLUG, true); const ticket = await mintTicket(g.jwt); const res = await fetch(base + '/api/v1/export/zip?ticket=' + encodeURIComponent(ticket)); diff --git a/e2e/specs/10-flow-review/export-reopen-rerelease.spec.ts b/e2e/specs/10-flow-review/export-reopen-rerelease.spec.ts index 49580ab..203a8f8 100644 --- a/e2e/specs/10-flow-review/export-reopen-rerelease.spec.ts +++ b/e2e/specs/10-flow-review/export-reopen-rerelease.spec.ts @@ -11,20 +11,16 @@ * `export_*_ready = TRUE` on an archive that predated the reopen — so uploads added * during the reopen window were permanently missing from a "ready" keepsake. * - * The fix: a per-(event,type) generation counter `release_seq`, bumped on every (re)release. - * A worker captures the seq it claimed and writes per-generation temp/final paths - * (`Gallery..zip`); its finalize and ready-flag flip are guarded by `release_seq`, so a - * superseded worker discards its output instead of resurrecting a stale keepsake. The - * download follows the current `done` row's `file_path`, never a fixed name. + * The fix (migration 014): ONE monotonic `event.export_epoch`, bumped in the SAME UPDATE as any + * change to `export_released_at` — release and reopen are its only writers. Each export_job row + * carries a COPY of the epoch it was enqueued for. An export is downloadable IFF + * `released AND job.epoch = event.export_epoch AND job.status = 'done'` — readiness is DERIVED, + * never stored, so it cannot drift and no worker can resurrect it. * - * A second, narrower window on the SAME class of bug (fixed alongside): a reopen (`open_event`) - * landing between a *current* worker's `finalize_job` and its ready-flag flip could resurrect - * `export_zip_ready=TRUE` on a pre-reopen snapshot, and the next re-release would - * `if ready { continue }` and skip regeneration. Two layers close it: (1) `open_event` now bumps - * every `export_job.release_seq`, so a reopen is itself a supersession point — a stale worker's - * captured seq no longer matches its `release_seq`-guarded flip; (2) the flip UPDATEs are also - * anchored on `export_released_at IS NOT NULL` as belt-and-suspenders. Layer (1) is what defeats - * the double-race where a re-release re-arms `export_released_at` before it bumps the seq. + * The consequence that kills the whole bug class: a worker holding a retired epoch is INERT BY + * CONSTRUCTION. Every write it makes is guarded on `epoch` on the very row it updates, so it simply + * matches nothing — no cross-table EXISTS (which would be unsound under READ COMMITTED), no flag to + * flip, nothing to skip. Losing a race can no longer corrupt state; it can only waste work. * * Coverage: * - churn integrity: rapid reopen→re-release yields exactly one INTACT ZIP, no stuck jobs. @@ -32,9 +28,11 @@ * keepsake (the direct data-loss regression). * - supersession (re-release): a re-release SUPERSEDES an in-flight (running) job — it bumps the * generation and regenerates, rather than leaving the stale worker to finalize. - * - supersession (reopen): a reopen ALONE bumps the generation, so a worker holding the - * pre-reopen seq can never flip a stale ready flag (closes the finalize↔flip double-race - * deterministically, without depending on sub-millisecond worker timing). + * - supersession (reopen): a reopen ALONE bumps the event epoch, so a worker holding the + * pre-reopen epoch is retired and can publish nothing. + * - acceptance: every upload the server ACCEPTED (201) is in the keepsake — the upload-path + * TOCTOU that was the real cause of the "stale keepsake" bug. + * - takedown: deleting a photo AFTER release regenerates the keepsake without it. */ import { test, expect } from '../../fixtures/test'; import { Client } from 'pg'; @@ -49,6 +47,8 @@ import { SseListener } from '../../helpers/sse-listener'; const BASE = process.env.E2E_FRONTEND_URL ?? 'http://localhost:3101'; const SLUG = 'e2e-test-event'; const N_UPLOADS = 6; +/** A ZIP holding no entries: just the 22-byte end-of-central-directory record. */ +const EMPTY_ZIP_BYTES = 22; const PG = { host: process.env.E2E_DB_HOST ?? 'localhost', @@ -98,22 +98,40 @@ async function downloadZipEntries(jwt: string): Promise { expect(res.status).toBe(200); const bytes = Buffer.from(await res.arrayBuffer()); expect(bytes.subarray(0, 2).toString('latin1')).toBe('PK'); // ZIP magic — not a stomped file + // A ZIP with no entries is exactly its 22-byte end-of-central-directory record. That is + // structurally VALID (an event with nothing to export), but `unzip -t` exits non-zero on it — so + // check for it explicitly rather than letting it masquerade as corruption. + if (bytes.length === EMPTY_ZIP_BYTES) return []; const dir = mkdtempSync(join(tmpdir(), 'es-zip-')); const zipPath = join(dir, 'Gallery.zip'); writeFileSync(zipPath, bytes); - execFileSync('unzip', ['-t', zipPath], { stdio: 'pipe' }); // CRC integrity; throws on corruption + try { + execFileSync('unzip', ['-t', zipPath], { stdio: 'pipe' }); // CRC integrity; throws on corruption + } catch (e: any) { + const out = `${e.stdout?.toString() ?? ''}${e.stderr?.toString() ?? ''}`; + throw new Error(`keepsake ZIP failed integrity check (${bytes.length} bytes):\n${out}`); + } return execFileSync('unzip', ['-Z1', zipPath], { encoding: 'utf8' }) .split('\n') .map((l) => l.trim()) .filter((l) => l.length > 0 && !l.endsWith('/')); } -async function zipReleaseSeq(): Promise { - const [row] = await pgQuery<{ release_seq: string }>( - `SELECT ej.release_seq FROM export_job ej JOIN event e ON e.id = ej.event_id +/** The epoch the CURRENT zip job row was enqueued for. */ +async function zipJobEpoch(): Promise { + const [row] = await pgQuery<{ epoch: string }>( + `SELECT ej.epoch FROM export_job ej JOIN event e ON e.id = ej.event_id WHERE e.slug = '${SLUG}' AND ej.type = 'zip'` ); - return Number(row.release_seq); + return Number(row.epoch); +} + +/** The event's authoritative generation counter. */ +async function eventEpoch(): Promise { + const [row] = await pgQuery<{ export_epoch: string }>( + `SELECT export_epoch FROM event WHERE slug = '${SLUG}'` + ); + return Number(row.export_epoch); } test.describe('Flow re-review — reopen→re-release export integrity + stale-keepsake (H1)', () => { @@ -138,9 +156,9 @@ test.describe('Flow re-review — reopen→re-release export integrity + stale-k }); expect(rejected.status).toBe(403); - // Churn: reopen (clears release + ready flags) then re-release, several times in - // quick succession. This is the path that used to be able to double-spawn workers - // over the shared temp file. End on a release so the gallery is left released. + // Churn: reopen (retires the generation) then re-release, several times in quick + // succession. This is the path that used to double-spawn workers over a shared temp file. + // End on a release so the gallery is left released. for (let i = 0; i < 4; i++) { expect((await post('/api/v1/host/event/open', host.jwt)).status).toBe(204); const rel = await post('/api/v1/host/gallery/release', host.jwt); @@ -194,14 +212,14 @@ test.describe('Flow re-review — reopen→re-release export integrity + stale-k // // OLD behavior (the bug): the re-enqueue left a `running` row alone, so the stale worker // would eventually finalize and re-flip the ready flag on its pre-reopen snapshot. - // NEW behavior (the fix): the re-enqueue bumps `release_seq` and resets the row, so the + // NEW behavior (the fix): the re-enqueue bumps `epoch` and resets the row, so the // stale generation is superseded — its sentinel is wiped and a fresh worker regenerates. await seedUpload(host.jwt, { caption: 'a' }); await seedUpload(host.jwt, { caption: 'b' }); expect((await post('/api/v1/host/gallery/release', host.jwt)).status).toBe(204); await waitExportDone(host.jwt); - const seqBefore = await zipReleaseSeq(); + const seqBefore = await zipJobEpoch(); // Simulate an in-flight worker owning the zip job at the current generation. await pgQuery( @@ -210,61 +228,65 @@ test.describe('Flow re-review — reopen→re-release export integrity + stale-k WHERE e.id = ej.event_id AND e.slug = '${SLUG}' AND ej.type = 'zip'` ); - // Reopen (clears release + ready flags AND bumps every export_job's release_seq) then + // Reopen (clears release + ready flags AND bumps every export_job's epoch) then // re-release (bumps + resets again). Either bump alone supersedes the pinned running row. expect((await post('/api/v1/host/event/open', host.jwt)).status).toBe(204); expect((await post('/api/v1/host/gallery/release', host.jwt)).status).toBe(204); - // The fresh generation must settle to done at a HIGHER release_seq — proving the stale + // The fresh generation must settle to done at a HIGHER epoch — proving the stale // running row was superseded (not left to finalize) and the sentinel is gone. await waitExportDone(host.jwt); - const [row] = await pgQuery<{ status: string; progress_pct: number; release_seq: string }>( - `SELECT ej.status::text AS status, ej.progress_pct, ej.release_seq + const [row] = await pgQuery<{ status: string; progress_pct: number; epoch: string }>( + `SELECT ej.status::text AS status, ej.progress_pct, ej.epoch FROM export_job ej JOIN event e ON e.id = ej.event_id WHERE e.slug = '${SLUG}' AND ej.type = 'zip'` ); expect(row.status, 'zip job status').toBe('done'); expect(Number(row.progress_pct), 'zip job progress').toBe(100); - expect(Number(row.release_seq), 'release_seq advanced past the superseded generation').toBeGreaterThan( + expect(Number(row.epoch), 'epoch advanced past the superseded generation').toBeGreaterThan( seqBefore ); }); - test('a reopen ALONE bumps release_seq — a pre-reopen worker can never flip a stale ready flag', async ({ + test('a reopen ALONE bumps the epoch — a pre-reopen worker is retired and can publish nothing', async ({ host, }) => { - // Deterministic proof of the reopen-supersession that closes the finalize↔flip double-race. - // The nasty interleaving (worker finalized `done@S`, THEN reopen, THEN a re-release re-arms - // `export_released_at` BEFORE it bumps the seq, THEN the stale worker's flip runs and both - // guards pass) can't be forced by sub-millisecond timing — but its ROOT cause is observable: - // whether a reopen retires the current generation's seq. If it does, the seq the worker - // captured (S) is gone for good, so its `release_seq = S`-guarded flip matches nothing no - // matter when it lands. We assert exactly that invariant. + // Deterministic proof that a reopen retires the current generation. The nasty interleavings + // (a worker finalizing around a reopen/re-release) can't be forced by sub-millisecond timing — + // but their ROOT cause is observable: whether a reopen moves the epoch. If it does, the epoch + // the worker captured is gone for good, so its epoch-guarded finalize matches nothing no matter + // when it lands, and there is no separate ready flag left for it to resurrect. await seedUpload(host.jwt, { caption: 'x' }); expect((await post('/api/v1/host/gallery/release', host.jwt)).status).toBe(204); await waitExportDone(host.jwt); - const seqAtRelease = await zipReleaseSeq(); + const epochAtRelease = await eventEpoch(); + expect(await zipJobEpoch(), 'the done job carries the live epoch').toBe(epochAtRelease); - // Reopen only — no re-release. The export_job row is left `done` at the OLD seq by the old - // code; the fix bumps it here. + // Reopen only — no re-release. expect((await post('/api/v1/host/event/open', host.jwt)).status).toBe(204); - const seqAfterReopen = await zipReleaseSeq(); + const epochAfterReopen = await eventEpoch(); expect( - seqAfterReopen, - 'reopen must bump export_job.release_seq so a pre-reopen worker (seq=' + - seqAtRelease + - ') is superseded and can never flip a stale ready flag' - ).toBeGreaterThan(seqAtRelease); + epochAfterReopen, + `reopen must move the event epoch past ${epochAtRelease}, retiring any worker holding it` + ).toBeGreaterThan(epochAtRelease); - // And the reopen must have cleared readiness (belt-and-suspenders: the stale worker also - // can't satisfy `export_released_at IS NOT NULL`). - const [ev] = await pgQuery<{ zip_ready: boolean; released_at: string | null }>( - `SELECT export_zip_ready AS zip_ready, export_released_at AS released_at - FROM event WHERE slug = '${SLUG}'` + // The job row still carries the OLD epoch — so it no longer matches the event's, which is + // exactly what makes it (and any worker holding it) invisible and inert. + expect(await zipJobEpoch(), 'the stale job row is left behind at the retired epoch').toBe( + epochAtRelease + ); + + // Readiness is DERIVED, so there is no flag to check — the keepsake simply stops being + // downloadable the moment the epoch moves. Assert the observable contract instead. + const [ev] = await pgQuery<{ released_at: string | null }>( + `SELECT export_released_at AS released_at FROM event WHERE slug = '${SLUG}'` ); - expect(ev.zip_ready, 'reopen clears zip readiness').toBe(false); expect(ev.released_at, 'reopen clears the release timestamp').toBeNull(); + + const ticket = await mintTicket(host.jwt); + const dl = await fetch(BASE + '/api/v1/export/zip?ticket=' + encodeURIComponent(ticket)); + expect(dl.status, 'a reopened event serves no keepsake').toBe(404); }); test('open ‖ release churn always converges to a consistent, downloadable keepsake', async ({ @@ -273,9 +295,9 @@ test.describe('Flow re-review — reopen→re-release export integrity + stale-k // Convergence STRESS test, not a regression guard — be honest about what this does and does // not prove. `open_event` reopens the event AND bumps every export_job generation, and those // two statements must be atomic (they run in one transaction): if they were separate - // autocommits, a re-release could slip between them, spawn a FRESH worker at seq N+1, and the - // reopen's second statement would bump that live generation to N+2 — orphaning the worker, - // stranding its row at `running` forever with the host unable to retry. + // autocommits, a re-release could slip between them and spawn a FRESH worker whose generation the + // reopen's second statement would then retire — orphaning it and stranding its row at `running` + // forever with the host unable to retry. open_event is now a single UPDATE, so this cannot occur. // // That interleaving is NOT reproducible from outside the process (verified: this test passes // against a deliberately non-transactional build even with ~90 concurrent open/release pairs — @@ -320,6 +342,84 @@ test.describe('Flow re-review — reopen→re-release export integrity + stale-k expect((await downloadZipEntries(host.jwt)).length).toBe(1); }); + test('EVERY upload the server accepted is in the keepsake (upload-path release TOCTOU)', async ({ + host, + }) => { + // THE contract, and the bug that survived three rounds of fixes to the export state machine + // because it was never in the state machine: + // + // `upload` checked `uploads_locked_at` BEFORE streaming the body (minutes, for a big video) + // and then committed the row WITHOUT re-checking. So a release landing mid-upload let the row + // commit AFTER the export snapshot: the photo appeared in the live feed and was PERMANENTLY + // missing from the keepsake. Nothing ever regenerated it. + // + // The invariant is absolute and does not depend on winning any race: if the server said 201, + // that photo is in the keepsake. If it said 403, it isn't. Race uploads against a release and + // hold the server to exactly that. + const bytes = readFileSync(join(process.cwd(), 'fixtures', 'media', 'sample.jpg')); + + // One upload that is definitely committed before the race, so the keepsake is never trivially + // empty (the release can otherwise win every race and the assertion would prove nothing). + await seedUpload(host.jwt, { caption: 'pre-race' }); + + const inflight = Array.from({ length: 10 }, (_, i) => + uploadRaw(host.jwt, bytes, { filename: `race${i}.jpg`, contentType: 'image/jpeg' }) + ); + // Fire the release while those are in flight. + const releasing = post('/api/v1/host/gallery/release', host.jwt); + + const results = await Promise.all(inflight); + await releasing; + + const accepted = results.filter((r) => r.status >= 200 && r.status < 300).length; + const rejected = results.filter((r) => r.status === 403).length; + expect(accepted + rejected, 'every upload either succeeded or was locked out').toBe( + results.length + ); + + await waitExportDone(host.jwt); + const listing = await downloadZipEntries(host.jwt); + + // +1 for the pre-race upload. + expect( + listing.length, + `server accepted ${accepted} racing upload(s) (+1 pre-race) and locked out ${rejected}, but ` + + `the keepsake holds ${listing.length}. An accepted photo missing from the keepsake is ` + + `silent, permanent data loss — exactly the bug this guards.` + ).toBe(accepted + 1); + }); + + test('deleting a photo AFTER release regenerates the keepsake without it (takedown)', async ({ + host, + }) => { + // A host deleting a photo on a removal request used to remove it from the feed but leave it in + // the already-generated archive FOREVER — the `if ready { continue }` skip guaranteed the + // export was never rebuilt. The keepsake is the one place it most needs to be gone. + const keep = await seedUpload(host.jwt, { caption: 'keep' }); + const takedown = await seedUpload(host.jwt, { caption: 'takedown' }); + + expect((await post('/api/v1/host/gallery/release', host.jwt)).status).toBe(204); + await waitExportDone(host.jwt); + expect((await downloadZipEntries(host.jwt)).length).toBe(2); + + // Take it down while the gallery is released. + const del = await fetch(BASE + `/api/v1/host/upload/${takedown}`, { + method: 'DELETE', + headers: { Authorization: `Bearer ${host.jwt}` }, + }); + expect(del.status).toBe(204); + + // The keepsake must rebuild — and must NOT still contain the removed photo. + await waitExportDone(host.jwt); + const listing = await downloadZipEntries(host.jwt); + expect(listing.length, `zip entries:\n${listing.join('\n')}`).toBe(1); + expect( + listing.some((n) => n.includes(takedown)), + 'the taken-down photo must be gone from the regenerated keepsake' + ).toBe(false); + expect(listing.some((n) => n.includes(keep)), 'the kept photo survives').toBe(true); + }); + test('a release broadcasts export-progress 100 for both types and one export-available', async ({ host, }) => {