Three ways the compiled-in viewer could be wrong, none of which anything would have reported. Found by mutation-testing the guard added below — it failed when it should have passed, and the reason was the second bullet. * `include_dir!` registers NO rebuild dependency. Run `npm run build` in frontend/export-viewer, then `cargo build`, and cargo sees no source change and reuses the cached binary — carrying the PREVIOUS index.html. The file on disk and the file in the binary disagree, git is clean, every check passes, and Memories.zip ships a stale viewer. Confirmed empirically: after replacing the artifact the compiled-in copy did not change until a source file was touched. A build.rs now declares `rerun-if-changed` for `static/export-viewer` AND `migrations` — sqlx::migrate!() embeds its directory the same way, and there the stale snapshot is worse still: the binary boots against a database that already ran a newer migration and crash-loops with VersionMissing. * `emptyOutDir: true` deleted the committed artifact BEFORE generating. That was safe while the build could not fail; it no longer is, because `inlineThemeFonts` now calls `this.error` on a keepsake that is not self-contained. A failed build left the directory empty — and include_dir! over an empty directory compiles fine, while `write_viewer_with_data` iterates zero files and returns Ok. The result is a valid archive with every photo and no viewer. The output is one overwritten file, so nothing accumulates without the wipe. * Nothing asserted the viewer was there at all. Now asserted at the point of use (bail rather than write a viewer-less keepsake) and in a test that checks presence, plausible size, and that no `url(/...)` survived inlining — the three ways it can be present but useless. The Dockerfile copies build.rs with the sources rather than with Cargo.toml, so the dependency-cache layer stays byte-identical and the dummy build does not run it.
2661 lines
122 KiB
Rust
2661 lines
122 KiB
Rust
use std::path::{Path, PathBuf};
|
|
use std::time::Duration;
|
|
|
|
use anyhow::{Context, Result};
|
|
use async_zip::tokio::write::ZipFileWriter;
|
|
use async_zip::{Compression, ZipEntryBuilder};
|
|
use chrono::{DateTime, Utc};
|
|
use futures::io::{AllowStdIo, copy as fcopy};
|
|
use include_dir::{Dir, include_dir};
|
|
use serde::Serialize;
|
|
use sqlx::PgPool;
|
|
use tokio::io::AsyncWriteExt;
|
|
use tokio::sync::broadcast;
|
|
use tokio_util::compat::TokioAsyncReadCompatExt;
|
|
use uuid::Uuid;
|
|
|
|
use crate::state::SseEvent;
|
|
|
|
// ── Embedded viewer assets (pre-built SvelteKit static output) ──────────────
|
|
|
|
static VIEWER_DIR: Dir<'_> = include_dir!("$CARGO_MANIFEST_DIR/static/export-viewer");
|
|
|
|
// ── Shared visibility filter ─────────────────────────────────────────────────
|
|
|
|
/// The predicate that decides what lands in a keepsake, as ONE definition.
|
|
///
|
|
/// Two queries have to agree on it: [`query_uploads`], which selects the rows the archives are
|
|
/// built from, and [`estimate_export_bytes`], which sizes them for the disk preflight. They used
|
|
/// to state it separately, and the direction of drift matters — an estimate that misses rows the
|
|
/// archive writes UNDER-reserves, which is the exact ENOSPC the preflight exists to prevent.
|
|
///
|
|
/// A `SRC:`-marked copy in the integration tests cannot catch that: drift means production moved
|
|
/// and the copy didn't, so both sides of such a test sit still and it keeps passing. Sharing the
|
|
/// fragment removes the failure by construction instead, and leaves the test doing what it is
|
|
/// actually good at — pinning the behaviour.
|
|
///
|
|
/// CONTRACT: callers must alias `upload` as `u` and join `"user"` as `usr`, and bind the event id
|
|
/// as `$1`.
|
|
macro_rules! export_visibility_where {
|
|
() => {
|
|
"WHERE u.event_id = $1 AND u.deleted_at IS NULL
|
|
AND usr.uploads_hidden = FALSE AND usr.is_banned = FALSE"
|
|
};
|
|
}
|
|
|
|
// ── DB query rows ────────────────────────────────────────────────────────────
|
|
|
|
#[derive(sqlx::FromRow)]
|
|
struct ExportUploadRow {
|
|
id: Uuid,
|
|
original_path: String,
|
|
mime_type: String,
|
|
caption: Option<String>,
|
|
uploader_name: String,
|
|
like_count: i64,
|
|
created_at: DateTime<Utc>,
|
|
}
|
|
|
|
#[derive(sqlx::FromRow)]
|
|
struct ExportCommentRow {
|
|
upload_id: Uuid,
|
|
uploader_name: String,
|
|
body: String,
|
|
created_at: DateTime<Utc>,
|
|
}
|
|
|
|
// ── Viewer JSON structs (serialised to data.json) ───────────────────────────
|
|
|
|
#[derive(Serialize)]
|
|
struct ViewerData {
|
|
event: ViewerEvent,
|
|
posts: Vec<ViewerPost>,
|
|
}
|
|
|
|
#[derive(Serialize)]
|
|
struct ViewerEvent {
|
|
name: String,
|
|
exported_at: String,
|
|
// Mirrors the live COMMENTS_ENABLED flag so the offline keepsake hides all comment
|
|
// UI (buttons, counts, sections) when the feature was off for the event.
|
|
comments_enabled: bool,
|
|
}
|
|
|
|
#[derive(Serialize)]
|
|
struct ViewerPost {
|
|
id: String,
|
|
uploader: String,
|
|
caption: String,
|
|
tags: Vec<String>,
|
|
timestamp: String,
|
|
likes: i64,
|
|
comments: Vec<ViewerComment>,
|
|
media: ViewerMedia,
|
|
}
|
|
|
|
#[derive(Serialize)]
|
|
struct ViewerComment {
|
|
author: String,
|
|
text: String,
|
|
timestamp: String,
|
|
}
|
|
|
|
#[derive(Serialize)]
|
|
struct ViewerMedia {
|
|
#[serde(rename = "type")]
|
|
media_type: String,
|
|
thumb: String,
|
|
full: String,
|
|
}
|
|
|
|
// ── Entry point ──────────────────────────────────────────────────────────────
|
|
|
|
/// 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,
|
|
epoch: i64,
|
|
) -> Result<()> {
|
|
enqueue_types_at_epoch(conn, event_id, epoch, &["zip", "html"]).await
|
|
}
|
|
|
|
/// Arm a specific subset of the export types at `epoch` (see [`enqueue_jobs_at_epoch`]).
|
|
pub async fn enqueue_types_at_epoch(
|
|
conn: &mut sqlx::PgConnection,
|
|
event_id: Uuid,
|
|
epoch: i64,
|
|
types: &[&str],
|
|
) -> Result<()> {
|
|
for export_type in types {
|
|
sqlx::query(
|
|
"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,
|
|
epoch = EXCLUDED.epoch
|
|
WHERE export_job.status <> 'done' OR export_job.epoch <> EXCLUDED.epoch",
|
|
)
|
|
.bind(event_id)
|
|
.bind(export_type)
|
|
.bind(epoch)
|
|
.execute(&mut *conn)
|
|
.await?;
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
/// The export artifacts a content change can invalidate.
|
|
///
|
|
/// The ZIP holds only the media originals; the HTML viewer additionally embeds captions, likes and
|
|
/// comments. So a comment moderation needs only the viewer rebuilt — rebuilding the multi-GB ZIP for
|
|
/// it would 404 the photo download for minutes to change nothing in it.
|
|
#[derive(Clone, Copy, PartialEq, Eq)]
|
|
pub enum Affects {
|
|
/// Media changed (an upload removed/hidden) — both artifacts are stale.
|
|
Both,
|
|
/// Only viewer content changed (a comment) — carry the ZIP forward untouched.
|
|
ViewerOnly,
|
|
}
|
|
|
|
/// Handle for a regeneration that a caller has ARMED inside its transaction but not yet started.
|
|
/// The workers must only be spawned AFTER that transaction commits, or they could snapshot the
|
|
/// database before the change that triggered them is visible.
|
|
pub struct PendingRegen {
|
|
pub event_id: Uuid,
|
|
pub event_name: String,
|
|
pub epoch: i64,
|
|
}
|
|
|
|
/// Invalidate the current keepsake and arm a rebuild — IN THE CALLER'S TRANSACTION.
|
|
///
|
|
/// Any content removal that happens after a release (a takedown, a ban, a guest deleting their own
|
|
/// photo) must reach the keepsake: the archive is the artifact people keep forever, and "please take
|
|
/// my photo out" is the one request that most needs to be honoured there. Bumping the epoch retires
|
|
/// the current archive instantly (readiness is derived from the epoch), so the stale download stops
|
|
/// being served the moment the change commits, and a fresh worker rebuilds without the content.
|
|
///
|
|
/// Runs in the caller's tx ON PURPOSE. If the removal committed but the regeneration didn't, the
|
|
/// taken-down photo would stay downloadable forever and nothing would notice: the keepsake still
|
|
/// looks complete at the current epoch, so recovery skips it, and the host can no longer even find
|
|
/// the upload to retry. Returns `None` when the event isn't released (nothing to invalidate — the
|
|
/// export is built fresh at release time).
|
|
pub async fn invalidate_and_arm(
|
|
conn: &mut sqlx::PgConnection,
|
|
event_slug: &str,
|
|
affects: Affects,
|
|
) -> Result<Option<PendingRegen>> {
|
|
let bumped: Option<(Uuid, String, i64)> = sqlx::query_as(
|
|
"UPDATE event SET export_epoch = export_epoch + 1
|
|
WHERE slug = $1 AND export_released_at IS NOT NULL
|
|
RETURNING id, name, export_epoch",
|
|
)
|
|
.bind(event_slug)
|
|
.fetch_optional(&mut *conn)
|
|
.await?;
|
|
|
|
let Some((event_id, event_name, epoch)) = bumped else {
|
|
return Ok(None);
|
|
};
|
|
|
|
// A comment-only change doesn't alter the ZIP's contents (the ZIP holds media, not comments), so
|
|
// we'd rather carry the finished archive into the new epoch than spend minutes rebuilding it.
|
|
//
|
|
// But "finished" is the whole precondition, and it is NOT guaranteed: between `release_gallery`
|
|
// and the ZIP worker completing, the row sits at `pending`/`running` — MINUTES, for a real
|
|
// multi-GB gallery — and deleting a comment right after release is an utterly ordinary thing to
|
|
// do. If we blindly re-armed only the viewer, the carry-forward would match nothing, the ZIP row
|
|
// would be left stranded at the retired epoch, and NOTHING would ever re-arm it: the in-flight
|
|
// worker finishes and writes `done` at an epoch `export_current` no longer matches, so
|
|
// `GET /export/zip` 404s forever (short of a boot or the host finding the rebuild button).
|
|
//
|
|
// So the carry-forward's OWN result decides. It matched ⇒ there is a current, finished ZIP and
|
|
// only the viewer needs rebuilding. It didn't ⇒ there is no ZIP to preserve, and the ZIP must be
|
|
// rebuilt at the new epoch like any other invalidation. Never assume; ask the UPDATE.
|
|
let carried = if affects == Affects::ViewerOnly {
|
|
sqlx::query(
|
|
"UPDATE export_job SET epoch = $2
|
|
WHERE event_id = $1 AND type = 'zip'::export_type
|
|
AND status = 'done' AND epoch = $2 - 1",
|
|
)
|
|
.bind(event_id)
|
|
.bind(epoch)
|
|
.execute(&mut *conn)
|
|
.await?
|
|
.rows_affected()
|
|
== 1
|
|
} else {
|
|
false
|
|
};
|
|
|
|
// (`prune_stale_export_files` protects any file a current-epoch row still points at, so a
|
|
// carried archive isn't swept for having an older epoch in its name.)
|
|
let types: &[&str] = if carried { &["html"] } else { &["zip", "html"] };
|
|
enqueue_types_at_epoch(&mut *conn, event_id, epoch, types).await?;
|
|
|
|
Ok(Some(PendingRegen {
|
|
event_id,
|
|
event_name,
|
|
epoch,
|
|
}))
|
|
}
|
|
|
|
/// 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,
|
|
comments_enabled: bool,
|
|
sse_tx: broadcast::Sender<SseEvent>,
|
|
) {
|
|
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
|
|
{
|
|
Ok(r) => r,
|
|
Err(e) => {
|
|
tracing::error!("export recovery: failed to query released events: {e:#}");
|
|
return;
|
|
}
|
|
};
|
|
|
|
// Crash-orphaned temps are immortal otherwise: `prune_stale_export_files` deliberately never
|
|
// touches `.tmp` (a superseded worker may still be streaming into one), and nothing else sweeps
|
|
// them. A hard kill mid-export — especially one whose epoch has since moved on — strands a
|
|
// full-gallery-sized file forever. Boot is the one moment this is unambiguously safe.
|
|
sweep_orphan_temps(&export_path).await;
|
|
|
|
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,
|
|
comments_enabled,
|
|
Duration::ZERO,
|
|
pool.clone(),
|
|
media_path.clone(),
|
|
export_path.clone(),
|
|
sse_tx.clone(),
|
|
);
|
|
}
|
|
}
|
|
|
|
/// Remove every export temp artifact at boot. Called from `recover_exports`, after the startup sweep
|
|
/// has marked all `running` jobs `failed` and before any worker is spawned — so no live worker can
|
|
/// own one of these. A worker that is re-armed will recreate its temp from scratch.
|
|
async fn sweep_orphan_temps(export_path: &Path) {
|
|
let mut rd = match tokio::fs::read_dir(export_path).await {
|
|
Ok(rd) => rd,
|
|
Err(_) => return,
|
|
};
|
|
let mut removed = 0u32;
|
|
while let Ok(Some(entry)) = rd.next_entry().await {
|
|
let name = entry.file_name();
|
|
let name = name.to_string_lossy();
|
|
let is_temp = name.ends_with(".tmp") || name.starts_with("viewer_tmp_");
|
|
if !is_temp {
|
|
continue;
|
|
}
|
|
let is_dir = entry.file_type().await.map(|t| t.is_dir()).unwrap_or(false);
|
|
let r = if is_dir {
|
|
tokio::fs::remove_dir_all(entry.path()).await
|
|
} else {
|
|
tokio::fs::remove_file(entry.path()).await
|
|
};
|
|
if r.is_ok() {
|
|
removed += 1;
|
|
}
|
|
}
|
|
if removed > 0 {
|
|
tracing::warn!("export recovery: swept {removed} orphaned export temp artifact(s)");
|
|
}
|
|
}
|
|
|
|
/// 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(())
|
|
}
|
|
|
|
/// How long a regeneration waits before claiming. A takedown pass ("remove these five photos") is a
|
|
/// burst of independent requests, each of which retires the previous generation. Without a delay,
|
|
/// each one immediately spawns two full exports — and a superseded worker is INERT, not STOPPED, so
|
|
/// it still runs every ffmpeg spawn and image resize and writes an entire archive before discovering
|
|
/// it lost. Five deletes would leave ten workers alive, each holding a full-gallery-sized temp file,
|
|
/// inside a 1 GB container.
|
|
///
|
|
/// Sleeping before `claim_job` collapses the burst for free: a worker whose epoch is retired during
|
|
/// the delay fails its claim and does ZERO work. Release and boot-recovery pass ZERO — those must
|
|
/// start immediately.
|
|
pub const REGEN_DEBOUNCE: Duration = Duration::from_secs(20);
|
|
|
|
/// Tracks when the CURRENT un-served burst of invalidations began, per event.
|
|
///
|
|
/// The starvation this fixes (B7): every invalidation bumps `export_epoch` immediately and arms a
|
|
/// worker that sleeps `REGEN_DEBOUNCE` before claiming — and any further bump inside that window
|
|
/// retires it. Nothing rate-limits epoch bumps; only the HTTP requests are limited, at 120/min for
|
|
/// comment deletion and 30/min for caption edits. So one ordinary guest deleting a comment per
|
|
/// second, or flipping a caption A/B/A, keeps `GET /export/html` at 404 for the rest of the event
|
|
/// while the UI shows "Wird vorbereitet…" forever. **A host moderating faster than one action per
|
|
/// 20 seconds produces the same result by accident** — which is what a takedown pass looks like.
|
|
///
|
|
/// The fix is to measure the debounce from the FIRST request in a burst rather than the latest, so
|
|
/// the wait is bounded no matter how long the burst runs: coalescing still collapses a rapid pass
|
|
/// into one build, but a build always starts within `REGEN_DEBOUNCE` of the burst beginning.
|
|
///
|
|
/// In-memory on purpose. It is a scheduling hint, not state: losing it on restart is harmless
|
|
/// because `recover_exports` re-arms anything unfinished at boot anyway, and the worst case of a
|
|
/// stale entry is one build starting immediately instead of debounced.
|
|
type BurstStarts = std::collections::HashMap<Uuid, std::time::Instant>;
|
|
static REGEN_BURST_START: std::sync::LazyLock<std::sync::Mutex<BurstStarts>> =
|
|
std::sync::LazyLock::new(|| std::sync::Mutex::new(BurstStarts::new()));
|
|
|
|
/// How long to defer the next regen worker for `event_id`, and record that a burst is running.
|
|
///
|
|
/// Returns `REGEN_DEBOUNCE` for the first invalidation of a burst and progressively less for
|
|
/// each one after it, reaching zero once the burst has been going for a full debounce window.
|
|
pub fn regen_delay_for(event_id: Uuid) -> Duration {
|
|
let mut map = match REGEN_BURST_START.lock() {
|
|
Ok(m) => m,
|
|
// A poisoned mutex must not take the keepsake down: fall back to the plain debounce,
|
|
// which is the pre-existing behaviour.
|
|
Err(e) => e.into_inner(),
|
|
};
|
|
let started = *map.entry(event_id).or_insert_with(std::time::Instant::now);
|
|
REGEN_DEBOUNCE.saturating_sub(started.elapsed())
|
|
}
|
|
|
|
/// A regen worker has begun (or the event settled), so the next invalidation starts a fresh burst.
|
|
pub fn clear_regen_burst(event_id: Uuid) {
|
|
if let Ok(mut map) = REGEN_BURST_START.lock() {
|
|
map.remove(&event_id);
|
|
}
|
|
}
|
|
|
|
// Export worker entry point: every argument is state the spawned worker is BORN with (notably
|
|
// `epoch`). Bundling them into a struct would be a pure-refactor risk on the epoch logic for no
|
|
// gain, so the arity stands.
|
|
#[allow(clippy::too_many_arguments)]
|
|
pub fn spawn_export_jobs(
|
|
event_id: Uuid,
|
|
event_name: String,
|
|
epoch: i64,
|
|
comments_enabled: bool,
|
|
delay: Duration,
|
|
pool: PgPool,
|
|
media_path: PathBuf,
|
|
export_path: PathBuf,
|
|
sse_tx: broadcast::Sender<SseEvent>,
|
|
) {
|
|
let pool2 = pool.clone();
|
|
let media_path2 = media_path.clone();
|
|
let export_path2 = export_path.clone();
|
|
let sse_tx2 = sse_tx.clone();
|
|
let event_name2 = event_name.clone();
|
|
|
|
tokio::spawn(async move {
|
|
// 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 !delay.is_zero() {
|
|
tokio::time::sleep(delay).await;
|
|
}
|
|
// The burst has now been served: whatever arrives next starts a fresh debounce window.
|
|
clear_regen_burst(event_id);
|
|
// Run the body in its OWN task so a panic surfaces as a `JoinError` here rather than
|
|
// unwinding past `mark_failed` — see `supervise_export`.
|
|
let (p, m, x, s) = (
|
|
pool.clone(),
|
|
media_path.clone(),
|
|
export_path.clone(),
|
|
sse_tx.clone(),
|
|
);
|
|
let inner =
|
|
tokio::spawn(async move { run_zip_export(event_id, epoch, &p, &m, &x, &s).await });
|
|
supervise_export(inner, &pool, event_id, "zip", epoch).await;
|
|
maybe_broadcast_complete(&pool, event_id, &sse_tx).await;
|
|
});
|
|
|
|
tokio::spawn(async move {
|
|
if !delay.is_zero() {
|
|
tokio::time::sleep(delay).await;
|
|
}
|
|
let (p, m, x, s) = (
|
|
pool2.clone(),
|
|
media_path2.clone(),
|
|
export_path2.clone(),
|
|
sse_tx2.clone(),
|
|
);
|
|
let inner = tokio::spawn(async move {
|
|
run_html_export(
|
|
event_id,
|
|
epoch,
|
|
&event_name2,
|
|
comments_enabled,
|
|
&p,
|
|
&m,
|
|
&x,
|
|
&s,
|
|
)
|
|
.await
|
|
});
|
|
supervise_export(inner, &pool2, event_id, "html", epoch).await;
|
|
maybe_broadcast_complete(&pool2, event_id, &sse_tx2).await;
|
|
});
|
|
}
|
|
|
|
/// Await an export worker and make sure the job row NEVER stays at `running`.
|
|
///
|
|
/// Both workers were bare `tokio::spawn`s whose `Err` path was handled correctly — but a PANIC
|
|
/// unwound straight past `mark_failed` and `maybe_broadcast_complete`, leaving the row at whatever
|
|
/// `progress_pct` it had reached. The UI renders that as "Wird erstellt (77%)" with the download
|
|
/// disabled, permanently: nothing sweeps `running` rows, and `recover_exports` only runs at boot.
|
|
/// So the single most likely cause of a stuck keepsake was also the one the error handling missed.
|
|
///
|
|
/// Running the body as a child task turns that panic into a `JoinError` we can act on. The
|
|
/// maintenance loop already supervises itself this way; this extends the same pattern to the two
|
|
/// workers that actually produce the thing guests came for.
|
|
async fn supervise_export(
|
|
handle: tokio::task::JoinHandle<Result<()>>,
|
|
pool: &PgPool,
|
|
event_id: Uuid,
|
|
export_type: &str,
|
|
epoch: i64,
|
|
) {
|
|
match handle.await {
|
|
Ok(Ok(())) => {}
|
|
Ok(Err(e)) => {
|
|
tracing::error!(
|
|
"{export_type} export failed for event {event_id} @ epoch {epoch}: {e:#}"
|
|
);
|
|
mark_failed(pool, event_id, export_type, epoch, &e.to_string()).await;
|
|
}
|
|
Err(join_err) => {
|
|
// Panicked, or cancelled at shutdown. Either way the row must not be left claiming to
|
|
// be in progress — a failed job the host can retry beats one that lies forever.
|
|
tracing::error!(
|
|
panicked = join_err.is_panic(),
|
|
"{export_type} export task for event {event_id} @ epoch {epoch} died without \
|
|
reporting: {join_err}"
|
|
);
|
|
mark_failed(
|
|
pool,
|
|
event_id,
|
|
export_type,
|
|
epoch,
|
|
"Interner Fehler beim Erstellen des Keepsakes. Bitte über \"Neu erzeugen\" \
|
|
erneut versuchen.",
|
|
)
|
|
.await;
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Preflight that will sacrifice the previous generation rather than deadlock, and the order is
|
|
/// the whole point.
|
|
///
|
|
/// Deferring the prune (so a failed rebuild can never leave the event with no archive at all) has
|
|
/// a cost the first version of this did not follow through on: at rebuild time the previous
|
|
/// generation is still on disk and still counted against free space, so the preflight demands room
|
|
/// for BOTH. That halves the gallery size a rebuild can survive relative to the size the upload
|
|
/// gate allows — and every path that bumps the epoch (a guest deleting their own photo, a caption
|
|
/// edit, a ban, "Neu erzeugen") retires the current keepsake IMMEDIATELY on commit. So above that
|
|
/// threshold the download 404s, the rebuild is refused, and the only thing that could free the
|
|
/// space is the prune that now only runs on success. Permanently stuck, unreachable from any
|
|
/// handler.
|
|
///
|
|
/// So: try to build while preserving the old generation. If that genuinely does not fit, the old
|
|
/// generation is the one thing we can reclaim — sacrifice it and try once more. A keepsake that
|
|
/// exists beats one we preserved but can never replace.
|
|
///
|
|
/// But ONLY when the sacrifice is sufficient: we measure what pruning would free and compare it to
|
|
/// the shortfall first. Pruning and hoping meant a failed second check left nothing on disk at all
|
|
/// — no old archive and no new one — which is worse than either outcome this function chooses
|
|
/// between. When the old archive cannot buy us a rebuild, it stays.
|
|
///
|
|
/// SHARED by both halves deliberately. This started as two copies and one of them (HTML) silently
|
|
/// kept the single-phase form, so the ZIP archive rebuilt and the viewer stayed permanently stuck
|
|
/// — the exact deadlock above, on half the product. `prefix` is the only thing that differs, and
|
|
/// it must be the caller's OWN prefix: pruning the other half's archives from here would reclaim
|
|
/// space a sibling worker is about to need, on its behalf, without its knowledge.
|
|
async fn ensure_export_space_reclaiming(
|
|
pool: &PgPool,
|
|
event_id: Uuid,
|
|
export_path: &Path,
|
|
prefix: &str,
|
|
epoch: i64,
|
|
) -> Result<()> {
|
|
if ensure_export_space(pool, event_id, export_path)
|
|
.await
|
|
.is_ok()
|
|
{
|
|
return Ok(());
|
|
}
|
|
|
|
// LOOK BEFORE YOU DESTROY.
|
|
//
|
|
// This used to prune unconditionally and then re-check. When the re-check ALSO failed, the old
|
|
// keepsake was already gone and the rebuild returned `Err` — leaving **no archive on disk at
|
|
// all**, which is strictly worse than the deadlock the reclaim exists to avoid. It needs no
|
|
// host action to reach: any guest deleting their own photo arms this path, and the doc comment
|
|
// above asserts the opposite invariant.
|
|
//
|
|
// So sacrifice the old generation only when doing so is actually sufficient. When it is not,
|
|
// that old keepsake is the only one anybody will ever have — keep it, and report the failure
|
|
// the same way the non-reclaiming check does. `rebuild_export` is the host's retry once they
|
|
// have freed space.
|
|
let deficit = match export_space_deficit(pool, event_id, export_path).await? {
|
|
// Raced back into having room (a concurrent prune, a guest deleting an upload). Nothing
|
|
// to reclaim and nothing to fail.
|
|
None => return Ok(()),
|
|
Some(d) => d,
|
|
};
|
|
let reclaimable =
|
|
reclaimable_superseded_bytes(pool, export_path, prefix, event_id, epoch).await;
|
|
|
|
// PRUNE EVEN WHEN IT IS NOT ENOUGH ON ITS OWN. This used to refuse unless
|
|
// `reclaimable >= deficit`, on the premise that "deleting it would leave no archive at all" —
|
|
// and that premise does not survive contact with `prune_superseded_archives`, which can only
|
|
// ever touch generations `n < keep_seq` that `protected_files` does not name. Those are exactly
|
|
// the archives no handler can serve: a download resolves through `export_current`, which
|
|
// requires `j.epoch = e.export_epoch`, and the epoch only ever increments. The bytes this
|
|
// branch was protecting were already unreachable from every route, and the next successful
|
|
// build deletes them anyway.
|
|
//
|
|
// What the refusal did cost was the only in-app way out of the deadlock this function exists to
|
|
// break. `reclaimable` is scoped to the caller's OWN prefix — one old archive, ~1.1x the
|
|
// gallery — while `deficit` is sized for both halves plus the 10 GB reserve. So on a tight
|
|
// disk each worker independently measures its own share as insufficient and neither prunes,
|
|
// while the two shares are JOINTLY sufficient. Every "Neu erzeugen" reruns the identical
|
|
// arithmetic and refuses identically: permanently stuck, with dead archives on the volume that
|
|
// nothing will reclaim and nothing can serve.
|
|
//
|
|
// So reclaim what we can and let the re-check below decide. If it still does not fit, we fail
|
|
// exactly as the plain preflight would — but the sibling worker's prune has now freed its share
|
|
// too, and the host's retry converges instead of looping.
|
|
if reclaimable < deficit {
|
|
tracing::warn!(
|
|
deficit,
|
|
reclaimable,
|
|
"pruning the previous {prefix} keepsake will not free the full shortfall on its own; \
|
|
reclaiming anyway — it is already unservable, and the sibling half frees the rest"
|
|
);
|
|
} else {
|
|
// `else`, not a second unconditional line: both used to fire in the shortfall case, and
|
|
// they read as contradicting each other ("will not free the shortfall" / "reclaiming it
|
|
// first") to whoever is reading logs at 2am.
|
|
tracing::warn!(
|
|
deficit,
|
|
reclaimable,
|
|
"not enough room to rebuild {prefix} alongside the previous keepsake; reclaiming it first"
|
|
);
|
|
}
|
|
prune_superseded_archives(pool, export_path, prefix, event_id, epoch).await;
|
|
ensure_export_space(pool, event_id, export_path).await
|
|
}
|
|
|
|
/// Largest share of an event's media an archive may silently omit and still publish.
|
|
///
|
|
/// Not zero, deliberately. A skip is *expected* in ordinary operation — a guest deleting their own
|
|
/// photo mid-build, or the hourly reclaim collecting an original past its retention window — and
|
|
/// the writers go out of their way to degrade one entry rather than fail the whole keepsake,
|
|
/// because a released event cannot be rebuilt by the host without reopening uploads. Making any
|
|
/// skip fatal would turn a tolerated one-photo gap into total loss of the keepsake, which is the
|
|
/// regression the TOCTOU comments in each writer warn about.
|
|
///
|
|
/// What must never happen is publishing an archive that is missing *most* of the event. 10% of a
|
|
/// ~100-photo wedding is ~10 photos: far above the one-or-two a live delete explains, far below
|
|
/// the "media_path is wrong so nothing opened" catastrophe.
|
|
const MAX_SKIPPED_FRACTION: f64 = 0.10;
|
|
|
|
/// Skips tolerated regardless of how small the event is.
|
|
///
|
|
/// A pure fraction inverts this check on a small gallery: at nine uploads one skip is 11%, so the
|
|
/// single most ordinary event there is — a guest deleting their own photo while the archive builds
|
|
/// — failed the whole keepsake. That is exactly the "one-photo gap becomes total loss" outcome
|
|
/// [`MAX_SKIPPED_FRACTION`]'s own comment says it exists to avoid, and it is worst early on, when
|
|
/// a host testing a release has only a handful of photos.
|
|
///
|
|
/// Two is the number of concurrent live deletes worth absorbing; beyond that on a tiny gallery the
|
|
/// `written == 0` guard still catches the misconfiguration case, which is the one that matters.
|
|
const MIN_TOLERATED_SKIPS: usize = 2;
|
|
|
|
/// Decide whether an archive that skipped some media may still be published.
|
|
///
|
|
/// Rejects two cases:
|
|
/// * `written == 0` with media expected — the misconfiguration case. This is what produced a
|
|
/// few-hundred-byte ZIP containing zero photos that passed every automated check, was
|
|
/// advertised green, and was handed to every guest.
|
|
/// * more than [`MAX_SKIPPED_FRACTION`] omitted — enough missing that the archive misrepresents
|
|
/// the event even though it is structurally valid.
|
|
///
|
|
/// A tolerated partial still logs at `error` level with the exact counts, so the gap is visible in
|
|
/// the record rather than buried in per-entry warnings nobody aggregates.
|
|
fn check_export_completeness(
|
|
kind: &str,
|
|
event_id: Uuid,
|
|
expected: usize,
|
|
written: usize,
|
|
skipped: usize,
|
|
) -> Result<()> {
|
|
if skipped == 0 {
|
|
return Ok(());
|
|
}
|
|
if written == 0 && expected > 0 {
|
|
anyhow::bail!(
|
|
"{kind}-Export enthält keine einzige Datei ({expected} erwartet, alle \
|
|
übersprungen). Das deutet auf einen falschen MEDIA_PATH oder fehlende \
|
|
Zugriffsrechte hin — das Keepsake wurde NICHT veröffentlicht."
|
|
);
|
|
}
|
|
// A PROPORTIONAL skip is loud, but it is NOT fatal, and the difference is the whole point.
|
|
//
|
|
// This used to `bail!` once skips passed `max(2, 10% of expected)`. That inverts
|
|
// MAX_SKIPPED_FRACTION's own justification two doc comments up — "making any skip fatal would
|
|
// turn a tolerated one-photo gap into total loss of the keepsake" — because the refusal is
|
|
// DETERMINISTIC ACROSS RETRIES. The files that could not be read are still unreadable when the
|
|
// host taps "Neu erzeugen", and by then the gallery is released and the uploads cannot be
|
|
// collected again. So on a 30-photo event, 4 unreadable originals stopped publishing the other
|
|
// 26 — not once, but forever. Refusing to hand over an incomplete keepsake is only defensible
|
|
// if something better can still arrive; here nothing can.
|
|
//
|
|
// `written == 0` above stays fatal, and it is the case that actually mattered: a wrong
|
|
// MEDIA_PATH produced a few-hundred-byte archive with zero photos that passed every automated
|
|
// check and was advertised as ready. That one is a misconfiguration the host CAN fix and retry.
|
|
//
|
|
// Everything short of that publishes, with the counts at `error` level so the gap is in the
|
|
// record rather than buried in per-entry warnings. A partial keepsake is what the guests get to
|
|
// keep; an aborted one is nothing at all.
|
|
let budget = MIN_TOLERATED_SKIPS.max((expected as f64 * MAX_SKIPPED_FRACTION).floor() as usize);
|
|
let level_note = if expected > 0 && skipped > budget {
|
|
"MATERIALLY INCOMPLETE — published anyway because a refused rebuild is not recoverable"
|
|
} else {
|
|
"published with missing media"
|
|
};
|
|
tracing::error!(
|
|
%event_id, kind, expected, written, skipped, budget,
|
|
"{level_note} — {skipped} of {expected} entries could not be read"
|
|
);
|
|
Ok(())
|
|
}
|
|
|
|
/// Take the process-wide heavy-image permit if this file is big enough to need it.
|
|
///
|
|
/// Mirrors the compression worker's gate exactly (a header probe, no pixels decoded), so the two
|
|
/// producers of heavy image work agree on what "heavy" means and serialise against each other
|
|
/// rather than each against itself.
|
|
async fn heavy_permit_for(path: &Path) -> Option<tokio::sync::SemaphorePermit<'static>> {
|
|
let estimate = crate::services::imaging::estimated_processing_peak_bytes(path, 2048);
|
|
match estimate {
|
|
Some(bytes) if bytes > crate::services::imaging::HEAVY_IMAGE_BYTES => {
|
|
crate::services::imaging::HEAVY_IMAGE_PERMITS
|
|
.acquire()
|
|
.await
|
|
.ok()
|
|
}
|
|
_ => None,
|
|
}
|
|
}
|
|
|
|
// ── ZIP export ───────────────────────────────────────────────────────────────
|
|
|
|
async fn run_zip_export(
|
|
event_id: Uuid,
|
|
epoch: i64,
|
|
pool: &PgPool,
|
|
media_path: &Path,
|
|
export_path: &Path,
|
|
sse_tx: &broadcast::Sender<SseEvent>,
|
|
) -> Result<()> {
|
|
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(());
|
|
}
|
|
|
|
// AFTER the claim, not before. A preflight that bailed before claiming would leave the row
|
|
// `pending` with no worker and no error — the spinner-forever state `mark_failed`'s status
|
|
// guard was widened to prevent. Failing here goes through the caller's `mark_failed`, so the
|
|
// host gets the reason.
|
|
ensure_export_space_reclaiming(pool, event_id, export_path, "Gallery", epoch).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;
|
|
}
|
|
|
|
// Reclaim the PREVIOUS generation only once this one has actually landed.
|
|
//
|
|
// This used to run before the preflight, reasoning that the superseded archive is already
|
|
// unreachable and its space is usually exactly what the rebuild needs. That is true about
|
|
// REACHABILITY and false about RECOVERABILITY: an epoch is a database value that can be
|
|
// rolled back, deleted bytes cannot. Any rebuild that then failed — ENOSPC mid-write, an
|
|
// OOM, a hung ffmpeg, a host tapping "Neu erzeugen" on a bad day — left the event with NO
|
|
// archive at all, which is the one outcome the whole product exists to prevent, at the one
|
|
// moment nobody is watching.
|
|
//
|
|
// The cost of deferring is that a rebuild now needs room for both generations at once, and
|
|
// `ensure_export_space` above no longer gets to count the old archive's bytes as available.
|
|
// That is the correct trade: it converts "silently destroyed the only copy" into "refused
|
|
// to start, and said why".
|
|
if res.is_ok() {
|
|
prune_superseded_archives(pool, export_path, "Gallery", event_id, epoch).await;
|
|
}
|
|
|
|
abandon_if_superseded("ZIP", event_id, epoch, res)
|
|
}
|
|
|
|
/// A worker that discovers mid-export that its generation was retired has not FAILED — it simply
|
|
/// lost. Swallow the sentinel so we don't log an error or (pointlessly) try to mark a row we no
|
|
/// longer own as failed. Its temp artifacts were already removed by the caller.
|
|
fn abandon_if_superseded(kind: &str, event_id: Uuid, epoch: i64, res: Result<()>) -> Result<()> {
|
|
match res {
|
|
Err(e) if e.downcast_ref::<Superseded>().is_some() => {
|
|
tracing::info!(
|
|
"{kind} export for event {event_id} superseded mid-export (epoch {epoch} retired); abandoned"
|
|
);
|
|
Ok(())
|
|
}
|
|
other => other,
|
|
}
|
|
}
|
|
|
|
/// 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(
|
|
epoch: 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;
|
|
|
|
// 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?;
|
|
|
|
// 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);
|
|
|
|
// Skips are TOLERATED but no longer SILENT — see `check_export_completeness`. All three
|
|
// writers continued past an unreadable source with only a `warn!`, nothing counted them,
|
|
// nothing reached `error_message`, and the job finalized at `progress_pct = 100,
|
|
// status = 'done'`. So a wrong `media_path` after a compose edit produced a few-hundred-byte
|
|
// ZIP with zero media that passed the liveness check, was advertised green, and was handed to
|
|
// every guest as their keepsake.
|
|
let mut written = 0usize;
|
|
let mut skipped = 0usize;
|
|
|
|
{
|
|
let file = tokio::fs::File::create(&tmp_path).await?;
|
|
let mut zip = ZipFileWriter::with_tokio(file);
|
|
|
|
for (i, row) in uploads.iter().enumerate() {
|
|
let src = media_path.join(&row.original_path);
|
|
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);
|
|
let folder = if row.mime_type.starts_with("video/") {
|
|
"Videos"
|
|
} else {
|
|
"Photos"
|
|
};
|
|
let entry_name = format!("{folder}/{date}_{name_safe}_{}.{ext}", row.id);
|
|
|
|
let builder = keepsake_entry(entry_name, Compression::Stored);
|
|
|
|
// 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()
|
|
);
|
|
skipped += 1;
|
|
continue;
|
|
}
|
|
};
|
|
written += 1;
|
|
|
|
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;
|
|
// Also our liveness check: if we've been retired, stop NOW rather than grinding through
|
|
// the rest of the gallery to build an archive we would immediately delete.
|
|
if !update_progress(pool, event_id, "zip", epoch, pct.min(99)).await {
|
|
return Err(Superseded.into());
|
|
}
|
|
}
|
|
|
|
// 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?;
|
|
}
|
|
|
|
// Refuse to publish an archive that lost media it was supposed to contain. Checked BEFORE the
|
|
// rename so a rejected build never reaches a servable path.
|
|
check_export_completeness("ZIP", event_id, uploads.len(), written, skipped)
|
|
.inspect_err(|_| tracing::error!("ZIP export for event {event_id}: refusing to publish"))?;
|
|
|
|
tokio::fs::rename(&tmp_path, &out_path).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 (epoch {epoch} retired); discarded"
|
|
);
|
|
// Err, NOT Ok. `abandon_if_superseded` swallows this sentinel for the caller, so the
|
|
// outcome is unchanged — but `run_*_export`'s deferred prune keys off `res.is_ok()`,
|
|
// and a worker that LOST the epoch race reporting success made it reclaim generations
|
|
// older than its own RETIRED epoch. At that moment the winning generation is still
|
|
// `pending` with no file, so nothing protected the last good archive and it was deleted
|
|
// with no replacement on disk — the precise outcome deferring the prune exists to stop.
|
|
return Err(Superseded.into());
|
|
}
|
|
|
|
prune_stale_export_files(pool, &exports_dir, "Gallery", event_id, epoch).await;
|
|
|
|
let _ = sse_tx.send(SseEvent {
|
|
event_type: "export-progress".to_string(),
|
|
data: serde_json::json!({ "type": "zip", "progress_pct": 100 }).to_string(),
|
|
});
|
|
|
|
tracing::info!("ZIP export complete for event {event_id}");
|
|
Ok(())
|
|
}
|
|
|
|
// ── HTML viewer export ──────────────────────────────────────────────────────
|
|
|
|
/// Where a media entry's bytes come from at ZIP-writing time. Derived variants
|
|
/// (thumbnails, compressed images) are staged under the temp dir; original-fidelity
|
|
/// variants (videos, small images) are streamed straight from the source on disk so
|
|
/// the export never transiently doubles disk usage by copying large files to temp.
|
|
enum MediaSource {
|
|
Temp(PathBuf),
|
|
Original(PathBuf),
|
|
}
|
|
|
|
impl MediaSource {
|
|
fn path(&self) -> &Path {
|
|
match self {
|
|
MediaSource::Temp(p) | MediaSource::Original(p) => p,
|
|
}
|
|
}
|
|
}
|
|
|
|
#[allow(clippy::too_many_arguments)]
|
|
async fn run_html_export(
|
|
event_id: Uuid,
|
|
epoch: i64,
|
|
event_name: &str,
|
|
comments_enabled: bool,
|
|
pool: &PgPool,
|
|
media_path: &Path,
|
|
export_path: &Path,
|
|
sse_tx: &broadcast::Sender<SseEvent>,
|
|
) -> Result<()> {
|
|
if !claim_job(pool, event_id, "html", epoch).await? {
|
|
// Another worker owns this generation, or our epoch has been retired. Not ours.
|
|
return Ok(());
|
|
}
|
|
|
|
// See run_zip_export: refuse at the door rather than ENOSPC mid-write, and reclaim the
|
|
// superseded generation only AFTER this one lands — unless it is the only way to land at all.
|
|
ensure_export_space_reclaiming(pool, event_id, export_path, "Memories", epoch).await?;
|
|
|
|
let res = run_html_export_inner(
|
|
epoch,
|
|
event_id,
|
|
event_name,
|
|
comments_enabled,
|
|
pool,
|
|
media_path,
|
|
export_path,
|
|
sse_tx,
|
|
)
|
|
.await;
|
|
if res.is_err() {
|
|
// Clean up this generation's temp artifacts so a failing (or abandoned) 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;
|
|
}
|
|
|
|
// Only once the new keepsake exists — see the reasoning in run_zip_export. A failed rebuild
|
|
// must never be the reason the previous one is gone.
|
|
if res.is_ok() {
|
|
prune_superseded_archives(pool, export_path, "Memories", event_id, epoch).await;
|
|
}
|
|
|
|
abandon_if_superseded("HTML", event_id, epoch, res)
|
|
}
|
|
|
|
#[allow(clippy::too_many_arguments)]
|
|
async fn run_html_export_inner(
|
|
epoch: i64,
|
|
event_id: Uuid,
|
|
event_name: &str,
|
|
comments_enabled: bool,
|
|
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;
|
|
|
|
let _ = 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}_{epoch}"));
|
|
let media_tmp = tmp_dir.join("media");
|
|
tokio::fs::create_dir_all(&media_tmp).await?;
|
|
|
|
// 3. Process media and build post data
|
|
let mut viewer_posts: Vec<ViewerPost> = Vec::new();
|
|
// (zip entry name under media/, where its bytes come from). Built here, streamed
|
|
// into the ZIP in step 5 — so we also know the exact file count without a rescan.
|
|
// The bool is "this entry is the FULL variant", i.e. the one `data.json` advertises as the
|
|
// photo itself. It exists so `check_export_completeness` can count photos rather than files —
|
|
// see the call site. Exactly one full entry is pushed per upload that survives the stat.
|
|
let mut media_manifest: Vec<(String, MediaSource, bool)> = Vec::new();
|
|
// Uploads dropped at the stat below never enter `media_manifest`, so without counting them
|
|
// here a wholly-unreadable media directory yields an EMPTY manifest — expected 0, skipped 0 —
|
|
// and the completeness check downstream would wave it through as a legitimately empty event.
|
|
// These are the same loss as a skip at write time and are checked as one.
|
|
let mut upload_skipped = 0usize;
|
|
|
|
for (i, row) in uploads.iter().enumerate() {
|
|
let src = media_path.join(&row.original_path);
|
|
// Stat ONCE, up front, and skip this upload if the source is gone. The old code probed with
|
|
// `exists()` here and then did `metadata(&src).await?` further down — a TOCTOU whose `?`
|
|
// aborted the ENTIRE keepsake if the file vanished in between. It can still happen: the
|
|
// compression worker no longer deletes originals on failure, but the hourly sweep reclaims
|
|
// them once past the retention window, and an owner or host delete can land mid-export. A
|
|
// missing source must degrade one entry, never the whole archive (which, once released, the
|
|
// host cannot rebuild without reopening uploads).
|
|
let src_meta = match tokio::fs::metadata(&src).await {
|
|
Ok(m) => m,
|
|
Err(e) => {
|
|
tracing::warn!(
|
|
"HTML export: skipping upload {} — cannot stat {}: {e}",
|
|
row.id,
|
|
src.display()
|
|
);
|
|
upload_skipped += 1;
|
|
continue;
|
|
}
|
|
};
|
|
|
|
let is_video = row.mime_type.starts_with("video/");
|
|
let id_str = row.id.to_string();
|
|
|
|
// Generate thumbnail and full variant. `full_source` says where the full-res
|
|
// bytes come from at ZIP time — for videos and small images that's the original
|
|
// on disk (streamed directly, never copied to temp).
|
|
let (thumb_name, full_name, full_source) = if is_video {
|
|
let thumb = format!("{id_str}_thumb.jpg");
|
|
let full_ext = ext_from_path(&row.original_path);
|
|
let full = format!("{id_str}.{full_ext}");
|
|
|
|
// Poster frame via the shared helper, which owns the seek order, the 120s timeout
|
|
// (this call site had NONE — a hung ffmpeg would strand the export at `running`
|
|
// forever) and the artifact check.
|
|
let thumb_path = media_tmp.join(&thumb);
|
|
let produced =
|
|
match crate::services::video::extract_poster_frame(&src, &thumb_path, 400).await {
|
|
Ok(produced) => produced,
|
|
Err(e) => {
|
|
tracing::warn!("poster extraction errored for upload {}: {e:#}", row.id);
|
|
false
|
|
}
|
|
};
|
|
if !produced {
|
|
tracing::info!(
|
|
upload_id = %row.id,
|
|
"no poster frame for this video; exporting it without one"
|
|
);
|
|
}
|
|
|
|
// Stream the video full-res straight from the original at ZIP time — no
|
|
// copy to temp (that used to transiently double disk usage per video).
|
|
(
|
|
produced.then(|| thumb.clone()),
|
|
full,
|
|
MediaSource::Original(src.clone()),
|
|
)
|
|
} else {
|
|
let thumb = format!("{id_str}_thumb.jpg");
|
|
let ext = ext_from_path(&row.original_path);
|
|
let full = format!("{id_str}_full.{ext}");
|
|
|
|
// Image thumbnail: resize to 400px wide
|
|
let src_clone = src.clone();
|
|
let thumb_path = media_tmp.join(&thumb);
|
|
let thumb_path_clone = thumb_path.clone();
|
|
|
|
// Same process-wide memory permit the compression worker takes. Without it, a
|
|
// release fired while the last phone photos were still compressing put an export
|
|
// decode and a heavy compression job in the same 1 GiB cgroup — and the OOM kill
|
|
// marks the export failed, which `recover_exports` then re-spawns into the same
|
|
// conditions on the next boot.
|
|
let _heavy = heavy_permit_for(&src).await;
|
|
let thumb_result = tokio::task::spawn_blocking(move || -> Result<()> {
|
|
// `decode_oriented`, not `image::open`: the latter ignores the EXIF
|
|
// orientation tag AND applies no decode limits. Using it here is why every
|
|
// portrait photo came out sideways in the keepsake's HTML viewer grid — the
|
|
// re-encode below drops the tag, so the viewer cannot recover it.
|
|
let img = crate::services::imaging::decode_oriented(&src_clone)
|
|
.context("failed to open image for thumbnail")?;
|
|
let resized = img.resize(400, 400, image::imageops::FilterType::Lanczos3);
|
|
resized
|
|
.save_with_format(&thumb_path_clone, image::ImageFormat::Jpeg)
|
|
.context("failed to save thumbnail")?;
|
|
Ok(())
|
|
})
|
|
.await
|
|
// NOT `?`. The `?` here was on the JoinError, not on the closure's Result — so a
|
|
// decoder PANIC (the `image` crate can panic on malformed input, and a resize can
|
|
// abort on allocation) propagated out and failed the ENTIRE keepsake, where the very
|
|
// same file merely failing returns `Err` and costs one tile. Worse, it was
|
|
// deterministic: "Neu erzeugen" reads the same poison file and dies the same way. That
|
|
// is the failure shape the completeness guard was reversed to eliminate, arriving
|
|
// through the other door.
|
|
.unwrap_or_else(|e| Err(anyhow::anyhow!("thumbnail task panicked: {e}")));
|
|
|
|
// Same dangling-reference hazard as the video branch: a failure here left `thumb`
|
|
// pointing at a file the ZIP writer would then skip, so `data.json` advertised an
|
|
// entry the archive didn't contain. An undecodable image is rarer than a sub-second
|
|
// clip, but the broken tile is identical.
|
|
let thumb_ok = match thumb_result {
|
|
Ok(()) => true,
|
|
Err(e) => {
|
|
tracing::warn!("thumbnail generation failed for upload {}: {e:#}", row.id);
|
|
false
|
|
}
|
|
};
|
|
|
|
// Full variant: compress to temp if >5MB, otherwise stream the original
|
|
// as-is (no temp copy). `src_meta` was stat'd once at the top of the loop.
|
|
let full_source = if src_meta.len() > 5_000_000 {
|
|
let src_clone = src.clone();
|
|
let full_path = media_tmp.join(&full);
|
|
let full_path_clone = full_path.clone();
|
|
|
|
// See the thumbnail above. This branch is the more expensive of the two: it
|
|
// only runs for originals over 5 MB, i.e. exactly the giants.
|
|
let _heavy = heavy_permit_for(&src).await;
|
|
let compress_result = tokio::task::spawn_blocking(move || -> Result<()> {
|
|
// Same reason as the thumbnail above. This branch only runs for originals
|
|
// over 5 MB, which is why the viewer's full image looked correct for small
|
|
// photos and sideways for large ones — an inconsistency that reads as a
|
|
// viewer bug rather than an export one.
|
|
let img = crate::services::imaging::decode_oriented(&src_clone)
|
|
.context("failed to open image for compression")?;
|
|
let resized = img.resize(2000, 2000, image::imageops::FilterType::Lanczos3);
|
|
resized
|
|
.save_with_format(&full_path_clone, image::ImageFormat::Jpeg)
|
|
.context("failed to save compressed full image")?;
|
|
Ok(())
|
|
})
|
|
.await
|
|
// See the thumbnail branch: a panic here must cost this one full variant (the
|
|
// original is then streamed as-is below), never the whole keepsake.
|
|
.unwrap_or_else(|e| Err(anyhow::anyhow!("full-image task panicked: {e}")));
|
|
|
|
match compress_result {
|
|
Ok(()) => MediaSource::Temp(full_path),
|
|
Err(e) => {
|
|
tracing::warn!(
|
|
"compression failed for upload {}, using original: {e:#}",
|
|
row.id
|
|
);
|
|
MediaSource::Original(src.clone())
|
|
}
|
|
}
|
|
} else {
|
|
MediaSource::Original(src.clone())
|
|
};
|
|
|
|
(thumb_ok.then_some(thumb), full, full_source)
|
|
};
|
|
|
|
// Register this post's media entries. The thumbnail is registered ONLY when one was
|
|
// actually produced: pushing a manifest entry for a file that doesn't exist made the ZIP
|
|
// writer skip it silently while `data.json` still advertised it — the viewer then drew a
|
|
// broken image tile for an entry the archive never contained.
|
|
if let Some(name) = &thumb_name {
|
|
media_manifest.push((name.clone(), MediaSource::Temp(media_tmp.join(name)), false));
|
|
}
|
|
media_manifest.push((full_name.clone(), full_source, true));
|
|
|
|
// Build comments for this upload
|
|
let post_comments: Vec<ViewerComment> = comments
|
|
.iter()
|
|
.filter(|c| c.upload_id == row.id)
|
|
.map(|c| ViewerComment {
|
|
author: c.uploader_name.clone(),
|
|
text: c.body.clone(),
|
|
timestamp: c.created_at.to_rfc3339(),
|
|
})
|
|
.collect();
|
|
|
|
// Build tags for this upload
|
|
let tags: Vec<String> = hashtags_per_upload
|
|
.iter()
|
|
.filter(|(uid, _)| *uid == row.id)
|
|
.map(|(_, tag)| tag.clone())
|
|
.collect();
|
|
|
|
viewer_posts.push(ViewerPost {
|
|
id: id_str,
|
|
uploader: row.uploader_name.clone(),
|
|
caption: row.caption.clone().unwrap_or_default(),
|
|
tags,
|
|
timestamp: row.created_at.to_rfc3339(),
|
|
likes: row.like_count,
|
|
comments: post_comments,
|
|
media: ViewerMedia {
|
|
media_type: if is_video {
|
|
"video".to_string()
|
|
} else {
|
|
"image".to_string()
|
|
},
|
|
// Empty when there is no poster. The viewer already guards on this
|
|
// (`{#if post.media.thumb}` → a video tile with a play glyph, or the placeholder
|
|
// icon for an image), so telling it the truth is the entire fix — no schema
|
|
// change, no viewer rebuild. What was broken was the backend always claiming a
|
|
// thumbnail existed.
|
|
thumb: thumb_name
|
|
.as_ref()
|
|
.map(|n| format!("media/{n}"))
|
|
.unwrap_or_default(),
|
|
full: format!("media/{full_name}"),
|
|
},
|
|
});
|
|
|
|
let pct = 10 + ((i + 1) as f32 / total * 60.0) as i16;
|
|
if !update_progress(pool, event_id, "html", epoch, pct.min(69)).await {
|
|
return Err(Superseded.into());
|
|
}
|
|
}
|
|
|
|
// 4. Build data.json
|
|
let viewer_data = ViewerData {
|
|
event: ViewerEvent {
|
|
name: event_name.to_string(),
|
|
exported_at: Utc::now().to_rfc3339(),
|
|
comments_enabled,
|
|
},
|
|
posts: viewer_posts,
|
|
};
|
|
let data_json =
|
|
serde_json::to_string_pretty(&viewer_data).context("failed to serialize data.json")?;
|
|
|
|
// Match the live app's colour theme in the offline keepsake.
|
|
let (theme_primary, theme_accent) = resolve_theme_seeds(pool).await;
|
|
let theme_css = theme_override_css(&theme_primary, &theme_accent);
|
|
|
|
let _ = 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(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);
|
|
|
|
{
|
|
let file = tokio::fs::File::create(&tmp_path).await?;
|
|
let mut zip = ZipFileWriter::with_tokio(file);
|
|
|
|
// Write the embedded single-file viewer, injecting the export data as a
|
|
// `window.__EXPORT_DATA__` global into index.html. Guests double-click
|
|
// index.html (file://), where a cross-origin fetch() of a sibling file is
|
|
// blocked — so the data must be inlined rather than fetched from data.json.
|
|
// The viewer IS the keepsake — Memories.zip without it is a folder of files with no way to
|
|
// look at them. `write_viewer_with_data` walks `dir.files()`, which iterates nothing at all
|
|
// when the compiled-in directory is empty, so a viewer build that failed after Vite emptied
|
|
// its output directory used to produce a perfectly valid archive with no viewer in it,
|
|
// silently. Asserted here rather than trusted: this costs one lookup per export.
|
|
if VIEWER_DIR.get_file("index.html").is_none() {
|
|
anyhow::bail!(
|
|
"the keepsake viewer is missing from this binary (static/export-viewer/index.html \
|
|
was not compiled in). Run `npm run build` in frontend/export-viewer and rebuild — \
|
|
an archive without the viewer is not a keepsake."
|
|
);
|
|
}
|
|
write_viewer_with_data(&VIEWER_DIR, &mut zip, &data_json, theme_css.as_deref()).await?;
|
|
|
|
let _ = update_progress(pool, event_id, "html", epoch, 75).await;
|
|
|
|
// Write data.json
|
|
{
|
|
let builder = keepsake_entry("data.json".into(), Compression::Deflate);
|
|
let mut entry = zip.write_entry_stream(builder).await?;
|
|
let mut cursor = AllowStdIo::new(std::io::Cursor::new(data_json.as_bytes()));
|
|
fcopy(&mut cursor, &mut entry).await?;
|
|
entry.close().await?;
|
|
}
|
|
|
|
// Write README.txt
|
|
{
|
|
let builder = keepsake_entry("README.txt".into(), Compression::Deflate);
|
|
let mut entry = zip.write_entry_stream(builder).await?;
|
|
let mut cursor = AllowStdIo::new(std::io::Cursor::new(README_TEXT.as_bytes()));
|
|
fcopy(&mut cursor, &mut entry).await?;
|
|
entry.close().await?;
|
|
}
|
|
|
|
let _ = 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
|
|
// straight from the original on disk. Sources that don't exist (e.g. a thumb
|
|
// whose ffmpeg step failed) are skipped — the viewer tolerates gaps.
|
|
let file_total = media_manifest.len().max(1) as f32;
|
|
let mut files_written = 0u32;
|
|
// Photos, not files — the number `check_export_completeness` is actually about.
|
|
//
|
|
// `files_written` counts MANIFEST ROWS, and there are up to two per upload: a thumbnail and
|
|
// a full variant. Thumbnails are 400px JPEGs this export GENERATES ITSELF into its own temp
|
|
// dir, so they are no evidence that any original was captured. Counting them meant the one
|
|
// remaining fatal case — nothing at all was written — could not fire while thumbnails kept
|
|
// succeeding: if the media volume became unreadable after the stat pass, every original
|
|
// open failed and every thumb open succeeded, and a keepsake with 100 thumbnails and ZERO
|
|
// full-resolution photos published green, `done` at the live epoch, with the download
|
|
// button lit. Boot recovery skips a `done` job, so nothing would ever have rebuilt it.
|
|
let mut full_written = 0usize;
|
|
// See `check_export_completeness`: a viewer that tolerates gaps must still not publish an
|
|
// archive with no media in it at all.
|
|
let mut media_skipped = 0usize;
|
|
|
|
for (name, source, is_full) in &media_manifest {
|
|
let path = source.path();
|
|
// Open-first: a source that disappeared between the manifest being built and now (a
|
|
// delete, or the hourly sweep reclaiming a long-failed original) must skip this entry,
|
|
// not fail the whole viewer. Opening collapses the check and the use into one operation.
|
|
let src_file = match tokio::fs::File::open(path).await {
|
|
Ok(f) => f,
|
|
Err(e) => {
|
|
tracing::warn!(
|
|
"HTML export: skipping media {name} — cannot read {}: {e}",
|
|
path.display()
|
|
);
|
|
media_skipped += 1;
|
|
continue;
|
|
}
|
|
};
|
|
|
|
let entry_name = format!("media/{name}");
|
|
let builder = keepsake_entry(entry_name, Compression::Stored);
|
|
let mut zip_entry = zip.write_entry_stream(builder).await?;
|
|
let mut f = src_file.compat();
|
|
fcopy(&mut f, &mut zip_entry).await?;
|
|
zip_entry.close().await?;
|
|
|
|
files_written += 1;
|
|
if *is_full {
|
|
full_written += 1;
|
|
}
|
|
let pct = 78 + (files_written as f32 / file_total * 20.0) as i16;
|
|
if !update_progress(pool, event_id, "html", epoch, pct.min(98)).await {
|
|
return Err(Superseded.into());
|
|
}
|
|
}
|
|
|
|
// 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?;
|
|
|
|
// Before the rename, so a rejected viewer never reaches a servable path. Inside this
|
|
// scope because `media_manifest` and the counters are scoped here.
|
|
// Measured against the full upload set, not the manifest: an upload dropped at the stat
|
|
// and one dropped at the write are the same loss to the guest looking for their photo.
|
|
check_export_completeness(
|
|
"HTML",
|
|
event_id,
|
|
uploads.len(),
|
|
full_written,
|
|
upload_skipped + media_skipped,
|
|
)
|
|
.inspect_err(|_| {
|
|
tracing::error!("HTML export for event {event_id}: refusing to publish")
|
|
})?;
|
|
}
|
|
|
|
// 6. Finalise
|
|
tokio::fs::rename(&tmp_path, &out_path).await?;
|
|
|
|
// Clean up temp directory
|
|
let _ = tokio::fs::remove_dir_all(&tmp_dir).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 (epoch {epoch} retired); discarded"
|
|
);
|
|
// Err, NOT Ok. `abandon_if_superseded` swallows this sentinel for the caller, so the
|
|
// outcome is unchanged — but `run_*_export`'s deferred prune keys off `res.is_ok()`,
|
|
// and a worker that LOST the epoch race reporting success made it reclaim generations
|
|
// older than its own RETIRED epoch. At that moment the winning generation is still
|
|
// `pending` with no file, so nothing protected the last good archive and it was deleted
|
|
// with no replacement on disk — the precise outcome deferring the prune exists to stop.
|
|
return Err(Superseded.into());
|
|
}
|
|
|
|
prune_stale_export_files(pool, &exports_dir, "Memories", event_id, epoch).await;
|
|
|
|
let _ = sse_tx.send(SseEvent {
|
|
event_type: "export-progress".to_string(),
|
|
data: serde_json::json!({ "type": "html", "progress_pct": 100 }).to_string(),
|
|
});
|
|
|
|
tracing::info!("HTML viewer export complete for event {event_id}");
|
|
Ok(())
|
|
}
|
|
|
|
// ── DB helpers ───────────────────────────────────────────────────────────────
|
|
|
|
async fn query_uploads(pool: &PgPool, event_id: Uuid) -> Result<Vec<ExportUploadRow>> {
|
|
Ok(sqlx::query_as::<_, ExportUploadRow>(concat!(
|
|
"SELECT u.id, u.original_path, u.mime_type, u.caption,
|
|
usr.display_name AS uploader_name,
|
|
COUNT(DISTINCT l.user_id) AS like_count,
|
|
u.created_at
|
|
FROM upload u
|
|
JOIN \"user\" usr ON usr.id = u.user_id
|
|
LEFT JOIN \"like\" l ON l.upload_id = u.id
|
|
",
|
|
export_visibility_where!(),
|
|
"
|
|
GROUP BY u.id, usr.display_name
|
|
ORDER BY u.created_at ASC",
|
|
))
|
|
.bind(event_id)
|
|
.fetch_all(pool)
|
|
.await?)
|
|
}
|
|
|
|
async fn query_comments(pool: &PgPool, event_id: Uuid) -> Result<Vec<ExportCommentRow>> {
|
|
Ok(sqlx::query_as::<_, ExportCommentRow>(
|
|
"SELECT c.upload_id, usr.display_name AS uploader_name, c.body, c.created_at
|
|
FROM comment c
|
|
JOIN \"user\" usr ON usr.id = c.user_id
|
|
JOIN upload u ON u.id = c.upload_id
|
|
WHERE u.event_id = $1 AND c.deleted_at IS NULL AND u.deleted_at IS NULL
|
|
AND usr.uploads_hidden = FALSE AND usr.is_banned = FALSE
|
|
ORDER BY c.created_at ASC",
|
|
)
|
|
.bind(event_id)
|
|
.fetch_all(pool)
|
|
.await?)
|
|
}
|
|
|
|
async fn query_hashtags(pool: &PgPool, event_id: Uuid) -> Result<Vec<(Uuid, String)>> {
|
|
let rows: Vec<(Uuid, String)> = sqlx::query_as(
|
|
"SELECT uh.upload_id, h.tag
|
|
FROM upload_hashtag uh
|
|
JOIN hashtag h ON h.id = uh.hashtag_id
|
|
JOIN upload u ON u.id = uh.upload_id
|
|
WHERE h.event_id = $1 AND u.deleted_at IS NULL",
|
|
)
|
|
.bind(event_id)
|
|
.fetch_all(pool)
|
|
.await?;
|
|
Ok(rows)
|
|
}
|
|
|
|
/// Claim the pending job for `(event, type)` AT OUR EPOCH. `true` only if we won it.
|
|
///
|
|
/// 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.
|
|
///
|
|
/// NOTE what this does and does NOT guarantee. It compares our birth epoch to the JOB ROW's epoch —
|
|
/// not to `event.export_epoch`. `open_event` bumps the event's epoch and writes nothing to
|
|
/// `export_job` (that is the point of the design: one write retires everything). So after a reopen
|
|
/// the row is still `pending` at our epoch and this claim SUCCEEDS: the worker will build an archive
|
|
/// nobody can ever see, because retirement is enforced at READ time (`export_current` requires
|
|
/// `j.epoch = e.export_epoch`), not at write time. That is wasted work, not incorrectness. Do not
|
|
/// "optimise" this into a cross-table check: that is exactly the unsound guard we removed.
|
|
///
|
|
/// This used to claim that "the `update_progress` liveness check bails such a worker out early".
|
|
/// IT DOES NOT, and it cannot: `update_progress`'s predicate is `epoch = ours AND status =
|
|
/// 'running'` on the JOB ROW, which a reopen does not touch — so the check returns true on every
|
|
/// tick and the worker grinds the whole gallery to completion, every ffmpeg poster and every
|
|
/// Lanczos3 resize, before its `finalize_job` writes `done` at an epoch nothing reads.
|
|
///
|
|
/// The cost is real on a 2-vCPU box: a host reopening the event mid-export — the documented
|
|
/// "oops, one more photo" path — leaves a full export burning CPU and the heavy-image semaphore
|
|
/// DURING the live event, and lands a full-gallery-sized orphan that nothing reclaims until the
|
|
/// next successful build at a higher epoch. Bounded and not corrupting, so it is left as is; but
|
|
/// the mitigation the old comment promised was never there, and anyone sizing this box should
|
|
/// know that.
|
|
///
|
|
/// 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 epoch = $3 AND status = 'pending'",
|
|
)
|
|
.bind(event_id)
|
|
.bind(export_type)
|
|
.bind(epoch)
|
|
.execute(pool)
|
|
.await
|
|
.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 (`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.
|
|
///
|
|
/// ERRORS ARE DISTINGUISHED FROM A LOST RACE, for the same reason `claim_job` distinguishes them —
|
|
/// and the consequence here is strictly worse. This used to end `.unwrap_or(false)`, collapsing a
|
|
/// pool timeout into "we were superseded". At that point the archive is already built, fsynced and
|
|
/// renamed into place, so the caller went on to DELETE the finished multi-GB file and return the
|
|
/// `Superseded` sentinel — which `abandon_if_superseded` swallows into `Ok(())`, so
|
|
/// `spawn_export_jobs` never called `mark_failed` either. The row stayed `running` at 99% at the
|
|
/// LIVE epoch, which the host dashboard renders as "Wird erstellt (99 %)" with the download
|
|
/// disabled, forever: no sweep re-examines `running` rows, and `recover_exports` runs only at boot.
|
|
///
|
|
/// A pool timeout is not exotic here. `max_connections` is 10, `acquire_timeout` 5s, and this fires
|
|
/// at the end of a full-gallery export while ~100 guests are uploading. Returning `Err` instead
|
|
/// leaves the finished archive on disk and lets the caller's `mark_failed` record a real reason.
|
|
async fn finalize_job(
|
|
pool: &PgPool,
|
|
event_id: Uuid,
|
|
export_type: &str,
|
|
epoch: i64,
|
|
file_path: &str,
|
|
) -> Result<bool> {
|
|
let r = 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 epoch = $4 AND status = 'running'",
|
|
)
|
|
.bind(event_id)
|
|
.bind(export_type)
|
|
.bind(file_path)
|
|
.bind(epoch)
|
|
.execute(pool)
|
|
.await
|
|
.context("finalizing export job")?;
|
|
Ok(r.rows_affected() > 0)
|
|
}
|
|
|
|
/// 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()
|
|
}
|
|
|
|
/// Filenames a live (current-epoch, `done`) job row still points at — OFF LIMITS to every prune,
|
|
/// regardless of the epoch encoded in the name.
|
|
///
|
|
/// A ViewerOnly regeneration carries the finished ZIP forward by re-stamping its row to the new
|
|
/// epoch WITHOUT renaming the file, so `Gallery.<event>.<older>.zip` is still the served archive and
|
|
/// deleting it by filename-epoch would 404 the download.
|
|
async fn protected_files(pool: &PgPool, event_id: Uuid) -> Vec<String> {
|
|
sqlx::query_scalar::<_, String>(
|
|
"SELECT split_part(j.file_path, '/', -1) FROM export_job j
|
|
JOIN event e ON e.id = j.event_id
|
|
WHERE j.event_id = $1 AND j.epoch = e.export_epoch
|
|
AND j.status = 'done' AND j.file_path IS NOT NULL",
|
|
)
|
|
.bind(event_id)
|
|
.fetch_all(pool)
|
|
.await
|
|
.unwrap_or_default()
|
|
}
|
|
|
|
/// Reclaim superseded FINAL archives.
|
|
///
|
|
/// CALLED AFTER A SUCCESSFUL BUILD, not before one. This doc used to argue the opposite at
|
|
/// length — that since readiness is derived from `job.epoch = event.export_epoch`, a superseded
|
|
/// archive is already unreachable and keeping it "buys nothing". That reasoning is right about
|
|
/// REACHABILITY and wrong about RECOVERABILITY: an epoch is a database value that can be rolled
|
|
/// back, deleted bytes cannot. Pruning first meant any rebuild that then failed — ENOSPC, an OOM,
|
|
/// a hung ffmpeg — left the event with NO archive at all, which is the one outcome the product
|
|
/// exists to prevent, at the one moment nobody is watching.
|
|
///
|
|
/// The single exception is phase 2 of `ensure_export_space_reclaiming`, and BE PRECISE ABOUT WHAT
|
|
/// THAT EXCEPTION NOW COSTS, because the guarantee above is weaker than it reads. That phase used
|
|
/// to prune only when the reclaimed bytes would actually close the shortfall. It no longer does:
|
|
/// `reclaimable` is scoped to one prefix while `deficit` covers both halves plus the reserve, so on
|
|
/// a tight disk each worker measured its own share as insufficient, neither pruned, and every "Neu
|
|
/// erzeugen" refused identically — permanently stuck, with dead archives on the volume that nothing
|
|
/// would reclaim and nothing could serve. It now prunes anyway and lets the re-check decide.
|
|
///
|
|
/// The trade that buys: a rebuild can now delete the last physical copy and THEN fail, which is
|
|
/// precisely the "no archive at all" outcome this doc argues against. It is accepted because the
|
|
/// refusal it replaces was unrecoverable — deterministic across retries — whereas this failure
|
|
/// converges once the sibling worker frees its share. But "an epoch can be rolled back, deleted
|
|
/// bytes cannot" is no longer a guarantee this module provides end to end, and a manual
|
|
/// `UPDATE event SET export_epoch = <n>` can no longer rescue that case.
|
|
///
|
|
/// Both call sites carry the full reasoning; do not "restore" a pre-build prune on the strength of
|
|
/// this function's convenience.
|
|
///
|
|
/// Narrower than [`prune_stale_export_files`] on purpose: FINAL archives only. Those are inert — a
|
|
/// superseded worker either already renamed its file (and will delete it itself when its guarded
|
|
/// `finalize_job` fails) or never will. `.tmp` files and `viewer_tmp_` staging dirs are NOT touched
|
|
/// here: a superseded worker can still be streaming into those, and at build START it is far more
|
|
/// likely to be alive than at finalize time.
|
|
async fn prune_superseded_archives(
|
|
pool: &PgPool,
|
|
exports_dir: &Path,
|
|
prefix: &str,
|
|
event_id: Uuid,
|
|
keep_seq: i64,
|
|
) {
|
|
let protected = protected_files(pool, event_id).await;
|
|
let final_prefix = format!("{prefix}.{event_id}.");
|
|
let mut rd = match tokio::fs::read_dir(exports_dir).await {
|
|
Ok(rd) => rd,
|
|
Err(_) => return,
|
|
};
|
|
let mut reclaimed = 0u64;
|
|
while let Ok(Some(entry)) = rd.next_entry().await {
|
|
let name = entry.file_name();
|
|
let name = name.to_string_lossy();
|
|
if !is_superseded_archive(&name, &final_prefix, keep_seq, &protected) {
|
|
continue;
|
|
}
|
|
let len = entry.metadata().await.map(|m| m.len()).unwrap_or(0);
|
|
if tokio::fs::remove_file(entry.path()).await.is_ok() {
|
|
reclaimed += len;
|
|
}
|
|
}
|
|
if reclaimed > 0 {
|
|
tracing::info!(
|
|
"reclaimed {reclaimed} bytes of superseded {prefix} archives before rebuilding \
|
|
event {event_id} @ epoch {keep_seq}"
|
|
);
|
|
}
|
|
}
|
|
|
|
/// Is `name` a FINAL archive of a strictly-older generation that no live row points at?
|
|
///
|
|
/// Pure so the two dangerous cases can be pinned without a filesystem: the carried-forward archive
|
|
/// (protected despite an older epoch in its name) and the in-flight `.tmp` (never matched at all).
|
|
fn is_superseded_archive(
|
|
name: &str,
|
|
final_prefix: &str,
|
|
keep_seq: i64,
|
|
protected: &[String],
|
|
) -> bool {
|
|
if protected.iter().any(|p| p == name) {
|
|
return false;
|
|
}
|
|
parse_gen_seq(name, final_prefix, ".zip").is_some_and(|n| n < keep_seq)
|
|
}
|
|
|
|
/// Bytes the media in this event's keepsake will occupy, as an UPPER BOUND per archive.
|
|
///
|
|
/// Both archives write their media entries `Compression::Stored`, so an archive is essentially a
|
|
/// byte-for-byte second copy of the originals: `Gallery.zip` always, and `Memories.zip` for every
|
|
/// video ([`MediaSource::Original`]) and every image at or under 5 MB. Images over 5 MB are
|
|
/// downscaled to 2000px first, which only makes this estimate more conservative — the direction we
|
|
/// want, since being wrong low means ENOSPC halfway through.
|
|
///
|
|
/// Shares [`query_uploads`]' visibility filter via [`export_visibility_where`], so hidden/banned
|
|
/// uploads can't be counted here but skipped there (or the reverse, which under-reserves).
|
|
pub async fn estimate_export_bytes(pool: &PgPool, event_id: Uuid) -> Result<u64> {
|
|
let (bytes,): (i64,) = sqlx::query_as(concat!(
|
|
"SELECT COALESCE(SUM(u.original_size_bytes), 0)::bigint
|
|
FROM upload u
|
|
JOIN \"user\" usr ON usr.id = u.user_id
|
|
",
|
|
export_visibility_where!(),
|
|
))
|
|
.bind(event_id)
|
|
.fetch_one(pool)
|
|
.await
|
|
.context("estimating the export size")?;
|
|
Ok(bytes.max(0) as u64)
|
|
}
|
|
|
|
/// Headroom multiplier over the raw media sum: ZIP central directory, per-entry headers, the
|
|
/// embedded viewer, and the HTML export's temp staging.
|
|
const EXPORT_SIZE_OVERHEAD_PCT: u64 = 110;
|
|
|
|
/// Bytes this export must have available, given the raw media sum and how many jobs are competing.
|
|
///
|
|
/// `armed` is the multiplier that keeps two concurrent workers honest. `spawn_export_jobs` starts
|
|
/// the ZIP and the HTML halves at the same instant and both are gallery-sized, so a worker that
|
|
/// reserved only for itself would see "it fits", its sibling would independently see the same, and
|
|
/// together they would not fit — which is precisely the ENOSPC this preflight exists to prevent.
|
|
/// Computed in `u128` and clamped, NOT with `saturating_mul`: saturating first and then dividing by
|
|
/// 100 quietly turns an overflow into a number ~100x too small, which is the one direction that
|
|
/// matters here — an under-estimate authorises the very write the preflight exists to refuse.
|
|
pub(crate) fn required_free_bytes(media_bytes: u64, armed: i64) -> u64 {
|
|
let needed = media_bytes as u128 * EXPORT_SIZE_OVERHEAD_PCT as u128 / 100
|
|
* armed.max(1).min(i64::from(u32::MAX)) as u128;
|
|
needed.min(u64::MAX as u128) as u64
|
|
}
|
|
|
|
/// Free bytes a full keepsake build would need RIGHT NOW, both halves included.
|
|
///
|
|
/// The same arithmetic the preflight uses, exposed so the host dashboard can warn BEFORE the
|
|
/// release rather than reporting a failure after it. The preflight can only ever say "this didn't
|
|
/// fit"; at that point the gallery is full, the event is over, and the remedies (ask guests to stop
|
|
/// uploading, grow the volume) are all much harder. Hard-codes both halves because that is what a
|
|
/// release arms.
|
|
pub async fn keepsake_space_required(pool: &PgPool, event_id: Uuid) -> Result<u64> {
|
|
Ok(required_free_bytes(
|
|
estimate_export_bytes(pool, event_id).await?,
|
|
2,
|
|
))
|
|
}
|
|
|
|
/// Refuse to start an export that cannot fit, with a reason the host can act on.
|
|
///
|
|
/// Without this the failure mode is ENOSPC halfway through a multi-GB write, and the wreckage
|
|
/// outlives it: the epoch has already moved, so the job row is `failed` at the CURRENT generation
|
|
/// and `GET /export/zip` 404s, while the last good archive sits on disk unreferenced. The host's
|
|
/// only escape (`POST /host/export/rebuild`) needs the very space that isn't there. Failing at the
|
|
/// door instead leaves the disk untouched and puts a number in front of the operator.
|
|
///
|
|
/// Both halves are spawned concurrently and both are gallery-sized, so a worker must reserve for
|
|
/// its live sibling too — otherwise each independently sees "it fits", and together they don't.
|
|
async fn ensure_export_space(pool: &PgPool, event_id: Uuid, export_path: &Path) -> Result<()> {
|
|
let media_bytes = estimate_export_bytes(pool, event_id).await?;
|
|
|
|
// Every job armed at any epoch for this event that hasn't finished is competing for this disk.
|
|
let (armed,): (i64,) = sqlx::query_as(
|
|
"SELECT COUNT(*) FROM export_job
|
|
WHERE event_id = $1 AND status IN ('pending', 'running')",
|
|
)
|
|
.bind(event_id)
|
|
.fetch_one(pool)
|
|
.await
|
|
.context("counting armed export jobs")?;
|
|
let needed = required_free_bytes(media_bytes, armed);
|
|
|
|
// `None` = the mount couldn't be resolved. Fail OPEN, exactly as the upload quota does: refusing
|
|
// to build the keepsake because we can't read a number would be a worse failure than trying.
|
|
let Some(free) = crate::services::disk::free_bytes(export_path) else {
|
|
tracing::warn!("export preflight: disk snapshot unavailable; proceeding without the check");
|
|
return Ok(());
|
|
};
|
|
|
|
// The archive may not consume the last byte of the volume. `needed` alone authorised an
|
|
// export sized at exactly `free`: it would pass the check, run for half an hour, and land
|
|
// the box at zero — at which point Postgres cannot write WAL and the event is over, with
|
|
// the keepsake still unfinished. `postgres_data`, `media_data` and `exports_data` share one
|
|
// filesystem, so "enough room for the archive" was never the same question as "enough room
|
|
// for the archive AND a working database".
|
|
//
|
|
// Same reserve the upload path and the host dashboard's low-disk banner use, so all three
|
|
// agree on what "full" means.
|
|
let required = needed.saturating_add(crate::handlers::upload::DISK_RESERVE_BYTES as u64);
|
|
|
|
if free < required {
|
|
let gb = |b: u64| b as f64 / 1_000_000_000.0;
|
|
tracing::error!(
|
|
needed,
|
|
required,
|
|
free,
|
|
armed,
|
|
"export preflight: not enough free space to build the keepsake for event {event_id}"
|
|
);
|
|
anyhow::bail!(
|
|
"Nicht genug Speicherplatz für das Keepsake: benötigt ca. {:.1} GB (plus {:.0} GB \
|
|
Reserve), frei sind {:.1} GB. Bitte Speicher freigeben und das Keepsake \
|
|
anschließend neu erstellen.",
|
|
gb(needed),
|
|
gb(crate::handlers::upload::DISK_RESERVE_BYTES as u64),
|
|
gb(free)
|
|
);
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
/// How many bytes short of buildable we are, or `None` when there is already room.
|
|
///
|
|
/// Split out of `ensure_export_space` so the reclaiming wrapper can ask "would pruning be
|
|
/// enough?" BEFORE it destroys anything. Mirrors that function's arithmetic exactly; the one
|
|
/// deliberate difference is that an unresolvable mount reports `None` (fail open) rather than a
|
|
/// deficit, so a missing disk reading can never be the reason we delete the only archive.
|
|
async fn export_space_deficit(
|
|
pool: &PgPool,
|
|
event_id: Uuid,
|
|
export_path: &Path,
|
|
) -> Result<Option<u64>> {
|
|
let media_bytes = estimate_export_bytes(pool, event_id).await?;
|
|
let (armed,): (i64,) = sqlx::query_as(
|
|
"SELECT COUNT(*) FROM export_job
|
|
WHERE event_id = $1 AND status IN ('pending', 'running')",
|
|
)
|
|
.bind(event_id)
|
|
.fetch_one(pool)
|
|
.await
|
|
.context("counting armed export jobs")?;
|
|
let required = required_free_bytes(media_bytes, armed)
|
|
.saturating_add(crate::handlers::upload::DISK_RESERVE_BYTES as u64);
|
|
|
|
let Some(free) = crate::services::disk::free_bytes(export_path) else {
|
|
return Ok(None);
|
|
};
|
|
Ok((free < required).then(|| required - free))
|
|
}
|
|
|
|
/// Sum what `prune_superseded_archives` WOULD reclaim, without deleting anything.
|
|
///
|
|
/// Deliberately shares `is_superseded_archive` and `protected_files` with the real prune, so the
|
|
/// estimate cannot drift from what actually gets removed. An unreadable directory reports 0,
|
|
/// which makes the caller refuse to prune — the safe direction.
|
|
async fn reclaimable_superseded_bytes(
|
|
pool: &PgPool,
|
|
exports_dir: &Path,
|
|
prefix: &str,
|
|
event_id: Uuid,
|
|
keep_seq: i64,
|
|
) -> u64 {
|
|
let protected = protected_files(pool, event_id).await;
|
|
let final_prefix = format!("{prefix}.{event_id}.");
|
|
let Ok(mut rd) = tokio::fs::read_dir(exports_dir).await else {
|
|
return 0;
|
|
};
|
|
let mut total = 0u64;
|
|
while let Ok(Some(entry)) = rd.next_entry().await {
|
|
let name = entry.file_name();
|
|
let name = name.to_string_lossy();
|
|
if !is_superseded_archive(&name, &final_prefix, keep_seq, &protected) {
|
|
continue;
|
|
}
|
|
total = total.saturating_add(entry.metadata().await.map(|m| m.len()).unwrap_or(0));
|
|
}
|
|
total
|
|
}
|
|
|
|
/// 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(
|
|
pool: &PgPool,
|
|
exports_dir: &Path,
|
|
prefix: &str,
|
|
event_id: Uuid,
|
|
keep_seq: i64,
|
|
) {
|
|
let protected = protected_files(pool, event_id).await;
|
|
|
|
// 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,
|
|
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 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 protected.iter().any(|p| p == name.as_ref()) {
|
|
continue;
|
|
}
|
|
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` — ONLY for the generation it owns (`epoch` matches). A
|
|
/// superseded worker's failure is a no-op, so it can't clobber the fresh generation's row.
|
|
///
|
|
/// The status guard admits `pending` as well as `running`: if `claim_job` itself ERRORS (pool
|
|
/// timeout, connection reset) the row is still `pending`, and a guard of `status = 'running'` would
|
|
/// match nothing — leaving the job sitting at 0% with no worker and no error message, a spinner
|
|
/// forever. The epoch guard is what keeps this safe; the status guard is only there to avoid
|
|
/// stomping a `done` row.
|
|
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 epoch = $4 AND status IN ('running', 'pending')",
|
|
)
|
|
.bind(event_id)
|
|
.bind(export_type)
|
|
.bind(msg)
|
|
.bind(epoch)
|
|
.execute(pool)
|
|
.await;
|
|
}
|
|
|
|
/// Update the progress bar for THIS generation only, and report whether we're STILL THE LIVE
|
|
/// GENERATION.
|
|
///
|
|
/// The WHERE clause (`epoch = ours AND status = 'running'`) is exactly the liveness predicate, so
|
|
/// this write already tells us for free whether we've been superseded — it just used to throw the
|
|
/// answer away. Returning it lets the file loops bail the instant a reopen/re-release/takedown
|
|
/// retires us, instead of grinding through every remaining image and writing a whole archive we
|
|
/// will then delete. `false` = we lost (or the row is gone); a DB error is reported as still-live,
|
|
/// because a transient blip must not abandon a legitimate export.
|
|
async fn update_progress(
|
|
pool: &PgPool,
|
|
event_id: Uuid,
|
|
export_type: &str,
|
|
epoch: i64,
|
|
pct: i16,
|
|
) -> bool {
|
|
match sqlx::query(
|
|
"UPDATE export_job SET progress_pct = $3
|
|
WHERE event_id = $1 AND type = $2::export_type
|
|
AND epoch = $4 AND status = 'running'",
|
|
)
|
|
.bind(event_id)
|
|
.bind(export_type)
|
|
.bind(pct)
|
|
.bind(epoch)
|
|
.execute(pool)
|
|
.await
|
|
{
|
|
Ok(r) => r.rows_affected() > 0,
|
|
// Don't abandon a good export over a transient pool hiccup — the epoch-guarded finalize is
|
|
// still the authority, so at worst we do some extra work.
|
|
Err(e) => {
|
|
tracing::warn!("progress update failed for {export_type} @ epoch {epoch}: {e:#}");
|
|
true
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Sentinel error for "our generation was retired mid-export". The caller discards its output and
|
|
/// returns Ok — this is a normal, expected outcome, not a failure worth marking on the job row.
|
|
#[derive(Debug)]
|
|
struct Superseded;
|
|
|
|
impl std::fmt::Display for Superseded {
|
|
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
|
write!(f, "export generation superseded")
|
|
}
|
|
}
|
|
|
|
impl std::error::Error for Superseded {}
|
|
|
|
/// 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 complete: bool = sqlx::query_scalar(
|
|
"SELECT COUNT(*) = 2 FROM export_current
|
|
WHERE event_id = $1 AND status = 'done'",
|
|
)
|
|
.bind(event_id)
|
|
.fetch_one(pool)
|
|
.await
|
|
.unwrap_or(false);
|
|
|
|
{
|
|
if complete {
|
|
let _ = sse_tx.send(SseEvent {
|
|
event_type: "export-available".to_string(),
|
|
data: serde_json::json!({ "types": ["zip", "html"] }).to_string(),
|
|
});
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Write the embedded viewer into the ZIP, injecting the export data as a
|
|
/// `window.__EXPORT_DATA__` global into `index.html`. The keepsake is opened by
|
|
/// double-clicking `index.html` (file://), where browsers block a cross-origin
|
|
/// `fetch()` of a sibling `data.json` — so the data is inlined into the page.
|
|
/// (`data.json` is still written separately for the http-served case.)
|
|
/// Permissions stamped on every entry in both archives: `rw-r--r--`.
|
|
///
|
|
/// `ZipEntryBuilder::new` leaves the external file attribute at zero, and the host compatibility
|
|
/// defaults to Unix — so every entry was written with a stored mode of **0000**. Windows Explorer
|
|
/// ignores Unix modes and was fine, which is exactly why this survived: on Linux and macOS
|
|
/// `unzip` faithfully applies what the archive asks for, and the guest gets a directory of files
|
|
/// none of which they can open. `?---------` on every line of `unzip -Z`.
|
|
///
|
|
/// That is the keepsake — the artifact the whole event exists to produce — arriving unreadable,
|
|
/// after distribution, with no server-side symptom at all.
|
|
/// `S_IFREG | 0644`. The type bits are included because the mode is written whole into the high
|
|
/// half of the external file attribute: without them extractors see a file of type "unknown"
|
|
/// (`unzip -Z` renders `?rw-r--r--`), which works but is not what the archive means to say.
|
|
const KEEPSAKE_ENTRY_MODE: u16 = 0o100_644;
|
|
|
|
/// Build a ZIP entry for the keepsake. ALL entries in both archives go through here so the mode
|
|
/// can't be forgotten at one of the six call sites.
|
|
fn keepsake_entry(name: String, compression: Compression) -> ZipEntryBuilder {
|
|
ZipEntryBuilder::new(name.into(), compression).unix_permissions(KEEPSAKE_ENTRY_MODE)
|
|
}
|
|
|
|
/// Escape a JSON payload for inlining inside a `<script>` element.
|
|
///
|
|
/// See the call site in [`write_viewer_with_data`] for why this is every `<` and not just `</`.
|
|
/// Kept separate so the property that matters — no `<` survives, and the value still decodes to
|
|
/// the original — can be asserted without building a ZIP.
|
|
fn escape_json_for_script(data_json: &str) -> String {
|
|
data_json.replace('<', "\\u003c")
|
|
}
|
|
|
|
async fn write_viewer_with_data(
|
|
dir: &include_dir::Dir<'_>,
|
|
zip: &mut ZipFileWriter<tokio::fs::File>,
|
|
data_json: &str,
|
|
theme_css: Option<&str>,
|
|
) -> Result<()> {
|
|
for file in dir.files() {
|
|
let path = file.path().to_string_lossy().to_string();
|
|
if path == "index.html" {
|
|
let html = std::str::from_utf8(file.contents())
|
|
.context("export-viewer index.html is not valid UTF-8")?;
|
|
// Escape EVERY `<`, not just `</`.
|
|
//
|
|
// `</` -> `<\/` stops the obvious break-out (`</script><img onerror=…>`) and is inert
|
|
// against XSS. It does not stop the caption steering the HTML TOKENIZER. A caption
|
|
// containing `<!--<script` with no later `-->` puts the parser into
|
|
// script-data-double-escaped state; from there the template's own `</script>` only
|
|
// steps back to script-data-escaped instead of closing the element, and the rest of the
|
|
// document — including the viewer bundle — is swallowed as script data. Nothing
|
|
// executes and nothing leaks; `window.__EXPORT_DATA__` is simply never assigned and the
|
|
// keepsake opens blank.
|
|
//
|
|
// That failure is silent and POST-DISTRIBUTION: the export succeeds, the ZIP is
|
|
// well-formed, the job writes `done`, /export/status is green, and the host hands out a
|
|
// file that only fails when a guest double-clicks index.html — in every copy, with no
|
|
// way to fix it after the fact. Reachable from any guest-authored caption or comment,
|
|
// since both are embedded in the viewer.
|
|
//
|
|
// `<` never appears in JSON structural syntax — only inside string values — so a global
|
|
// replace is sound, and `<` is valid in both JSON and a JS string literal. One
|
|
// rule covers `</script`, `<!--` and `<script` together, which is the point: the
|
|
// previous escape was named for the single case it did handle.
|
|
//
|
|
// NOTE this is deliberately only for the INLINED copy. `data.json` is written
|
|
// separately, in no HTML context, and must stay literal.
|
|
let safe = escape_json_for_script(data_json);
|
|
// Match the live app's colour theme: inject the same `:root:root{…}` override
|
|
// the app builds at runtime so a rose/sage/custom event exports a rose/sage
|
|
// keepsake (not the embedded default gold). The CSS is generated purely from
|
|
// hex seeds (theme_override_css), so there's nothing to escape.
|
|
let mut head_inject = String::new();
|
|
if let Some(css) = theme_css {
|
|
head_inject.push_str(&format!("<style id=\"es-theme\">{css}</style>"));
|
|
}
|
|
head_inject.push_str(&format!("<script>window.__EXPORT_DATA__={safe};</script>"));
|
|
let injected = match html.find("</head>") {
|
|
Some(idx) => format!("{}{}{}", &html[..idx], head_inject, &html[idx..]),
|
|
None => format!("{head_inject}{html}"),
|
|
};
|
|
let builder = keepsake_entry(path, Compression::Deflate);
|
|
let mut entry = zip.write_entry_stream(builder).await?;
|
|
let mut cursor = AllowStdIo::new(std::io::Cursor::new(injected.as_bytes()));
|
|
fcopy(&mut cursor, &mut entry).await?;
|
|
entry.close().await?;
|
|
} else {
|
|
let builder = keepsake_entry(path, Compression::Deflate);
|
|
let mut entry = zip.write_entry_stream(builder).await?;
|
|
let mut cursor = AllowStdIo::new(std::io::Cursor::new(file.contents()));
|
|
fcopy(&mut cursor, &mut entry).await?;
|
|
entry.close().await?;
|
|
}
|
|
}
|
|
for sub_dir in dir.dirs() {
|
|
Box::pin(write_viewer_with_data(sub_dir, zip, data_json, theme_css)).await?;
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
/// Default champagne-gold seed — matches the hand-tuned ramp already embedded in the
|
|
/// viewer, so a default-themed event injects no override.
|
|
const KEEPSAKE_DEFAULT_SEED: &str = "#8a6a2b";
|
|
|
|
// stop → color-mix instruction. MIRRORS `LADDER` in frontend/src/lib/theme/palette.ts —
|
|
// keep the two in sync so an exported keepsake matches the live app pixel-for-pixel.
|
|
const THEME_LADDER: &[(u16, Option<&str>)] = &[
|
|
(50, Some("white 90%")),
|
|
(100, Some("white 80%")),
|
|
(200, Some("white 62%")),
|
|
(300, Some("white 42%")),
|
|
(400, Some("white 22%")),
|
|
(500, Some("white 9%")),
|
|
(600, None),
|
|
(700, Some("black 15%")),
|
|
(800, Some("black 30%")),
|
|
(900, Some("black 45%")),
|
|
(950, Some("black 63%")),
|
|
];
|
|
|
|
fn theme_stop_value(seed: &str, mix: Option<&str>) -> String {
|
|
match mix {
|
|
Some(m) => format!("color-mix(in oklab, {seed}, {m})"),
|
|
None => seed.to_string(),
|
|
}
|
|
}
|
|
|
|
fn theme_ramp(families: &[&str], seed: &str) -> String {
|
|
let mut out = String::new();
|
|
for (stop, mix) in THEME_LADDER {
|
|
let val = theme_stop_value(seed, *mix);
|
|
for fam in families {
|
|
out.push_str(&format!("--color-{fam}-{stop}:{val};"));
|
|
}
|
|
}
|
|
out
|
|
}
|
|
|
|
fn is_hex_color(s: &str) -> bool {
|
|
let b = s.as_bytes();
|
|
b.len() == 7 && b[0] == b'#' && b[1..].iter().all(|c| c.is_ascii_hexdigit())
|
|
}
|
|
|
|
/// Build the keepsake's `:root:root{…}` colour override from two seed colours, or None
|
|
/// for the default gold (viewer already carries that ramp) or an invalid seed (fall back
|
|
/// to the embedded default rather than emit unsafe CSS). MIRRORS `buildPaletteCss` in
|
|
/// frontend/src/lib/theme/palette.ts.
|
|
fn theme_override_css(primary: &str, accent: &str) -> Option<String> {
|
|
if !is_hex_color(primary) || !is_hex_color(accent) {
|
|
return None;
|
|
}
|
|
if primary.eq_ignore_ascii_case(KEEPSAKE_DEFAULT_SEED)
|
|
&& accent.eq_ignore_ascii_case(KEEPSAKE_DEFAULT_SEED)
|
|
{
|
|
return None;
|
|
}
|
|
let accent500 = theme_stop_value(accent, Some("white 9%"));
|
|
let mut css = String::from(":root:root{");
|
|
css.push_str(&theme_ramp(&["blue", "primary"], primary));
|
|
css.push_str(&theme_ramp(&["purple"], accent));
|
|
css.push_str(&format!(
|
|
"--color-violet-500:{accent500};--color-violet-600:{accent};"
|
|
));
|
|
css.push_str(&format!(
|
|
"--color-accent-500:{accent500};--color-accent-600:{accent};"
|
|
));
|
|
css.push('}');
|
|
Some(css)
|
|
}
|
|
|
|
/// Read the active theme seeds from the runtime `config` table (set by the admin UI),
|
|
/// falling back to the default gold. NOTE: an env-only default (THEME_PRIMARY set but
|
|
/// never saved via the admin UI) isn't stored in this table, so such a keepsake would
|
|
/// use gold; admin-set themes — the normal path — match the app exactly.
|
|
async fn resolve_theme_seeds(pool: &PgPool) -> (String, String) {
|
|
async fn read(pool: &PgPool, key: &str) -> String {
|
|
sqlx::query_scalar::<_, String>("SELECT value FROM config WHERE key = $1")
|
|
.bind(key)
|
|
.fetch_optional(pool)
|
|
.await
|
|
.ok()
|
|
.flatten()
|
|
.unwrap_or_else(|| KEEPSAKE_DEFAULT_SEED.to_string())
|
|
}
|
|
(
|
|
read(pool, "theme_primary").await,
|
|
read(pool, "theme_accent").await,
|
|
)
|
|
}
|
|
|
|
fn ext_from_path(path: &str) -> &str {
|
|
path.rsplit('.').next().unwrap_or("bin")
|
|
}
|
|
|
|
fn sanitize_name(name: &str) -> String {
|
|
name.chars()
|
|
.map(|c| {
|
|
if c.is_alphanumeric() || c == '-' {
|
|
c
|
|
} else {
|
|
'_'
|
|
}
|
|
})
|
|
.collect()
|
|
}
|
|
|
|
// ── Static content ───────────────────────────────────────────────────────────
|
|
|
|
const README_TEXT: &str = "EventSnap Offline-Galerie\n\
|
|
\n\
|
|
So geht's:\n\
|
|
1. Entpacke diese ZIP-Datei\n\
|
|
(Windows: Rechtsklick > \"Alle extrahieren\"; Mac: Doppelklick;\n\
|
|
Handy: Dateimanager-App verwenden).\n\
|
|
2. Öffne \"index.html\" im Browser\n\
|
|
(z. B. Chrome, Safari oder Firefox).\n\
|
|
3. Stöbere durch alle Fotos und Videos.\n\
|
|
Du kannst zwischen Listen- und Rasteransicht wechseln,\n\
|
|
nach Hashtags filtern und nach Nutzern suchen.\n\
|
|
4. Eine Internetverbindung ist nicht nötig.\n\
|
|
Alles ist lokal auf deinem Gerät gespeichert.\n\
|
|
\n\
|
|
Viel Freude mit den Erinnerungen!\n";
|
|
|
|
// ── Tests ────────────────────────────────────────────────────────────────────
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
/// The viewer must actually be compiled into the binary.
|
|
///
|
|
/// `include_dir!` over an empty directory is not an error, and `write_viewer_with_data`
|
|
/// iterates `dir.files()` — zero files, zero writes, `Ok(())`. So a viewer build that failed
|
|
/// after Vite emptied its output directory produced a binary whose Memories.zip contains every
|
|
/// photo and no way to view them, with nothing anywhere reporting it. This is the cheapest
|
|
/// place to notice, and it runs on every `cargo test`.
|
|
#[test]
|
|
fn the_keepsake_viewer_is_compiled_into_this_binary() {
|
|
let index = super::VIEWER_DIR
|
|
.get_file("index.html")
|
|
.expect("static/export-viewer/index.html must be compiled in — run `npm run build` in frontend/export-viewer");
|
|
// Not just present: substantial. An empty or truncated file would satisfy `get_file` and
|
|
// still ship a blank keepsake. The real artifact is ~235 KB with the fonts inlined.
|
|
assert!(
|
|
index.contents().len() > 50_000,
|
|
"the compiled-in viewer is only {} bytes — that is not a complete keepsake viewer",
|
|
index.contents().len()
|
|
);
|
|
// And it must be self-contained: the whole point of the inlining is that it opens from
|
|
// file:// with no network. A `/fonts/...` reference here is the bug shipping again.
|
|
let html = std::str::from_utf8(index.contents()).expect("viewer is valid UTF-8");
|
|
assert!(
|
|
!html.contains("url(/"),
|
|
"the compiled-in viewer references an external asset — it will 404 silently from file://"
|
|
);
|
|
}
|
|
|
|
use super::*;
|
|
|
|
const EVT: &str = "11111111-1111-1111-1111-111111111111";
|
|
|
|
fn gallery_prefix() -> String {
|
|
format!("Gallery.{EVT}.")
|
|
}
|
|
|
|
#[test]
|
|
fn a_burst_of_invalidations_still_starts_a_build_within_one_window() {
|
|
// B7's invariant, and the reason the burst start is tracked at all. A per-request delay
|
|
// meant every invalidation restarted the wait, so a host moderating faster than one action
|
|
// per 20 seconds deferred the build forever — `/export/html` 404s and the UI sits on
|
|
// "Wird vorbereitet…" for the rest of the event.
|
|
//
|
|
// What must hold is that the delay is measured from the FIRST request in the burst, so the
|
|
// wait never grows: request #50 of a takedown pass is scheduled no later than request #1
|
|
// was. Coalescing survives (later requests still collapse into one build); starvation does
|
|
// not.
|
|
let event = Uuid::new_v4();
|
|
clear_regen_burst(event);
|
|
|
|
let first = regen_delay_for(event);
|
|
assert!(
|
|
first <= REGEN_DEBOUNCE,
|
|
"the first invalidation of a burst waits at most one debounce window"
|
|
);
|
|
|
|
let mut previous = first;
|
|
for _ in 0..50 {
|
|
let next = regen_delay_for(event);
|
|
assert!(
|
|
next <= previous,
|
|
"a later invalidation must never push the build further out than an earlier one"
|
|
);
|
|
previous = next;
|
|
}
|
|
|
|
// And the burst is a scheduling window, not a permanent state: once a worker has run,
|
|
// the next invalidation is a fresh burst that gets the full coalescing delay again.
|
|
clear_regen_burst(event);
|
|
assert!(
|
|
regen_delay_for(event) >= previous,
|
|
"clearing the burst starts a fresh debounce window"
|
|
);
|
|
clear_regen_burst(event);
|
|
}
|
|
|
|
#[test]
|
|
fn one_events_burst_never_schedules_another_events_rebuild() {
|
|
// The map is keyed per event because the delay is a property of THAT event's moderation
|
|
// pass. A shared window would let a busy event drag an idle one's keepsake along with it —
|
|
// or worse, let an idle event's stale entry start a busy one's build immediately, which is
|
|
// the stampede the debounce exists to prevent.
|
|
let busy = Uuid::new_v4();
|
|
let quiet = Uuid::new_v4();
|
|
clear_regen_burst(busy);
|
|
clear_regen_burst(quiet);
|
|
|
|
for _ in 0..10 {
|
|
regen_delay_for(busy);
|
|
}
|
|
clear_regen_burst(busy);
|
|
|
|
assert!(
|
|
regen_delay_for(quiet) <= REGEN_DEBOUNCE,
|
|
"an untouched event gets its own full window"
|
|
);
|
|
clear_regen_burst(quiet);
|
|
}
|
|
|
|
#[test]
|
|
fn a_strictly_older_archive_is_reclaimed() {
|
|
// The whole point: at the start of a rebuild at epoch 5, generation 4's archive is dead
|
|
// weight — readiness is derived from `epoch = event.export_epoch`, so it is already
|
|
// unreachable — and its bytes are very often exactly the bytes the rebuild needs.
|
|
assert!(is_superseded_archive(
|
|
&format!("Gallery.{EVT}.4.zip"),
|
|
&gallery_prefix(),
|
|
5,
|
|
&[]
|
|
));
|
|
}
|
|
|
|
#[test]
|
|
fn our_own_and_newer_generations_are_never_touched() {
|
|
// `keep_seq` is OUR generation; a NEWER one belongs to a re-release that has already
|
|
// superseded us, and deleting it would let a lagging worker nuke a live keepsake.
|
|
for seq in [5, 6] {
|
|
assert!(
|
|
!is_superseded_archive(
|
|
&format!("Gallery.{EVT}.{seq}.zip"),
|
|
&gallery_prefix(),
|
|
5,
|
|
&[]
|
|
),
|
|
"generation {seq} must survive a prune keeping 5"
|
|
);
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn a_carried_forward_archive_survives_despite_an_older_epoch_in_its_name() {
|
|
// THE dangerous case. A ViewerOnly regeneration (a moderated comment) re-stamps the
|
|
// finished ZIP's row to the new epoch WITHOUT renaming the file, so the SERVED archive
|
|
// legitimately carries an older generation number. Pruning it by filename-epoch would 404
|
|
// the photo download to change nothing in it. The protected set is what stops that, and
|
|
// moving the prune to build-start makes this case reachable far more often.
|
|
let carried = format!("Gallery.{EVT}.4.zip");
|
|
assert!(!is_superseded_archive(
|
|
&carried,
|
|
&gallery_prefix(),
|
|
5,
|
|
std::slice::from_ref(&carried)
|
|
));
|
|
}
|
|
|
|
#[test]
|
|
fn temps_and_staging_dirs_are_out_of_scope_for_the_early_prune() {
|
|
// A superseded worker may still be streaming into these, and at build START it is much
|
|
// more likely to be alive than at finalize time. Only inert FINAL archives are reclaimed
|
|
// here; `prune_stale_export_files` still handles the rest after we win.
|
|
for name in [
|
|
format!("Gallery.{EVT}.4.tmp"),
|
|
format!("viewer_tmp_{EVT}_4"),
|
|
format!("Memories.{EVT}.4.zip"),
|
|
] {
|
|
assert!(
|
|
!is_superseded_archive(&name, &gallery_prefix(), 5, &[]),
|
|
"{name} must not be reclaimed by the pre-build prune"
|
|
);
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn another_events_archive_is_never_reclaimed() {
|
|
// All events share one exports volume, so the prefix carries the event id.
|
|
let other = "22222222-2222-2222-2222-222222222222";
|
|
assert!(!is_superseded_archive(
|
|
&format!("Gallery.{other}.4.zip"),
|
|
&gallery_prefix(),
|
|
5,
|
|
&[]
|
|
));
|
|
}
|
|
|
|
#[test]
|
|
fn unrelated_files_are_left_alone() {
|
|
for name in ["Gallery.zip", "notes.txt", "Gallery..4.zip"] {
|
|
assert!(!is_superseded_archive(name, &gallery_prefix(), 5, &[]));
|
|
}
|
|
}
|
|
|
|
/// Every `<` is escaped, whatever it is part of.
|
|
///
|
|
/// PREVENTS the regression to `</` -> `<\\/`, which is named for the one case it handles.
|
|
/// `<!--<script` with no later `-->` drives the HTML tokenizer into
|
|
/// script-data-double-escaped state, where the template's own `</script>` no longer closes
|
|
/// the element — the viewer bundle is swallowed as script data, `__EXPORT_DATA__` is never
|
|
/// assigned, and the keepsake opens blank in every copy the host has already handed out.
|
|
#[test]
|
|
fn no_left_angle_bracket_survives_inlining() {
|
|
for payload in [
|
|
r#"{"caption":"<!--<script"}"#,
|
|
r#"{"caption":"</script><img src=x onerror=alert(1)>"}"#,
|
|
r#"{"caption":"<!--"}"#,
|
|
r#"{"caption":"<script>"}"#,
|
|
r#"{"caption":"a < b"}"#,
|
|
] {
|
|
let escaped = escape_json_for_script(payload);
|
|
assert!(
|
|
!escaped.contains('<'),
|
|
"a surviving `<` can still steer the tokenizer: {escaped}"
|
|
);
|
|
}
|
|
}
|
|
|
|
/// The escape must not change what the viewer READS — it is a transport encoding, not a
|
|
/// sanitiser. A caption is guest-authored text that has to render back exactly.
|
|
#[test]
|
|
fn the_payload_still_decodes_to_the_original_value() {
|
|
// `<` appears only inside JSON string values, never in structural syntax, so a global
|
|
// replace is sound — this is the assertion that says so.
|
|
for caption in [
|
|
"<!--<script",
|
|
"</script><img src=x onerror=alert(1)>",
|
|
"a < b und c > d",
|
|
"ganz normale Bildunterschrift",
|
|
"Herz <3",
|
|
] {
|
|
let json = serde_json::json!({ "posts": [{ "caption": caption }] }).to_string();
|
|
let escaped = escape_json_for_script(&json);
|
|
let back: serde_json::Value =
|
|
serde_json::from_str(&escaped).expect("the escaped form must still be valid JSON");
|
|
assert_eq!(
|
|
back["posts"][0]["caption"].as_str(),
|
|
Some(caption),
|
|
"the caption must survive the round trip unchanged"
|
|
);
|
|
}
|
|
}
|
|
|
|
/// Nothing else in the document is touched.
|
|
#[test]
|
|
fn a_payload_with_no_angle_brackets_is_unchanged() {
|
|
let json = r#"{"posts":[{"caption":"schönes Foto"}]}"#;
|
|
assert_eq!(escape_json_for_script(json), json);
|
|
}
|
|
|
|
#[test]
|
|
fn a_lone_armed_job_reserves_for_one_archive() {
|
|
// A ViewerOnly regeneration re-arms only the HTML half — reserving for two would refuse
|
|
// a rebuild that fits perfectly well.
|
|
assert_eq!(required_free_bytes(1_000, 1), 1_100);
|
|
}
|
|
|
|
#[test]
|
|
fn two_concurrent_halves_reserve_for_both() {
|
|
// The bug this exists to prevent: each worker independently sees "it fits", and together
|
|
// they don't. Both halves are gallery-sized, so the reservation must be for the pair.
|
|
assert_eq!(required_free_bytes(1_000, 2), 2_200);
|
|
}
|
|
|
|
#[test]
|
|
fn a_zero_count_still_reserves_for_one() {
|
|
// Defensive: a racing status transition must never yield a zero requirement, which would
|
|
// wave through an export of any size onto a full disk.
|
|
assert_eq!(required_free_bytes(1_000, 0), 1_100);
|
|
}
|
|
|
|
#[test]
|
|
fn an_empty_gallery_needs_nothing() {
|
|
assert_eq!(required_free_bytes(0, 2), 0);
|
|
}
|
|
|
|
#[test]
|
|
fn a_pathological_size_saturates_instead_of_wrapping() {
|
|
// u64 overflow would wrap to a TINY requirement and authorise the exact write we're
|
|
// guarding against — the failure mode must be "refuse", never "wrap and allow".
|
|
assert_eq!(required_free_bytes(u64::MAX, 2), u64::MAX);
|
|
}
|
|
|
|
/// The completeness gate decides whether a partially-readable gallery still ships. Both
|
|
/// directions are dangerous: too strict and one live delete costs the guests every photo
|
|
/// (a released event cannot be rebuilt without reopening uploads); too loose and a wrong
|
|
/// MEDIA_PATH ships an archive that silently misrepresents the whole party.
|
|
#[test]
|
|
fn a_clean_export_and_a_totally_empty_one_are_unambiguous() {
|
|
// Nothing skipped is always fine, at any size.
|
|
assert!(check_export_completeness("zip", Uuid::nil(), 0, 0, 0).is_ok());
|
|
assert!(check_export_completeness("zip", Uuid::nil(), 100, 100, 0).is_ok());
|
|
|
|
// Everything skipped is the misconfiguration case — never publish it, at any size.
|
|
assert!(check_export_completeness("zip", Uuid::nil(), 100, 0, 100).is_err());
|
|
assert!(check_export_completeness("zip", Uuid::nil(), 3, 0, 3).is_err());
|
|
assert!(check_export_completeness("zip", Uuid::nil(), 1, 0, 1).is_err());
|
|
}
|
|
|
|
#[test]
|
|
fn a_small_gallery_survives_an_ordinary_live_delete() {
|
|
// These are the regressions a pure 10% fraction caused: on a small gallery a single
|
|
// guest deleting their own photo mid-build is >10%, so the whole keepsake failed.
|
|
assert!(check_export_completeness("zip", Uuid::nil(), 9, 8, 1).is_ok());
|
|
assert!(check_export_completeness("zip", Uuid::nil(), 3, 2, 1).is_ok());
|
|
assert!(check_export_completeness("zip", Uuid::nil(), 4, 2, 2).is_ok());
|
|
// Half a tiny gallery is published too, and logged as materially incomplete. Refusing it
|
|
// would hand the guests nothing at all rather than the three photos that DID survive, and
|
|
// the rebuild that refusal invites reads the same unreadable files again.
|
|
assert!(check_export_completeness("zip", Uuid::nil(), 6, 3, 3).is_ok());
|
|
}
|
|
|
|
#[test]
|
|
fn a_materially_incomplete_archive_is_still_published_but_shouted_about() {
|
|
// This used to assert `is_err()` past the 10% tolerance, and that was the wrong trade.
|
|
// The refusal is deterministic across retries — the unreadable files are still unreadable
|
|
// when the host taps "Neu erzeugen", and the gallery is already released so the photos
|
|
// cannot be collected again. So refusing did not buy a better archive later; it converted
|
|
// "90 of 100 photos" into "no keepsake, ever". Past the tolerance we publish and log at
|
|
// `error` with the counts.
|
|
assert!(check_export_completeness("zip", Uuid::nil(), 100, 90, 10).is_ok());
|
|
assert!(check_export_completeness("zip", Uuid::nil(), 100, 89, 11).is_ok());
|
|
assert!(check_export_completeness("zip", Uuid::nil(), 300, 200, 100).is_ok());
|
|
}
|
|
|
|
#[test]
|
|
fn an_archive_with_nothing_in_it_is_still_refused() {
|
|
// The one fatal case, and the only one the host can actually act on: a wrong MEDIA_PATH
|
|
// produced a few-hundred-byte ZIP with zero photos that passed every automated check and
|
|
// was advertised as ready. That IS recoverable — fix the path, rebuild — so refusing to
|
|
// publish it is the correct answer, unlike the partial case above.
|
|
assert!(check_export_completeness("zip", Uuid::nil(), 100, 0, 100).is_err());
|
|
// ...but an event that genuinely has no media is not an error.
|
|
assert!(check_export_completeness("zip", Uuid::nil(), 0, 0, 0).is_ok());
|
|
}
|
|
}
|