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

refactor(export): single epoch authority; fix upload-path TOCTOU losing photos

Browse Source 
A deep investigative review of the export concurrency (four parallel adversarial
reviewers) found that three rounds of race fixes had been chasing the wrong bug, and
that the model itself was the root cause. This replaces the model and fixes the real
data loss.

THE ACTUAL BUG (upload path, not the state machine)
  `upload` checked `uploads_locked_at` BEFORE streaming the body — minutes, for a large
  video — then committed the row without re-checking. 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, with nothing to regenerate it. release_gallery's
  comment claims to prevent exactly this; it never did. Every prior round fixed the DB
  state machine, and the leak was upstream of it.
  Fix: re-assert the lock under `SELECT ... FOR SHARE` INSIDE the commit tx. That row lock
  conflicts with release_gallery's `UPDATE event`, so either the upload commits first (and
  the release — hence the snapshot — is ordered after it) or the release wins and the
  upload is rejected as `uploads_locked` (reversible; the client keeps the blob).
  Regression test verified NON-VACUOUS: with the fix reverted it fails, reporting accepted
  uploads missing from the keepsake.

THE MODEL (migration 014)
  "Which generation is current?" had no single answer: it was assembled from three
  separately-written pieces across two tables (`export_released_at`, a per-job
  `release_seq`, and cached `export_{zip,html}_ready` flags). Keeping them in agreement
  took ~10 hand-written guards; each review round found one more path that slipped through.
  Now: ONE monotonic `event.export_epoch`, bumped in the same UPDATE as any change to
  `export_released_at`. Each job carries a copy. Downloadable IFF
  `released AND job.epoch = event.export_epoch AND status = 'done'` — readiness DERIVED,
  never stored. A retired-epoch worker is INERT BY CONSTRUCTION: losing a race can no
  longer corrupt state, only waste work.
  Deleted: both ready-flag columns, both ready-flip blocks, `if ready { continue }`, the
  reopen seq-bump, open_event's transaction, and the cross-table claim guard.

  That last one was also UNSOUND: under READ COMMITTED, a blocked UPDATE re-evaluates its
  WHERE against the updated target row but answers subqueries on OTHER tables from the
  original snapshot — so the previous `EXISTS (SELECT ... FROM event ...)` claim guard
  could admit a claim against an already-reopened event while RETURNING the post-bump seq.
  Every guard is now same-row (`epoch = $n`), which EPQ re-evaluates correctly.

OTHER REAL DEFECTS FIXED
  - release_gallery committed the release and THEN awaited the enqueue inside the handler
    future. Axum drops that future on client disconnect → event released, uploads locked,
    ZERO export_job rows, host unable to retry ("bereits freigegeben"), restart-only
    recovery. Now one tx; workers spawn (detached) after commit.
  - No flush/fsync before the tmp→final rename. async_zip's close() never flushes the inner
    file and tokio's File drop DISCARDS write errors — so a full disk silently produced a
    truncated archive that was then marked done and published, after which the last good
    generation was pruned. Now flush + sync_all, which also surfaces ENOSPC.
  - Download read the ready flag and the file path in TWO queries; a reopen between them
    served a keepsake that should 404. Now one read through the `export_current` view.
  - `if !src.exists() { continue }` then `open()?` — a TOCTOU that turned a tolerated gap
    into a hard failure of the ENTIRE keepsake (unretryable while released). Now open-first,
    warn, and skip.
  - Recovery was flag-driven and never checked the disk: a `done` row whose archive was gone
    was a permanent dead end needing manual DB surgery. Now verifies the file and re-arms.
  - Temp files/dirs leaked on every error path (the leak is what fills the disk that
    corrupts the next archive). Now cleaned up.
  - Export archives were not event-scoped (`viewer_tmp_` already was), so a second event
    would collide on paths and one event's prune would delete another's live keepsake.
  - Host deletes after release never regenerated the keepsake — a photo removed on request
    lived on in the archive forever. Now bumps the epoch and rebuilds without it.
  - claim_job swallowed DB errors as "someone else won", stranding a job pending at 0%.
  - export_status reported retired-epoch rows (a progress bar that never moves).
  - The startup sweep's single-instance assumption is now documented, not hidden.

Verified: backend 40 tests, svelte-check 0 errors, e2e typecheck clean,
e2e 160 passed / 1 skipped on chromium-desktop (incl. 2 new regressions; the
upload-acceptance one confirmed to fail without the fix).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
fabi
2026-07-14 19:48:38 +02:00
parent 1148c2e906
commit 8e906d7866
12 changed files with 802 additions and 430 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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(),