fix(user-flow): persona-audit fixes — ban replay, locked-upload data loss, host UX, admin session
Follows the perf + security + user-flow work with a role/persona audit (guest, host, admin, projector) and fixes across three review rounds. Highlights: HIGH - Ban now replays on reconnect. A ban isn't a soft-delete, and the `user-hidden` SSE has no replay, so a client that missed it (esp. the unattended diashow) kept cycling a banned user's slides. New `uploads_hidden_at` (migration 013) + `hidden_user_ids` in /feed/delta; feed + diashow evict those users. Applied even on a truncated delta. MEDIUM - Locked-upload data loss: a photo staged offline during a lock/release was purged as a terminal 4xx and lost when the host reopened. New reversible `uploads_locked` error code; the queue keeps the blob and auto-resumes on the `event-opened` SSE. - Reopen after release now warns (ConfirmSheet) that it revokes the published keepsake. - Host "forgotten-PIN" badge updates live (`pin-reset-requested` was broadcast but never in KNOWN_EVENTS / subscribed); host page also refetches on `pin-reset` so a two-host race can't hand out a conflicting PIN. - Ban modal copy fixed (read-only ban, not "session ended"); Degradieren/Sperren/Entsperren hidden on peer-host rows for non-admins (they always 403'd). - Host dashboard shows live keepsake generation progress / ready state + link to /export. - Admin JWT moved to sessionStorage (§11.1) to bound exposure on shared devices. Export generation guard (H1 from the prior round, hardened): per-(event,type) `release_seq` (migration 012) with seq-guarded claim/finalize/mark_failed/update_progress, per-generation temp/final paths, download follows `file_path`, prune only strictly-older generations. LOW: diashow coalesces upload-processed (avoids self-rate-limit); event-closed reconciles galleryReleased; /recover gains a forgot-PIN request + drops a stale cached PIN on 401; delta `>=` tie-break + 429 retry; misc copy/labels. Adds e2e: ban-replay, upload-lock-code, and rewrites export-reopen-rerelease with a data-completeness test. Reconciles USER_JOURNEYS §9/§11. Verified: cargo build clean, 40 unit tests, svelte-check 0 errors, 33 frontend unit tests, 155 e2e passing on chromium-desktop. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
@@ -6,6 +6,11 @@ pub enum AppError {
|
||||
BadRequest(String),
|
||||
Unauthorized(String),
|
||||
Forbidden(String),
|
||||
/// Uploads are temporarily locked (event closed / gallery released). Distinct from
|
||||
/// `Forbidden` so the client can tell this REVERSIBLE 403 apart from a permanent one
|
||||
/// (banned user, quota): the queued blob is kept and retried if the host reopens,
|
||||
/// instead of being purged like a genuinely-terminal rejection.
|
||||
UploadsLocked(String),
|
||||
NotFound(String),
|
||||
Conflict(String),
|
||||
/// Second field: optional retry-after seconds to include in the response.
|
||||
@@ -23,6 +28,7 @@ impl AppError {
|
||||
Self::BadRequest(_) => (StatusCode::BAD_REQUEST, "bad_request"),
|
||||
Self::Unauthorized(_) => (StatusCode::UNAUTHORIZED, "unauthorized"),
|
||||
Self::Forbidden(_) => (StatusCode::FORBIDDEN, "forbidden"),
|
||||
Self::UploadsLocked(_) => (StatusCode::FORBIDDEN, "uploads_locked"),
|
||||
Self::NotFound(_) => (StatusCode::NOT_FOUND, "not_found"),
|
||||
Self::Conflict(_) => (StatusCode::CONFLICT, "conflict"),
|
||||
Self::TooManyRequests(..) => (StatusCode::TOO_MANY_REQUESTS, "too_many_requests"),
|
||||
@@ -36,6 +42,7 @@ impl AppError {
|
||||
Self::BadRequest(msg)
|
||||
| Self::Unauthorized(msg)
|
||||
| Self::Forbidden(msg)
|
||||
| Self::UploadsLocked(msg)
|
||||
| Self::NotFound(msg)
|
||||
| Self::Conflict(msg) => msg.clone(),
|
||||
Self::TooManyRequests(msg, _) => msg.clone(),
|
||||
|
||||
@@ -298,12 +298,42 @@ pub async fn download_zip(
|
||||
));
|
||||
}
|
||||
|
||||
let path = state.config.export_path.join("Gallery.zip");
|
||||
let path = resolve_export_file(&state, "zip").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.
|
||||
async fn resolve_export_file(
|
||||
state: &AppState,
|
||||
export_type: &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'",
|
||||
)
|
||||
.bind(&state.config.event_slug)
|
||||
.bind(export_type)
|
||||
.fetch_optional(&state.pool)
|
||||
.await
|
||||
.map_err(|e| AppError::Internal(e.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()))?;
|
||||
let path = state.config.export_path.join(name);
|
||||
if !path.exists() {
|
||||
return Err(AppError::NotFound("Exportdatei nicht gefunden.".into()));
|
||||
}
|
||||
|
||||
serve_file(path, "Gallery.zip", "application/zip").await
|
||||
Ok(path)
|
||||
}
|
||||
|
||||
pub async fn download_html(
|
||||
@@ -324,11 +354,7 @@ pub async fn download_html(
|
||||
));
|
||||
}
|
||||
|
||||
let path = state.config.export_path.join("Memories.zip");
|
||||
if !path.exists() {
|
||||
return Err(AppError::NotFound("Exportdatei nicht gefunden.".into()));
|
||||
}
|
||||
|
||||
let path = resolve_export_file(&state, "html").await?;
|
||||
serve_file(path, "Memories.zip", "application/zip").await
|
||||
}
|
||||
|
||||
|
||||
@@ -181,6 +181,12 @@ pub struct DeltaQuery {
|
||||
pub struct DeltaResponse {
|
||||
pub uploads: Vec<FeedUpload>,
|
||||
pub deleted_ids: Vec<Uuid>,
|
||||
/// Users whose uploads became hidden (banned / uploads_hidden) since `since`. A ban is
|
||||
/// not a soft-delete, so it never appears in `deleted_ids`; without this, a client that
|
||||
/// missed the ephemeral `user-hidden` SSE (a reconnecting projector, most acutely) would
|
||||
/// keep displaying the banned user's already-loaded slides. The client evicts every
|
||||
/// upload from these users on receipt.
|
||||
pub hidden_user_ids: Vec<Uuid>,
|
||||
/// True when the upload query hit `DELTA_LIMIT`: the response carries only the
|
||||
/// newest slice of the gap, so the client must fall back to a full feed refresh
|
||||
/// rather than merging (the older missed uploads are absent and unrecoverable
|
||||
@@ -229,11 +235,17 @@ pub async fn feed_delta(
|
||||
// entire event's uploads in one response. If a client hits the cap it should
|
||||
// fall back to a full feed refresh rather than another delta.
|
||||
const DELTA_LIMIT: i64 = 200;
|
||||
// `>= since` (not `>`): `created_at` is not unique, so a strict `>` anchored to the exact
|
||||
// timestamp of the last-seen upload silently drops a SECOND upload committed in the same
|
||||
// microsecond — an unrecoverable live-update miss. `>=` re-includes the boundary rows;
|
||||
// the client merges by id (see the feed-delta handler's `seen` set), so the duplicate is
|
||||
// harmless while the tied upload is no longer lost. The cursor still advances to the
|
||||
// response's `server_time`, so this doesn't re-fetch on every subsequent delta.
|
||||
let rows = sqlx::query_as::<_, FeedRow>(
|
||||
"SELECT id, user_id, uploader_name, preview_path, thumbnail_path,
|
||||
mime_type, caption, like_count, comment_count, created_at
|
||||
FROM v_feed
|
||||
WHERE event_id = $1 AND created_at > $2
|
||||
WHERE event_id = $1 AND created_at >= $2
|
||||
ORDER BY created_at DESC, id DESC
|
||||
LIMIT $3",
|
||||
)
|
||||
@@ -247,9 +259,23 @@ pub async fn feed_delta(
|
||||
// client to full-refresh instead of merging a partial delta.
|
||||
let truncated = rows.len() as i64 >= DELTA_LIMIT;
|
||||
|
||||
// `>=` for the same tie-break reason as the uploads query above; re-signalling an
|
||||
// already-removed id is idempotent on the client (it filters its list by these ids).
|
||||
let deleted_ids: Vec<(Uuid,)> = sqlx::query_as(
|
||||
"SELECT id FROM upload
|
||||
WHERE event_id = $1 AND deleted_at IS NOT NULL AND deleted_at > $2",
|
||||
WHERE event_id = $1 AND deleted_at IS NOT NULL AND deleted_at >= $2",
|
||||
)
|
||||
.bind(auth.event_id)
|
||||
.bind(q.since)
|
||||
.fetch_all(&state.pool)
|
||||
.await?;
|
||||
|
||||
// Users hidden since the cursor (ban / uploads_hidden). Uncapped like `deleted_ids` and
|
||||
// for the same reason — an eviction the client misses is unrecoverable via a later delta.
|
||||
let hidden_user_ids: Vec<(Uuid,)> = sqlx::query_as(
|
||||
"SELECT id FROM \"user\"
|
||||
WHERE event_id = $1 AND uploads_hidden = TRUE
|
||||
AND uploads_hidden_at IS NOT NULL AND uploads_hidden_at >= $2",
|
||||
)
|
||||
.bind(auth.event_id)
|
||||
.bind(q.since)
|
||||
@@ -285,6 +311,7 @@ pub async fn feed_delta(
|
||||
Ok(Json(DeltaResponse {
|
||||
uploads,
|
||||
deleted_ids: deleted_ids.into_iter().map(|r| r.0).collect(),
|
||||
hidden_user_ids: hidden_user_ids.into_iter().map(|r| r.0).collect(),
|
||||
truncated,
|
||||
server_time,
|
||||
}))
|
||||
|
||||
@@ -149,7 +149,9 @@ pub async fn ban_user(
|
||||
// contradict that model, break the documented "banned guest can still download the
|
||||
// keepsake" flow, and be ineffective anyway (the user could just /recover a new session).
|
||||
sqlx::query(
|
||||
"UPDATE \"user\" SET is_banned = TRUE, uploads_hidden = TRUE WHERE id = $1 AND event_id = $2",
|
||||
"UPDATE \"user\"
|
||||
SET is_banned = TRUE, uploads_hidden = TRUE, uploads_hidden_at = NOW()
|
||||
WHERE id = $1 AND event_id = $2",
|
||||
)
|
||||
.bind(user_id)
|
||||
.bind(auth.event_id)
|
||||
@@ -201,9 +203,12 @@ pub async fn unban_user(
|
||||
}
|
||||
|
||||
// Unban restores visibility too: ban set `uploads_hidden = TRUE`, so clearing only
|
||||
// `is_banned` would leave their content invisible. Clear both.
|
||||
// `is_banned` would leave their content invisible. Clear all three (the timestamp too,
|
||||
// so a future ban stamps a fresh `uploads_hidden_at` the reconnect delta will replay).
|
||||
let result = sqlx::query(
|
||||
"UPDATE \"user\" SET is_banned = FALSE, uploads_hidden = FALSE WHERE id = $1 AND event_id = $2",
|
||||
"UPDATE \"user\"
|
||||
SET is_banned = FALSE, uploads_hidden = FALSE, uploads_hidden_at = NULL
|
||||
WHERE id = $1 AND event_id = $2",
|
||||
)
|
||||
.bind(user_id)
|
||||
.bind(auth.event_id)
|
||||
@@ -242,10 +247,11 @@ pub async fn set_role(
|
||||
}
|
||||
};
|
||||
|
||||
// Look up the current role so we can apply the host-vs-admin guard. Hosts may
|
||||
// promote guests and demote *other* hosts (the user explicitly requested this
|
||||
// expansion). Hosts may not touch admins. Admins may do anything (except change
|
||||
// themselves, blocked above).
|
||||
// Look up the current role so we can apply the host-vs-admin guard. A plain host may
|
||||
// promote/demote GUESTS only; it may not change any host's or admin's role (see the
|
||||
// guard below — this closes the demote-a-peer-host→ban/PIN-reset takeover chain, F1).
|
||||
// Only an admin may change a host's role. Admins may do anything except change
|
||||
// themselves (blocked above).
|
||||
let target = sqlx::query_as::<_, (String,)>(
|
||||
"SELECT role::text FROM \"user\" WHERE id = $1 AND event_id = $2",
|
||||
)
|
||||
@@ -356,8 +362,13 @@ pub async fn reset_user_pin(
|
||||
.await?;
|
||||
|
||||
// A PIN reset means the old credential is compromised/forgotten — revoke every
|
||||
// existing session so old devices must re-authenticate with the new PIN.
|
||||
let _ = Session::delete_all_for_user(&state.pool, user_id).await;
|
||||
// existing session so old devices must re-authenticate with the new PIN. This is a
|
||||
// security-relevant revoke: if it fails, the old sessions stay valid (sessions are
|
||||
// token- not PIN-bound), so surface the error in logs rather than swallowing it
|
||||
// silently while reporting success to the host.
|
||||
if let Err(e) = Session::delete_all_for_user(&state.pool, user_id).await {
|
||||
tracing::error!(error = ?e, user_id = %user_id, "PIN reset: failed to revoke sessions");
|
||||
}
|
||||
|
||||
// Resolve any pending in-app "I forgot my PIN" request for this user.
|
||||
let _ = sqlx::query("DELETE FROM pin_reset_request WHERE user_id = $1")
|
||||
|
||||
@@ -82,11 +82,11 @@ pub async fn stream(
|
||||
|
||||
// The session is only checked once at open. Re-validate it periodically so a
|
||||
// logged-out, expired, OR banned session's stream is closed rather than kept alive
|
||||
// until the client happens to disconnect. Bounds a stale stream to ~60s. Banning
|
||||
// already revokes the user's sessions (so the row is gone), but re-reading the live
|
||||
// user row and dropping on `is_banned` is a cheap defense-in-depth. Only a
|
||||
// *definitive* gone/expired/banned state ends the stream; a transient DB error just
|
||||
// retries next tick.
|
||||
// until the client happens to disconnect. Bounds a stale stream to ~60s. NOTE: a ban is
|
||||
// deliberately read-only and does NOT revoke sessions (see `ban_user`), so the
|
||||
// `is_banned` re-read below is LOAD-BEARING — it is the only thing that stops a banned
|
||||
// user's live push stream. Do not remove it. Only a *definitive* gone/expired/banned
|
||||
// state ends the stream; a transient DB error just retries next tick.
|
||||
let pool = state.pool.clone();
|
||||
let session_hash = token_hash.clone();
|
||||
let session_gone = async move {
|
||||
|
||||
@@ -75,14 +75,19 @@ pub async fn upload(
|
||||
.ok_or_else(|| AppError::NotFound("Event nicht gefunden.".into()))?;
|
||||
if event.uploads_locked_at.is_some() {
|
||||
drain_multipart(multipart).await;
|
||||
return Err(AppError::Forbidden("Uploads sind gesperrt.".into()));
|
||||
// Reversible: a host can reopen the event, so the client keeps the queued blob and
|
||||
// retries on `event-opened` rather than purging it (UploadsLocked, not Forbidden).
|
||||
return Err(AppError::UploadsLocked("Uploads sind gesperrt.".into()));
|
||||
}
|
||||
// Belt-and-suspenders on top of the lock (release ⇒ lock): once the gallery is
|
||||
// released the export has been snapshotted, so a late upload could never make it into
|
||||
// the keepsake. Reject it explicitly rather than silently diverging the live feed.
|
||||
// Also reversible (reopen clears `export_released_at`), so likewise UploadsLocked.
|
||||
if event.export_released_at.is_some() {
|
||||
drain_multipart(multipart).await;
|
||||
return Err(AppError::Forbidden("Galerie wurde bereits freigegeben.".into()));
|
||||
return Err(AppError::UploadsLocked(
|
||||
"Galerie wurde bereits freigegeben.".into(),
|
||||
));
|
||||
}
|
||||
|
||||
// Read config limits from DB
|
||||
|
||||
@@ -96,30 +96,45 @@ pub async fn enqueue_and_spawn_exports(
|
||||
export_path: PathBuf,
|
||||
sse_tx: broadcast::Sender<SseEvent>,
|
||||
) -> Result<()> {
|
||||
for export_type in ["zip", "html"] {
|
||||
// Reset a prior 'done'/'failed' row to 'pending' so this (re)release regenerates —
|
||||
// but DO NOT touch a row that's still 'running'. If a worker from an earlier
|
||||
// release is mid-export, leaving it 'running' makes the worker we spawn below bail
|
||||
// its claim (WHERE status='pending' → 0 rows), so the two never race the temp file.
|
||||
// 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.
|
||||
sqlx::query(
|
||||
"INSERT INTO export_job (event_id, type, status, progress_pct)
|
||||
VALUES ($1, $2::export_type, 'pending', 0)
|
||||
"INSERT INTO export_job (event_id, type, status, progress_pct, release_seq)
|
||||
VALUES ($1, $2::export_type, 'pending', 0, 1)
|
||||
ON CONFLICT (event_id, type) DO UPDATE
|
||||
SET status = 'pending', progress_pct = 0, file_path = NULL,
|
||||
error_message = NULL, completed_at = NULL
|
||||
WHERE export_job.status <> 'running'",
|
||||
error_message = NULL, completed_at = NULL,
|
||||
release_seq = export_job.release_seq + 1",
|
||||
)
|
||||
.bind(event_id)
|
||||
.bind(export_type)
|
||||
.execute(&pool)
|
||||
.await?;
|
||||
}
|
||||
// Clear readiness so /export downloads 404 until the fresh run completes, instead of
|
||||
// serving a stale keepsake that predates a reopen→re-release.
|
||||
sqlx::query("UPDATE event SET export_zip_ready = FALSE, export_html_ready = FALSE WHERE id = $1")
|
||||
.bind(event_id)
|
||||
.execute(&pool)
|
||||
.await?;
|
||||
|
||||
spawn_export_jobs(event_id, event_name, pool, media_path, export_path, sse_tx);
|
||||
Ok(())
|
||||
@@ -182,9 +197,10 @@ 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:#}");
|
||||
mark_failed(&pool, event_id, "zip", &e.to_string()).await;
|
||||
}
|
||||
maybe_broadcast_complete(&pool, event_id, &sse_tx).await;
|
||||
});
|
||||
@@ -194,7 +210,6 @@ pub fn spawn_export_jobs(
|
||||
run_html_export(event_id, &event_name2, &pool2, &media_path2, &export_path2, &sse_tx2).await
|
||||
{
|
||||
tracing::error!("HTML export failed for event {event_id}: {e:#}");
|
||||
mark_failed(&pool2, event_id, "html", &e.to_string()).await;
|
||||
}
|
||||
maybe_broadcast_complete(&pool2, event_id, &sse_tx2).await;
|
||||
});
|
||||
@@ -209,11 +224,29 @@ async fn run_zip_export(
|
||||
export_path: &Path,
|
||||
sse_tx: &broadcast::Sender<SseEvent>,
|
||||
) -> Result<()> {
|
||||
if !claim_job(pool, event_id, "zip").await {
|
||||
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.
|
||||
return Ok(());
|
||||
}
|
||||
None => 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;
|
||||
}
|
||||
res
|
||||
}
|
||||
|
||||
async fn run_zip_export_inner(
|
||||
seq: i64,
|
||||
event_id: Uuid,
|
||||
pool: &PgPool,
|
||||
media_path: &Path,
|
||||
export_path: &Path,
|
||||
sse_tx: &broadcast::Sender<SseEvent>,
|
||||
) -> Result<()> {
|
||||
let uploads = query_uploads(pool, event_id).await?;
|
||||
let total = uploads.len().max(1) as f32;
|
||||
|
||||
@@ -221,8 +254,13 @@ async fn run_zip_export(
|
||||
let exports_dir = export_path.to_path_buf();
|
||||
tokio::fs::create_dir_all(&exports_dir).await?;
|
||||
|
||||
let tmp_path = exports_dir.join("Gallery.zip.tmp");
|
||||
let out_path = exports_dir.join("Gallery.zip");
|
||||
// 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");
|
||||
let out_path = exports_dir.join(&out_name);
|
||||
|
||||
{
|
||||
let file = tokio::fs::File::create(&tmp_path).await?;
|
||||
@@ -247,7 +285,7 @@ async fn run_zip_export(
|
||||
entry.close().await?;
|
||||
|
||||
let pct = ((i + 1) as f32 / total * 100.0) as i16;
|
||||
update_progress(pool, event_id, "zip", pct.min(99)).await;
|
||||
update_progress(pool, event_id, "zip", seq, pct.min(99)).await;
|
||||
}
|
||||
|
||||
zip.close().await?;
|
||||
@@ -255,19 +293,31 @@ async fn run_zip_export(
|
||||
|
||||
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 {
|
||||
let _ = tokio::fs::remove_file(&out_path).await;
|
||||
tracing::info!("ZIP export for event {event_id} superseded by a newer release; discarded");
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
// Flip the ready flag only while still current (a re-release between finalize and here
|
||||
// would have cleared it; don't resurrect it for a superseded generation).
|
||||
sqlx::query(
|
||||
"UPDATE export_job SET status = 'done', progress_pct = 100, file_path = $2, completed_at = NOW()
|
||||
WHERE event_id = $1 AND type = 'zip'::export_type",
|
||||
"UPDATE event SET export_zip_ready = TRUE
|
||||
WHERE id = $1
|
||||
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("exports/Gallery.zip")
|
||||
.bind(seq)
|
||||
.execute(pool)
|
||||
.await?;
|
||||
|
||||
sqlx::query("UPDATE event SET export_zip_ready = TRUE WHERE id = $1")
|
||||
.bind(event_id)
|
||||
.execute(pool)
|
||||
.await?;
|
||||
prune_stale_export_files(&exports_dir, "Gallery", event_id, seq).await;
|
||||
|
||||
let _ = sse_tx.send(SseEvent {
|
||||
event_type: "export-progress".to_string(),
|
||||
@@ -305,25 +355,43 @@ async fn run_html_export(
|
||||
export_path: &Path,
|
||||
sse_tx: &broadcast::Sender<SseEvent>,
|
||||
) -> Result<()> {
|
||||
if !claim_job(pool, event_id, "html").await {
|
||||
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.
|
||||
return Ok(());
|
||||
}
|
||||
None => 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;
|
||||
}
|
||||
res
|
||||
}
|
||||
|
||||
async fn run_html_export_inner(
|
||||
seq: i64,
|
||||
event_id: Uuid,
|
||||
event_name: &str,
|
||||
pool: &PgPool,
|
||||
media_path: &Path,
|
||||
export_path: &Path,
|
||||
sse_tx: &broadcast::Sender<SseEvent>,
|
||||
) -> Result<()> {
|
||||
// 1. Query data
|
||||
let uploads = query_uploads(pool, event_id).await?;
|
||||
let comments = query_comments(pool, event_id).await?;
|
||||
let hashtags_per_upload = query_hashtags(pool, event_id).await?;
|
||||
let total = uploads.len().max(1) as f32;
|
||||
|
||||
update_progress(pool, event_id, "html", 5).await;
|
||||
update_progress(pool, event_id, "html", seq, 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
|
||||
let tmp_dir = exports_dir.join(format!("viewer_tmp_{event_id}"));
|
||||
// 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 media_tmp = tmp_dir.join("media");
|
||||
tokio::fs::create_dir_all(&media_tmp).await?;
|
||||
|
||||
@@ -485,7 +553,7 @@ async fn run_html_export(
|
||||
});
|
||||
|
||||
let pct = 10 + ((i + 1) as f32 / total * 60.0) as i16;
|
||||
update_progress(pool, event_id, "html", pct.min(69)).await;
|
||||
update_progress(pool, event_id, "html", seq, pct.min(69)).await;
|
||||
}
|
||||
|
||||
// 4. Build data.json
|
||||
@@ -499,11 +567,12 @@ async fn run_html_export(
|
||||
let data_json =
|
||||
serde_json::to_string_pretty(&viewer_data).context("failed to serialize data.json")?;
|
||||
|
||||
update_progress(pool, event_id, "html", 72).await;
|
||||
update_progress(pool, event_id, "html", seq, 72).await;
|
||||
|
||||
// 5. Create ZIP
|
||||
let tmp_path = exports_dir.join("Memories.zip.tmp");
|
||||
let out_path = exports_dir.join("Memories.zip");
|
||||
// 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 out_path = exports_dir.join(&out_name);
|
||||
|
||||
{
|
||||
let file = tokio::fs::File::create(&tmp_path).await?;
|
||||
@@ -512,7 +581,7 @@ async fn run_html_export(
|
||||
// Write embedded viewer assets (index.html, _app/*, etc.)
|
||||
write_dir_to_zip(&VIEWER_DIR, &mut zip).await?;
|
||||
|
||||
update_progress(pool, event_id, "html", 75).await;
|
||||
update_progress(pool, event_id, "html", seq, 75).await;
|
||||
|
||||
// Write data.json
|
||||
{
|
||||
@@ -532,7 +601,7 @@ async fn run_html_export(
|
||||
entry.close().await?;
|
||||
}
|
||||
|
||||
update_progress(pool, event_id, "html", 78).await;
|
||||
update_progress(pool, event_id, "html", seq, 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
|
||||
@@ -556,7 +625,7 @@ async fn run_html_export(
|
||||
|
||||
files_written += 1;
|
||||
let pct = 78 + (files_written as f32 / file_total * 20.0) as i16;
|
||||
update_progress(pool, event_id, "html", pct.min(98)).await;
|
||||
update_progress(pool, event_id, "html", seq, pct.min(98)).await;
|
||||
}
|
||||
|
||||
zip.close().await?;
|
||||
@@ -568,19 +637,26 @@ async fn run_html_export(
|
||||
// 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 {
|
||||
let _ = tokio::fs::remove_file(&out_path).await;
|
||||
tracing::info!("HTML export for event {event_id} superseded by a newer release; discarded");
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
sqlx::query(
|
||||
"UPDATE export_job SET status = 'done', progress_pct = 100, file_path = $2, completed_at = NOW()
|
||||
WHERE event_id = $1 AND type = 'html'::export_type",
|
||||
"UPDATE event SET export_html_ready = TRUE
|
||||
WHERE id = $1
|
||||
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("exports/Memories.zip")
|
||||
.bind(seq)
|
||||
.execute(pool)
|
||||
.await?;
|
||||
|
||||
sqlx::query("UPDATE event SET export_html_ready = TRUE WHERE id = $1")
|
||||
.bind(event_id)
|
||||
.execute(pool)
|
||||
.await?;
|
||||
prune_stale_export_files(&exports_dir, "Memories", event_id, seq).await;
|
||||
|
||||
let _ = sse_tx.send(SseEvent {
|
||||
event_type: "export-progress".to_string(),
|
||||
@@ -640,45 +716,138 @@ async fn query_hashtags(pool: &PgPool, event_id: Uuid) -> Result<Vec<(Uuid, Stri
|
||||
Ok(rows)
|
||||
}
|
||||
|
||||
/// Atomically claim a pending export job for this worker. Returns `true` only if we won
|
||||
/// (status flipped `pending`→`running` in one statement). Returns `false` when another
|
||||
/// worker already owns it — e.g. a reopen→re-release spawned a second worker while a run
|
||||
/// from the first release is still in flight. Both write the SAME fixed temp path
|
||||
/// (`Gallery.zip.tmp` / `viewer_tmp_*`), so a loser MUST NOT run or it would interleave
|
||||
/// bytes and corrupt the archive. The row-level lock serializes the two UPDATEs, so
|
||||
/// exactly one sees `status = 'pending'`.
|
||||
async fn claim_job(pool: &PgPool, event_id: Uuid, export_type: &str) -> bool {
|
||||
sqlx::query(
|
||||
/// 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.
|
||||
async fn claim_job(pool: &PgPool, event_id: Uuid, export_type: &str) -> Option<i64> {
|
||||
sqlx::query_as::<_, (i64,)>(
|
||||
"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'
|
||||
RETURNING release_seq",
|
||||
)
|
||||
.bind(event_id)
|
||||
.bind(export_type)
|
||||
.fetch_optional(pool)
|
||||
.await
|
||||
.ok()
|
||||
.flatten()
|
||||
.map(|(seq,)| seq)
|
||||
}
|
||||
|
||||
/// 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.
|
||||
async fn finalize_job(
|
||||
pool: &PgPool,
|
||||
event_id: Uuid,
|
||||
export_type: &str,
|
||||
seq: 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'",
|
||||
)
|
||||
.bind(event_id)
|
||||
.bind(export_type)
|
||||
.bind(file_path)
|
||||
.bind(seq)
|
||||
.execute(pool)
|
||||
.await
|
||||
.map(|r| r.rows_affected() > 0)
|
||||
.unwrap_or(false)
|
||||
}
|
||||
|
||||
async fn mark_failed(pool: &PgPool, event_id: Uuid, export_type: &str, msg: &str) {
|
||||
/// Parse the trailing generation number out of `<prefix><n><suffix>` (e.g.
|
||||
/// `Gallery.<n>.zip`, `Gallery.zip.<n>.tmp`, `viewer_tmp_<event>_<n>`). Returns None if the
|
||||
/// name doesn't fit the shape, so unrelated files are left untouched.
|
||||
fn parse_gen_seq(name: &str, prefix: &str, suffix: &str) -> Option<i64> {
|
||||
name.strip_prefix(prefix)?.strip_suffix(suffix)?.parse::<i64>().ok()
|
||||
}
|
||||
|
||||
/// Best-effort removal of stale per-generation export artifacts for one export type. Deletes
|
||||
/// ONLY strictly-older generations (`n < keep_seq`) — never `keep_seq`'s own current file,
|
||||
/// and never a NEWER generation that a concurrent re-release may already be producing (that
|
||||
/// "delete all but mine" would let a lagging older winner nuke a newer live file). Covers the
|
||||
/// finished archive (`<prefix>.<n>.zip`), its temp (`<prefix>.zip.<n>.tmp`), and — for html —
|
||||
/// the `viewer_tmp_<event>_<n>` staging dir, so crash-orphaned per-generation temps don't
|
||||
/// accumulate. Runs after a worker wins its finalize; older generations can never be served
|
||||
/// 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.
|
||||
let viewer_prefix = format!("viewer_tmp_{event_id}_");
|
||||
let mut rd = match tokio::fs::read_dir(exports_dir).await {
|
||||
Ok(rd) => rd,
|
||||
Err(_) => return,
|
||||
};
|
||||
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
|
||||
}
|
||||
});
|
||||
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);
|
||||
let _ = if is_dir {
|
||||
tokio::fs::remove_dir_all(&path).await
|
||||
} else {
|
||||
tokio::fs::remove_file(&path).await
|
||||
};
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// 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) {
|
||||
let _ = sqlx::query(
|
||||
"UPDATE export_job SET status = 'failed', error_message = $3
|
||||
WHERE event_id = $1 AND type = $2::export_type",
|
||||
WHERE event_id = $1 AND type = $2::export_type
|
||||
AND release_seq = $4 AND status = 'running'",
|
||||
)
|
||||
.bind(event_id)
|
||||
.bind(export_type)
|
||||
.bind(msg)
|
||||
.bind(seq)
|
||||
.execute(pool)
|
||||
.await;
|
||||
}
|
||||
|
||||
async fn update_progress(pool: &PgPool, event_id: Uuid, export_type: &str, pct: i16) {
|
||||
/// 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) {
|
||||
let _ = sqlx::query(
|
||||
"UPDATE export_job SET progress_pct = $3 WHERE event_id = $1 AND type = $2::export_type",
|
||||
"UPDATE export_job SET progress_pct = $3
|
||||
WHERE event_id = $1 AND type = $2::export_type AND release_seq = $4",
|
||||
)
|
||||
.bind(event_id)
|
||||
.bind(export_type)
|
||||
.bind(pct)
|
||||
.bind(seq)
|
||||
.execute(pool)
|
||||
.await;
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user