fix(user-flow): close offline-queue, export, session, ban & realtime gaps
Comprehensive user-flow review across guest/host/admin roles, then fixes with
e2e regression guards. Highlights:
- Offline upload queue auto-resumes on reconnect: network errors keep items
pending (not error), 4xx are terminal (no infinite retry), quota exhaustion
returns a distinct 413; queue cap + dedup.
- Export lifecycle: release atomically locks uploads; reopen invalidates and
re-release regenerates the keepsake; workers claim their job atomically so a
reopen->re-release can't corrupt the ZIP; startup re-spawns interrupted
exports.
- Sessions slide on activity (no 30-day cliff); JWT expiry deferred to the
revocable session row; sign-out-everywhere + revoke-on-PIN-reset.
- Ban always hides content (v_feed / find_visible_media / export filter
is_banned) but stays a read-only ban per USER_JOURNEYS §10 — sessions are
not revoked, read access + keepsake download preserved.
- Realtime: server-clock SSE delta cursor; event-closed/opened drive the UI
live; feed_delta rate-limited; like returns {liked, like_count} to fix
multi-device drift; lightbox live comments; diashow delta backfill.
- Forgotten-PIN in-app request flow; simultaneous same-name join returns 409;
quota increment is transactional; operator floor.
Adds e2e/specs/10-flow-review/ (offline resume, export integrity, deterministic
anti-race guard) and updates existing specs for the new contracts.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -82,6 +82,91 @@ 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(
|
||||
event_id: Uuid,
|
||||
event_name: String,
|
||||
pool: PgPool,
|
||||
media_path: PathBuf,
|
||||
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.
|
||||
sqlx::query(
|
||||
"INSERT INTO export_job (event_id, type, status, progress_pct)
|
||||
VALUES ($1, $2::export_type, 'pending', 0)
|
||||
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'",
|
||||
)
|
||||
.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(())
|
||||
}
|
||||
|
||||
/// 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).
|
||||
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)",
|
||||
)
|
||||
.fetch_all(&pool)
|
||||
.await
|
||||
{
|
||||
Ok(r) => r,
|
||||
Err(e) => {
|
||||
tracing::error!("export recovery: failed to query released events: {e:#}");
|
||||
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(
|
||||
event_id,
|
||||
event_name,
|
||||
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:#}");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub fn spawn_export_jobs(
|
||||
event_id: Uuid,
|
||||
event_name: String,
|
||||
@@ -124,7 +209,10 @@ async fn run_zip_export(
|
||||
export_path: &Path,
|
||||
sse_tx: &broadcast::Sender<SseEvent>,
|
||||
) -> Result<()> {
|
||||
mark_running(pool, event_id, "zip").await;
|
||||
if !claim_job(pool, event_id, "zip").await {
|
||||
// Another worker already owns this ZIP export — bail rather than race the temp file.
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
let uploads = query_uploads(pool, event_id).await?;
|
||||
let total = uploads.len().max(1) as f32;
|
||||
@@ -217,7 +305,10 @@ async fn run_html_export(
|
||||
export_path: &Path,
|
||||
sse_tx: &broadcast::Sender<SseEvent>,
|
||||
) -> Result<()> {
|
||||
mark_running(pool, event_id, "html").await;
|
||||
if !claim_job(pool, event_id, "html").await {
|
||||
// Another worker already owns this HTML export — bail rather than race the temp dir.
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
// 1. Query data
|
||||
let uploads = query_uploads(pool, event_id).await?;
|
||||
@@ -511,7 +602,8 @@ async fn query_uploads(pool: &PgPool, event_id: Uuid) -> Result<Vec<ExportUpload
|
||||
FROM upload u
|
||||
JOIN \"user\" usr ON usr.id = u.user_id
|
||||
LEFT JOIN \"like\" l ON l.upload_id = u.id
|
||||
WHERE u.event_id = $1 AND u.deleted_at IS NULL AND usr.uploads_hidden = FALSE
|
||||
WHERE u.event_id = $1 AND u.deleted_at IS NULL
|
||||
AND usr.uploads_hidden = FALSE AND usr.is_banned = FALSE
|
||||
GROUP BY u.id, usr.display_name
|
||||
ORDER BY u.created_at ASC",
|
||||
)
|
||||
@@ -548,14 +640,24 @@ async fn query_hashtags(pool: &PgPool, event_id: Uuid) -> Result<Vec<(Uuid, Stri
|
||||
Ok(rows)
|
||||
}
|
||||
|
||||
async fn mark_running(pool: &PgPool, event_id: Uuid, export_type: &str) {
|
||||
let _ = sqlx::query(
|
||||
"UPDATE export_job SET status = 'running' WHERE event_id = $1 AND type = $2::export_type",
|
||||
/// 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(
|
||||
"UPDATE export_job SET status = 'running'
|
||||
WHERE event_id = $1 AND type = $2::export_type AND status = 'pending'",
|
||||
)
|
||||
.bind(event_id)
|
||||
.bind(export_type)
|
||||
.execute(pool)
|
||||
.await;
|
||||
.await
|
||||
.map(|r| r.rows_affected() > 0)
|
||||
.unwrap_or(false)
|
||||
}
|
||||
|
||||
async fn mark_failed(pool: &PgPool, event_id: Uuid, export_type: &str, msg: &str) {
|
||||
|
||||
Reference in New Issue
Block a user