Merge branch 'fix/export-epoch-2026-07-14'
This commit is contained in:
@@ -515,6 +515,17 @@ pub async fn open_event(
|
|||||||
// release time, so allowing new uploads afterwards would silently diverge the live
|
// release time, so allowing new uploads afterwards would silently diverge the live
|
||||||
// feed from the frozen export. Clearing `export_released_at` (and the readiness
|
// 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.
|
// flags) lets the host re-release later to regenerate a correct, complete keepsake.
|
||||||
|
//
|
||||||
|
// 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?;
|
||||||
|
|
||||||
let result = sqlx::query(
|
let result = sqlx::query(
|
||||||
"UPDATE event
|
"UPDATE event
|
||||||
SET uploads_locked_at = NULL,
|
SET uploads_locked_at = NULL,
|
||||||
@@ -524,10 +535,12 @@ pub async fn open_event(
|
|||||||
WHERE slug = $1 AND (uploads_locked_at IS NOT NULL OR export_released_at IS NOT NULL)",
|
WHERE slug = $1 AND (uploads_locked_at IS NOT NULL OR export_released_at IS NOT NULL)",
|
||||||
)
|
)
|
||||||
.bind(&state.config.event_slug)
|
.bind(&state.config.event_slug)
|
||||||
.execute(&state.pool)
|
.execute(&mut *tx)
|
||||||
.await?;
|
.await?;
|
||||||
|
|
||||||
if result.rows_affected() > 0 {
|
let reopened = result.rows_affected() > 0;
|
||||||
|
|
||||||
|
if reopened {
|
||||||
// A reopen invalidates any released keepsake, so make it a *supersession point* for the
|
// 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
|
// 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
|
// — or one that has already `finalize_job`'d but not yet flipped its ready flag — captured
|
||||||
@@ -535,17 +548,20 @@ pub async fn open_event(
|
|||||||
// become no-ops. Without this, the `export_released_at IS NOT NULL` flip guard alone leaves
|
// 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 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
|
// a stale worker's flip landing in that gap satisfies both guards and resurrects a
|
||||||
// pre-reopen keepsake (silent data loss). Bumping here closes it — the old seq is retired
|
// pre-reopen keepsake (silent data loss). Bumping here retires the old seq for good.
|
||||||
// for good and the next re-release regenerates from a current snapshot.
|
|
||||||
sqlx::query(
|
sqlx::query(
|
||||||
"UPDATE export_job SET release_seq = release_seq + 1
|
"UPDATE export_job SET release_seq = release_seq + 1
|
||||||
FROM event e
|
FROM event e
|
||||||
WHERE e.id = export_job.event_id AND e.slug = $1",
|
WHERE e.id = export_job.event_id AND e.slug = $1",
|
||||||
)
|
)
|
||||||
.bind(&state.config.event_slug)
|
.bind(&state.config.event_slug)
|
||||||
.execute(&state.pool)
|
.execute(&mut *tx)
|
||||||
.await?;
|
.await?;
|
||||||
|
}
|
||||||
|
|
||||||
|
tx.commit().await?;
|
||||||
|
|
||||||
|
if reopened {
|
||||||
let _ = state.sse_tx.send(SseEvent::new("event-opened", "{}"));
|
let _ = state.sse_tx.send(SseEvent::new("event-opened", "{}"));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -326,7 +326,14 @@ async fn run_zip_export_inner(
|
|||||||
.await?;
|
.await?;
|
||||||
|
|
||||||
if flipped.rows_affected() == 0 {
|
if flipped.rows_affected() == 0 {
|
||||||
tracing::info!("ZIP export for event {event_id} superseded before ready-flip; not advertising");
|
// 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(());
|
return Ok(());
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -674,7 +681,10 @@ async fn run_html_export_inner(
|
|||||||
.await?;
|
.await?;
|
||||||
|
|
||||||
if flipped.rows_affected() == 0 {
|
if flipped.rows_affected() == 0 {
|
||||||
tracing::info!("HTML export for event {event_id} superseded before ready-flip; not advertising");
|
// 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(());
|
return Ok(());
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -746,10 +756,25 @@ async fn query_hashtags(pool: &PgPool, event_id: Uuid) -> Result<Vec<(Uuid, Stri
|
|||||||
/// `status = 'pending'`. The caller threads the returned seq into per-generation temp/final
|
/// `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
|
/// paths and into the guarded finalize, so a superseded worker never corrupts or resurrects
|
||||||
/// a stale keepsake.
|
/// a stale keepsake.
|
||||||
|
/// Claim the pending job for `(event, type)` and return the `release_seq` we own.
|
||||||
|
///
|
||||||
|
/// 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> {
|
async fn claim_job(pool: &PgPool, event_id: Uuid, export_type: &str) -> Option<i64> {
|
||||||
sqlx::query_as::<_, (i64,)>(
|
sqlx::query_as::<_, (i64,)>(
|
||||||
"UPDATE export_job SET status = 'running'
|
"UPDATE export_job SET status = 'running'
|
||||||
WHERE event_id = $1 AND type = $2::export_type AND status = 'pending'
|
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",
|
RETURNING release_seq",
|
||||||
)
|
)
|
||||||
.bind(event_id)
|
.bind(event_id)
|
||||||
|
|||||||
@@ -44,6 +44,7 @@ import { tmpdir } from 'node:os';
|
|||||||
import { join } from 'node:path';
|
import { join } from 'node:path';
|
||||||
import { seedUpload } from '../../helpers/seed';
|
import { seedUpload } from '../../helpers/seed';
|
||||||
import { uploadRaw } from '../../helpers/upload-client';
|
import { uploadRaw } from '../../helpers/upload-client';
|
||||||
|
import { SseListener } from '../../helpers/sse-listener';
|
||||||
|
|
||||||
const BASE = process.env.E2E_FRONTEND_URL ?? 'http://localhost:3101';
|
const BASE = process.env.E2E_FRONTEND_URL ?? 'http://localhost:3101';
|
||||||
const SLUG = 'e2e-test-event';
|
const SLUG = 'e2e-test-event';
|
||||||
@@ -265,4 +266,83 @@ test.describe('Flow re-review — reopen→re-release export integrity + stale-k
|
|||||||
expect(ev.zip_ready, 'reopen clears zip readiness').toBe(false);
|
expect(ev.zip_ready, 'reopen clears zip readiness').toBe(false);
|
||||||
expect(ev.released_at, 'reopen clears the release timestamp').toBeNull();
|
expect(ev.released_at, 'reopen clears the release timestamp').toBeNull();
|
||||||
});
|
});
|
||||||
|
|
||||||
|
test('open ‖ release churn always converges to a consistent, downloadable keepsake', async ({
|
||||||
|
host,
|
||||||
|
}) => {
|
||||||
|
// 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.
|
||||||
|
//
|
||||||
|
// 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 —
|
||||||
|
// the window is microseconds). So it does not guard the atomicity fix; that rests on the
|
||||||
|
// transaction being correct by construction. What this DOES pin down is that hammering both
|
||||||
|
// endpoints never leaves the export pipeline in a wedged state — no stuck job, no corrupt or
|
||||||
|
// missing archive — which is the user-visible property we actually care about.
|
||||||
|
await seedUpload(host.jwt, { caption: 'race' });
|
||||||
|
expect((await post('/api/v1/host/gallery/release', host.jwt)).status).toBe(204);
|
||||||
|
await waitExportDone(host.jwt);
|
||||||
|
|
||||||
|
// Hammer both endpoints concurrently. The bad window (between open_event's two autocommit
|
||||||
|
// statements) is microseconds wide, so a couple of sequential pairs never hit it — we need
|
||||||
|
// many in-flight requests contending for pool connections to land a release inside it.
|
||||||
|
// Deliberately unchecked: whichever loses a race legitimately 4xx's ("already released" /
|
||||||
|
// nothing to reopen). We only care that no interleaving strands a job.
|
||||||
|
for (let round = 0; round < 15; round++) {
|
||||||
|
const calls: Promise<unknown>[] = [];
|
||||||
|
for (let i = 0; i < 6; i++) {
|
||||||
|
calls.push(post('/api/v1/host/event/open', host.jwt));
|
||||||
|
calls.push(post('/api/v1/host/gallery/release', host.jwt));
|
||||||
|
}
|
||||||
|
await Promise.all(calls);
|
||||||
|
}
|
||||||
|
|
||||||
|
// Land deterministically in a released state so a fresh generation is definitely in flight.
|
||||||
|
await post('/api/v1/host/event/open', host.jwt);
|
||||||
|
expect((await post('/api/v1/host/gallery/release', host.jwt)).status).toBe(204);
|
||||||
|
|
||||||
|
// An orphaned worker leaves its row `running` forever, so this poll would time out.
|
||||||
|
await waitExportDone(host.jwt);
|
||||||
|
|
||||||
|
const jobs = await pgQuery<{ type: string; status: string }>(
|
||||||
|
`SELECT ej.type::text AS type, ej.status::text AS status
|
||||||
|
FROM export_job ej JOIN event e ON e.id = ej.event_id
|
||||||
|
WHERE e.slug = '${SLUG}'`
|
||||||
|
);
|
||||||
|
expect(jobs.length).toBe(2);
|
||||||
|
for (const j of jobs) expect(j.status, `${j.type} job must not be stranded`).toBe('done');
|
||||||
|
|
||||||
|
// And the keepsake is actually downloadable and intact.
|
||||||
|
expect((await downloadZipEntries(host.jwt)).length).toBe(1);
|
||||||
|
});
|
||||||
|
|
||||||
|
test('a release broadcasts export-progress 100 for both types and one export-available', async ({
|
||||||
|
host,
|
||||||
|
}) => {
|
||||||
|
// The ready-flip is now conditional, and a SUPERSEDED worker deliberately emits neither event.
|
||||||
|
// That makes the happy path worth pinning: `/export/status` polling (what waitExportDone uses)
|
||||||
|
// reads the export_job row, NOT the SSE — so a regression that silently stopped emitting these
|
||||||
|
// would leave every other test green while the /host and /export pages stopped updating live.
|
||||||
|
const sse = new SseListener();
|
||||||
|
await sse.start(host.jwt);
|
||||||
|
|
||||||
|
await seedUpload(host.jwt, { caption: 'sse' });
|
||||||
|
expect((await post('/api/v1/host/gallery/release', host.jwt)).status).toBe(204);
|
||||||
|
|
||||||
|
for (const type of ['zip', 'html']) {
|
||||||
|
await sse.waitForEvent(
|
||||||
|
'export-progress',
|
||||||
|
(e) => e.data?.type === type && e.data?.progress_pct === 100,
|
||||||
|
30_000
|
||||||
|
);
|
||||||
|
}
|
||||||
|
// Fires once BOTH ready flags are set — the authoritative "keepsake is downloadable" signal.
|
||||||
|
await sse.waitForEvent('export-available', () => true, 30_000);
|
||||||
|
|
||||||
|
sse.stop();
|
||||||
|
});
|
||||||
});
|
});
|
||||||
|
|||||||
@@ -43,6 +43,12 @@ export function initExportStatus(): void {
|
|||||||
void refreshExportStatus();
|
void refreshExportStatus();
|
||||||
sseUnsubs.push(onSseEvent('export-progress', () => { void refreshExportStatus(); }));
|
sseUnsubs.push(onSseEvent('export-progress', () => { void refreshExportStatus(); }));
|
||||||
sseUnsubs.push(onSseEvent('export-available', () => { void refreshExportStatus(); }));
|
sseUnsubs.push(onSseEvent('export-available', () => { void refreshExportStatus(); }));
|
||||||
|
// A reopen INVALIDATES the released keepsake server-side (it clears `export_released_at` and
|
||||||
|
// both ready flags). Without refreshing here, the nav would keep advertising a download that
|
||||||
|
// now 404s — the host reopens to collect more photos and every guest still sees "keepsake
|
||||||
|
// ready" until a manual reload. `event-opened` is the only signal for this: a superseded
|
||||||
|
// worker deliberately emits no `export-progress`.
|
||||||
|
sseUnsubs.push(onSseEvent('event-opened', () => { void refreshExportStatus(); }));
|
||||||
}
|
}
|
||||||
|
|
||||||
export function teardownExportStatus(): void {
|
export function teardownExportStatus(): void {
|
||||||
|
|||||||
@@ -42,6 +42,11 @@
|
|||||||
}),
|
}),
|
||||||
onSseEvent('export-available', async () => {
|
onSseEvent('export-available', async () => {
|
||||||
await loadStatus();
|
await loadStatus();
|
||||||
|
}),
|
||||||
|
// A reopen invalidates the release (ready flags cleared server-side) — refresh so this
|
||||||
|
// page stops offering a download that would now 404.
|
||||||
|
onSseEvent('event-opened', async () => {
|
||||||
|
await loadStatus();
|
||||||
})
|
})
|
||||||
);
|
);
|
||||||
});
|
});
|
||||||
|
|||||||
Reference in New Issue
Block a user