fabi/EventSnap
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'refactor/export-epoch-2026-07-14'

Browse Source
This commit is contained in:
fabi
2026-07-14 19:48:38 +02:00
parent 1148c2e906 8e906d7866
commit 5affef47cf
12 changed files with 802 additions and 430 deletions
Download Patch File Download Diff File Expand all files Collapse all files

28
backend/migrations/014_export_epoch.down.sql Normal file
View File

@@ -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;

79
backend/migrations/014_export_epoch.up.sql Normal file
View File

@@ -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;

60
backend/src/handlers/admin.rs
View File

@@ -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.<seq>.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<std::path::PathBuf, AppError> {
let file_path: Option<(Option<String>,)> = 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/<name>`; 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?;

169
backend/src/handlers/host.rs
View File

@@ -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<AppState>,
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<AppState>,
RequireHost(_auth): RequireHost,
) -> Result<StatusCode, AppError> {
// 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<AppState>,
RequireHost(_auth): RequireHost,
) -> Result<StatusCode, AppError> {
// 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)
}

31
backend/src/handlers/upload.rs
View File

@@ -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<Upload, AppError> = 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<DateTime<Utc>>, Option<DateTime<Utc>>) =
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

7
backend/src/models/event.rs
View File

@@ -11,8 +11,11 @@ pub struct Event {
pub is_active: bool,
pub uploads_locked_at: Option<DateTime<Utc>>,
pub export_released_at: Option<DateTime<Utc>>,
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<Utc>,
}

548
backend/src/services/export.rs
View File

@@ -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<SseEvent>,
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 = <captured>`) 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<SseEvent>,
) {
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<String>)> = 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<SseEvent>,
) -> 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<SseEvent>,
) -> 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<Vec<(Uuid, Stri
Ok(rows)
}
/// Atomically claim a pending export job for this worker and return the generation
/// (`release_seq`) it claimed. Returns `Some(seq)` only if we won (status flipped
/// `pending`→`running` in one statement); `None` when another worker already owns it — e.g.
/// a reopen→re-release spawned a second worker while a run from a prior release is still in
/// flight. The row-level lock serializes the two UPDATEs, so exactly one sees
/// `status = 'pending'`. The caller threads the returned seq into per-generation temp/final
/// paths and into the guarded finalize, so a superseded worker never corrupts or resurrects
/// a stale keepsake.
/// Claim the pending job for `(event, type)` and return the `release_seq` we own.
/// Claim the pending job for `(event, type)` AT OUR EPOCH. `true` only if we won it.
///
/// The `export_released_at IS NOT NULL` guard pins a worker to a LIVE release epoch. Without it,
/// a reopen landing between `release_gallery`'s spawn and this claim would leave the row `pending`
/// at the *bumped* seq, and this worker would then claim that seq — a CURRENT generation — and
/// spend minutes exporting a snapshot taken while the event was open and guests were still
/// uploading. On the next re-release (which re-arms `export_released_at` before it bumps the seq)
/// that worker's finalize + ready-flip would both pass their guards, flip ready on its stale
/// snapshot, and make `enqueue_and_spawn_exports` skip regeneration (`if ready { continue }`) —
/// silently serving a keepsake missing every upload from the reopen window. Refusing the claim
/// while the event is un-released closes that: a worker's generation is now always born AND
/// finalized inside a single live release. The unclaimed `pending` row is harmless — the next
/// release resets and re-spawns it.
async fn claim_job(pool: &PgPool, event_id: Uuid, export_type: &str) -> Option<i64> {
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<bool> {
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<i64> {
/// 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.<event>. / Memories.<event>.
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.<n>.tmp` also starts with
// `Gallery.` but isn't a `<n>.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<SseEvent>,
) {
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(),

10
backend/src/services/maintenance.rs
View File

@@ -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',

18
docs/USER_JOURNEYS.md
View File

@@ -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.

45
e2e/fixtures/db.ts
View File

@@ -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,
]
);
});
},

23
e2e/specs/06-export/export.spec.ts
View File

@@ -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));

214
e2e/specs/10-flow-review/export-reopen-rerelease.spec.ts
View File

@@ -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.<seq>.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<string[]> {
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<number> {
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<number> {
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<number> {
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,
}) => {