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

Merge branch 'fix/deploy-unattended-blockers' into main

Browse Source 
Two independent lines of production hardening diverged at 7d0334b and attacked
overlapping problems. Neither was a superset, so this is a merge of substance
rather than a fast-forward: every conflict was resolved on the merits, and the
losing side's intent was re-checked against the winner rather than assumed.

MIGRATIONS. The branch's 021/022/023 collided with main's already-DEPLOYED
021_hashtag_counts_respect_bans and 022_client_upload_idempotency. Renumbered to
023/024/025 in a prior commit — main's versions are applied in production, so
their version numbers are immutable and the branch's had to move. Verified by
running the full sqlx::test suite, which applies the whole chain from scratch.

RESOLVED IN MAIN'S FAVOUR (the branch would have regressed these):
  * upload-queue.ts wholesale — the branch's copy has ZERO client_upload_id
    references, so taking it would have silently destroyed end-to-end upload
    idempotency, the one thing standing between a lost response and a duplicate
    photo charged twice against the guest's quota.
  * maintenance.rs supervisor — the branch replaced it with a bare tokio::spawn,
    where one panic silently stops session pruning, media reclaim, the temp
    sweep and both HashMap prunes, permanently and with no log line.
  * The decode-budget probe on spawn_blocking, not inline on the async runtime.
  * feed/+page.svelte's 8s debounce + jitter + max-wait + hidden-tab deferral,
    against the branch's naive 800ms — at 100 guests the branch's version walks
    straight into the per-user feed rate limit.
  * db.rs pool tuning, /uploaders, and the docker-compose deployment story.
  * ONE /health, still DB-backed. The branch's split (dependency-free liveness +
    DB-backed readiness) is defensible, but a constant-"ok" /health is the exact
    defect faea555 fixed and verified live, its motive (Caddy's boot gate) is
    already covered by app depends_on db: service_healthy, and the two handlers
    were the same SELECT 1 under two names.

TAKEN FROM THE BRANCH:
  * The large-PNG OOM guard and its bounded-retry counter (023). Together these
    turn a single upload that can OOM-kill a 1G container into a bounded failure
    instead of an infinite restart loop under `restart: unless-stopped`.
  * 024_feed_scalar_counts — the feed no longer aggregates the whole event per
    page. Pure SQL; column names, order and types are unchanged by design.
  * The admin-lockout fix: look the admin up BY ROLE, never by name. 025 also
    frees any guest already squatting on a reserved name.
  * PIN lockout tier ordering, bounded caption/hashtag reads, SSE ticket caps,
    PoolTimedOut -> 503 + Retry-After, and the ffmpeg stderr drain.
  * backfill_video_posters, which main lacked entirely.
  * TempFileGuard, plus sweep_orphan_originals wired into main's SUPERVISED loop
    (not the branch's bare one) — it reclaims final-named originals whose commit
    never happened, a class main's .tmp-only sweep structurally cannot see.
  * shouldAbortForStall, hand-ported into main's upload-queue.ts since that file
    was resolved to main. Widens the watchdog at loadend instead of disarming it,
    bounding a half-open socket at 2 minutes rather than handing the window to
    xhr.timeout (5-60 min) with the whole queue's `processing` latch held.

ALSO: RUST_LOG and EXPORT_PATH pinned in compose. The code fallback was
`debug` (a line per request, all night) and EXPORT_PATH was the one path with a
mount-shaped default that nothing validated.

Verified: cargo check --all-targets, cargo clippy (clean), 144/144 backend tests
against a live Postgres including upload_idempotency and upload_concurrency,
51/51 vitest, svelte-check 0 errors, eslint clean, vite build.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
MechaCat02
2026-08-08 21:16:31 +02:00
parent d5b4bf0ac1 e1c689d1a7
commit 61119be817
30 changed files with 2192 additions and 439 deletions
Download Patch File Download Diff File Expand all files Collapse all files

547
backend/src/services/compression.rs
View File

@@ -13,6 +13,9 @@ use crate::state::SseEvent;
#[derive(Clone)]
pub struct CompressionWorker {
semaphore: Arc<Semaphore>,
/// Serialises the memory-heavy image jobs — see `HEAVY_IMAGE_BYTES`. Separate from
/// `semaphore` so ordinary photos keep full concurrency.
heavy: Arc<Semaphore>,
pool: PgPool,
media_path: PathBuf,
sse_tx: broadcast::Sender<SseEvent>,
@@ -31,6 +34,7 @@ impl CompressionWorker {
) -> Self {
Self {
semaphore: Arc::new(Semaphore::new(concurrency)),
heavy: Arc::new(Semaphore::new(1)),
pool,
media_path,
sse_tx,
@@ -58,6 +62,21 @@ impl CompressionWorker {
/// next start. Rev 1 = EXIF orientation is applied.
const DERIVATIVES_REV: i16 = 1;
/// How many times derivative generation may be ATTEMPTED for one upload before it is left
/// alone. Counted write-ahead and reset on success — see `Upload::begin_derivative_attempt`.
///
/// This is what turns a fatal input from an outage into a blemish. The startup backfill
/// runs unconditionally on every boot, so before this bound a row whose processing killed
/// the process was re-selected and re-run forever, and `restart: unless-stopped` made that
/// an infinite loop that also dropped every SSE stream and truncated every in-flight
/// upload on each cycle. Three attempts absorbs genuinely transient infrastructure
/// failures (an ENOSPC spike, a pool blip) without ever becoming unbounded.
const MAX_DERIVATIVE_ATTEMPTS: i16 = 3;
/// Rows regenerated per boot. Bounds both the query and the amount of work a single start
/// can queue; whatever is left is picked up on the next boot.
const BACKFILL_BATCH: i64 = 200;
/// Spawn a background task to process an uploaded file.
pub fn process(&self, upload_id: Uuid, original_path: String, mime_type: String) {
let worker = self.clone();
@@ -193,6 +212,21 @@ impl CompressionWorker {
let original = self.media_path.join(original_path);
if mime_type.starts_with("image/") {
// Count the attempt BEFORE doing the work — see `begin_derivative_attempt`. If this
// input is the one that kills the container, this write is the only record that
// survives, and it is what stops the boot backfill replaying it forever.
match Upload::begin_derivative_attempt(&self.pool, upload_id).await? {
Some(attempts) if attempts > Self::MAX_DERIVATIVE_ATTEMPTS => {
anyhow::bail!(
"derivative generation gave up after {} attempt(s)",
attempts - 1
);
}
Some(_) => {}
// The row vanished while this task waited on the semaphore. Nothing to do, and
// reporting a failure would broadcast into a stream that no longer has a card.
None => return Ok(()),
}
let (preview_rel, display_rel) = self
.generate_image_derivatives(upload_id, &original, mime_type)
.await?;
@@ -256,6 +290,38 @@ impl CompressionWorker {
/// Longest edge of the phone-feed "preview" (data-saver default).
const PREVIEW_MAX_EDGE: u32 = 800;
/// Above this pixel count the PNG original is stored as uploaded, unoptimised.
///
/// oxipng's peak memory scales with PIXELS, not file size: it decodes the PNG itself and
/// then evaluates row filters, each trial holding a full-size buffer. That is why a 2.82
/// MiB file could measure 1250 MiB of peak RSS inside a 1 GiB container — smooth,
/// synthetic content compresses to almost nothing on disk while still being 8000x8000.
/// 8 MP covers every real phone photo; beyond it we decline the (lossless, cosmetic)
/// saving rather than risk the OOM kill.
const OXIPNG_MAX_PIXELS: u64 = 8_000_000;
/// Estimated peak heap above which an image job takes the exclusive `heavy` permit.
///
/// `compression_concurrency` (default 2) bounds how many jobs run at once, but says
/// nothing about how much memory each one costs, and the container gets 1 GiB total. A
/// single 8000x8000 original measures ~516 MiB peak even with the decode correctly scoped
/// — two of those overlapping is 1032 MiB and another OOM kill, from nothing more exotic
/// than two guests uploading big photos at the same moment.
///
/// 150 MiB sits far above a normal phone photo (a 12 MP JPEG costs ~50 MiB all-in) so the
/// common path never serialises, and far below the point where two jobs stop fitting.
/// Throughput is unaffected for everything except the rare giant, which is exactly the
/// case that must not run in parallel with another giant.
const HEAVY_IMAGE_BYTES: u64 = 150 * 1024 * 1024;
/// Wall-clock ceiling for one oxipng run.
///
/// Bounds TIME, NOT MEMORY — oxipng checks the deadline between trials, so a single trial
/// still allocates in full. The pixel gate above and the sequential build (see
/// `default-features = false` in Cargo.toml) are what bound memory. Do not treat this
/// constant as the OOM fix.
const OXIPNG_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(20);
/// Decode the image ONCE and emit both derivatives — the 800px `preview` (phone feed)
/// and the 2048px `display` (diashow). Returns `(preview_rel, display_rel)`.
async fn generate_image_derivatives(
@@ -274,53 +340,28 @@ impl CompressionWorker {
let display_path = displays_dir.join(&filename);
let original = original.to_path_buf();
let mime_owned = mime_type.to_string();
let preview_max = Self::PREVIEW_MAX_EDGE;
let display_max = Self::DISPLAY_MAX_EDGE;
// Estimate the peak from the HEADER (no pixels decoded — the same kind of cheap probe
// the upload handler already does via `exceeds_decode_budget`) and, if this job is a
// giant, take the exclusive permit so it cannot overlap another giant. Held for the
// whole blocking section, released on drop including on error.
let estimate =
crate::services::imaging::estimated_processing_peak_bytes(&original, Self::DISPLAY_MAX_EDGE);
let _heavy_permit = match estimate {
Some(bytes) if bytes > Self::HEAVY_IMAGE_BYTES => {
tracing::debug!(
%upload_id,
estimated_mib = bytes / (1024 * 1024),
"waiting for the heavy-image permit"
);
Some(self.heavy.acquire().await)
}
_ => None,
};
// Run blocking image operations in a spawn_blocking task
tokio::task::spawn_blocking(move || -> Result<()> {
// Decompression-bomb limits + EXIF orientation, both in one place — see
// services::imaging for why neither may be skipped.
let img = crate::services::imaging::decode_oriented(&original)?;
// Preview: max 800px, preserving aspect ratio (data-saver feed).
img.resize(
preview_max,
preview_max,
image::imageops::FilterType::Lanczos3,
)
.save_with_format(&preview_path, image::ImageFormat::Jpeg)
.context("failed to save preview")?;
// Display: max 2048px for the diashow. Only DOWNSCALE — never upscale a smaller
// original (that adds bytes with no quality gain); re-encode it as JPEG as-is.
let display = if img.width() > display_max || img.height() > display_max {
img.resize(
display_max,
display_max,
image::imageops::FilterType::Lanczos3,
)
} else {
img
};
display
.save_with_format(&display_path, image::ImageFormat::Jpeg)
.context("failed to save display")?;
// If the original is PNG, try lossless compression in-place
if mime_owned == "image/png" {
let opts = oxipng::Options::from_preset(2);
let _ = oxipng::optimize(
&oxipng::InFile::Path(original),
&oxipng::OutFile::Path {
path: None,
preserve_attrs: true,
},
&opts,
);
}
Ok(())
tokio::task::spawn_blocking(move || {
write_image_derivatives(upload_id, &original, &mime_owned, &preview_path, &display_path)
})
.await??;
@@ -341,17 +382,29 @@ impl CompressionWorker {
///
/// Unlike the failure path in `process`, a backfill error is logged and skipped — it must
/// NEVER destroy or soft-delete an upload that already has a working preview.
///
/// Bounded in three ways, all of them load-bearing on a box that restarts itself:
/// `derivative_attempts` stops a fatal row being replayed on every boot, `BACKFILL_BATCH`
/// stops one start queueing unbounded work, and the whole thing runs as ONE task walking
/// the rows sequentially rather than N tasks racing for the same semaphore.
pub async fn backfill_stale_derivatives(&self) {
// `original_path IS NOT NULL` was dead — the column is NOT NULL. What actually needs
// excluding is the blanked path `cleanup_deleted_media` leaves behind.
let rows = sqlx::query_as::<_, (Uuid, String, String)>(
"SELECT id, original_path, mime_type FROM upload
WHERE deleted_at IS NULL AND mime_type LIKE 'image/%'
AND original_path IS NOT NULL
AND original_path <> ''
AND derivative_attempts < $2
AND (
(display_path IS NULL AND preview_path IS NOT NULL)
OR derivatives_rev < $1
)",
)
ORDER BY created_at DESC
LIMIT $3",
)
.bind(Self::DERIVATIVES_REV)
.bind(Self::MAX_DERIVATIVE_ATTEMPTS)
.bind(Self::BACKFILL_BATCH)
.fetch_all(&self.pool)
.await;
let rows = match rows {
@@ -361,14 +414,32 @@ impl CompressionWorker {
return;
}
};
self.report_exhausted_derivatives().await;
if rows.is_empty() {
return;
}
tracing::info!("regenerating derivatives for {} upload(s)", rows.len());
for (id, original_path, mime_type) in rows {
let worker = self.clone();
tokio::spawn(async move {
// ONE task for the whole batch. The previous shape spawned a task per row, so a large
// backlog created thousands of live tasks that each held a pool handle and queued on
// the same two semaphore permits, competing with live uploads for the entire boot.
let worker = self.clone();
tokio::spawn(async move {
for (id, original_path, mime_type) in rows {
let _permit = worker.semaphore.acquire().await;
// Write-ahead, exactly as in the live path: if this row is the one that kills
// the process, this increment is the only thing that outlives the SIGKILL.
match Upload::begin_derivative_attempt(&worker.pool, id).await {
Ok(Some(n)) if n > Self::MAX_DERIVATIVE_ATTEMPTS => continue,
Ok(Some(_)) => {}
Ok(None) => continue,
Err(e) => {
tracing::warn!(error = ?e, %id, "could not record a backfill attempt; skipping");
continue;
}
}
let original = worker.media_path.join(&original_path);
match worker
.generate_image_derivatives(id, &original, &mime_type)
@@ -377,6 +448,8 @@ impl CompressionWorker {
Ok((preview_rel, display_rel)) => {
let _ = Upload::set_preview_path(&worker.pool, id, &preview_rel).await;
let _ = Upload::set_display_path(&worker.pool, id, &display_rel).await;
// Clears derivative_attempts too, so a row that failed transiently is
// not one boot closer to being abandoned.
let _ =
Upload::set_derivatives_rev(&worker.pool, id, Self::DERIVATIVES_REV)
.await;
@@ -384,12 +457,133 @@ impl CompressionWorker {
}
Err(e) => {
// Leave the existing derivatives and the original intact; this row is
// simply retried on the next start. The rev stays behind, which is the
// marker that it still needs doing.
// retried on the next start until its attempt budget runs out. The rev
// stays behind, which is the marker that it still needs doing.
tracing::warn!(error = ?e, %id, "derivative backfill failed; leaving as-is");
let _ =
Upload::record_derivative_failure(&worker.pool, id, &format!("{e:#}"))
.await;
}
}
});
}
});
}
/// Re-extract poster frames for videos that never got one.
///
/// A video interrupted by a restart is stranded: `startup_recovery` flips its
/// `compression_status` from `processing` to `failed` and nothing re-enqueues it, so
/// `thumbnail_path` stays NULL forever while the clip itself plays fine. The feed shows a
/// posterless tile for the rest of the event, and after
/// `FAILED_ORIGINAL_RETENTION_DAYS` the reclaim sweep is entitled to the original.
///
/// Shares `derivative_attempts` with the image backfill on purpose. Note the consequence,
/// which is intended rather than a bug to fix later: `extract_poster_frame` returning
/// `Ok(false)` is a NORMAL, permanent outcome for a sub-second clip (Live Photos,
/// mis-taps), and since the counter is write-ahead and only cleared by a real success,
/// those clips stop being re-ffmpeg'd on every boot once the budget is spent.
pub async fn backfill_video_posters(&self) {
let rows = sqlx::query_as::<_, (Uuid, String)>(
"SELECT id, original_path FROM upload
WHERE deleted_at IS NULL AND mime_type LIKE 'video/%'
AND thumbnail_path IS NULL
AND original_path <> ''
AND derivative_attempts < $1
ORDER BY created_at DESC
LIMIT $2",
)
.bind(Self::MAX_DERIVATIVE_ATTEMPTS)
.bind(Self::BACKFILL_BATCH)
.fetch_all(&self.pool)
.await;
let rows = match rows {
Ok(r) => r,
Err(e) => {
tracing::warn!(error = ?e, "video poster backfill query failed");
return;
}
};
if rows.is_empty() {
return;
}
tracing::info!("re-extracting posters for {} video(s)", rows.len());
let worker = self.clone();
tokio::spawn(async move {
for (id, original_path) in rows {
let _permit = worker.semaphore.acquire().await;
match Upload::begin_derivative_attempt(&worker.pool, id).await {
Ok(Some(n)) if n > Self::MAX_DERIVATIVE_ATTEMPTS => continue,
Ok(Some(_)) => {}
Ok(None) => continue,
Err(e) => {
tracing::warn!(error = ?e, %id, "could not record a poster attempt; skipping");
continue;
}
}
let original = worker.media_path.join(&original_path);
match worker.generate_video_thumbnail(id, &original).await {
Ok(Some(thumb_rel)) => {
if Upload::set_thumbnail_path(&worker.pool, id, &thumb_rel)
.await
.is_ok()
{
// Clears the attempt counter: a video that eventually succeeded
// must not carry a budget scar into a future pipeline revision.
let _ = Upload::set_derivatives_rev(
&worker.pool,
id,
Self::DERIVATIVES_REV,
)
.await;
tracing::info!("poster regenerated for upload {id}");
}
}
// No frame at all — normal for a very short clip. The tile stays
// posterless and the attempt is spent, which is what stops the retry.
Ok(None) => {
tracing::debug!(%id, "still no poster frame; leaving the tile as-is");
}
Err(e) => {
tracing::warn!(error = ?e, %id, "poster backfill failed; leaving as-is");
let _ =
Upload::record_derivative_failure(&worker.pool, id, &format!("{e:#}"))
.await;
}
}
}
});
}
/// Say out loud, once per boot, that some uploads have stopped being retried.
///
/// Without this the give-up is invisible: the loop stops (which is the point) but the
/// affected photos keep a stale or missing derivative forever with nothing to notice. The
/// originals are untouched, so this is recoverable once the cause is fixed — reset
/// `derivative_attempts` to 0 and restart.
async fn report_exhausted_derivatives(&self) {
let exhausted: Result<i64, _> = sqlx::query_scalar(
"SELECT count(*) FROM upload
WHERE deleted_at IS NULL AND mime_type LIKE 'image/%'
AND derivative_attempts >= $2
AND (
(display_path IS NULL AND preview_path IS NOT NULL)
OR derivatives_rev < $1
)",
)
.bind(Self::DERIVATIVES_REV)
.bind(Self::MAX_DERIVATIVE_ATTEMPTS)
.fetch_one(&self.pool)
.await;
if let Ok(count) = exhausted
&& count > 0
{
tracing::error!(
count,
"{count} upload(s) exhausted derivative regeneration and will no longer be \
retried; their originals are intact — see upload.derivative_last_error, fix \
the cause, then reset derivative_attempts to 0 and restart"
);
}
}
@@ -413,3 +607,250 @@ impl CompressionWorker {
Ok(produced.then(|| format!("thumbnails/{thumb_filename}")))
}
}
/// The blocking half of [`CompressionWorker::generate_image_derivatives`]: decode once, write
/// both derivatives, then optionally shrink a PNG original in place.
///
/// A free function rather than an inline closure so its memory behaviour is directly testable —
/// this is the code path that OOM-killed the container, and the fix is a scoping property that a
/// future edit could silently undo.
fn write_image_derivatives(
upload_id: Uuid,
original: &Path,
mime_type: &str,
preview_path: &Path,
display_path: &Path,
) -> Result<()> {
let preview_max = CompressionWorker::PREVIEW_MAX_EDGE;
let display_max = CompressionWorker::DISPLAY_MAX_EDGE;
// THE FULL-SIZE DECODE IS SCOPED TO THIS BLOCK ON PURPOSE, and the block yields the
// DISPLAY derivative rather than the original.
//
// `img` is up to 256 MiB (imaging::decode_limits max_alloc) and `resize` only BORROWS it,
// so it used to stay alive through both resizes AND the oxipng call below — which decodes
// the PNG a second time and holds a full-size buffer per filter trial. That measured
// ~1250 MiB of peak RSS for a 2.8 MiB input, inside a 1 GiB cgroup: the container was
// SIGKILLed, taking every SSE stream and every in-flight upload with it.
//
// A block rather than a bare `drop(img)` because a `drop` call is one careless edit away
// from being removed as redundant-looking — and note the `else` arm MOVES `img` out, which
// is what makes "the block's value is the only survivor" true in both arms.
let (display, width, height) = {
// Decompression-bomb limits + EXIF orientation, both in one place — see
// services::imaging for why neither may be skipped.
let img = crate::services::imaging::decode_oriented(original)?;
let (width, height) = (img.width(), img.height());
// Display: max 2048px for the diashow. Only DOWNSCALE — never upscale a smaller
// original (that adds bytes with no quality gain); re-encode it as JPEG as-is.
let display = if width > display_max || height > display_max {
img.resize(
display_max,
display_max,
image::imageops::FilterType::Lanczos3,
)
} else {
img
};
(display, width, height)
};
display
.save_with_format(display_path, image::ImageFormat::Jpeg)
.context("failed to save display")?;
// Preview: max 800px, derived from the DISPLAY, not from the original.
//
// Both derivatives used to resize the full-size decode independently, so a 8000x8000
// original paid for two full-size Lanczos passes and their intermediates — measured 520
// MiB peak even after the scoping fix above, which two concurrent workers cannot fit in a
// 1 GiB container. Chaining 8000 -> 2048 -> 800 makes the second pass operate on 2048px
// input, and the full-size buffer is already freed by the time it runs. Quality is not the
// trade-off here: a staged Lanczos3 downscale to 800px is visually indistinguishable from
// a single-step one (and is a standard technique for large ratios).
display
.resize(
preview_max,
preview_max,
image::imageops::FilterType::Lanczos3,
)
.save_with_format(preview_path, image::ImageFormat::Jpeg)
.context("failed to save preview")?;
drop(display);
let pixels = u64::from(width) * u64::from(height);
// If the original is PNG, try lossless compression in place — but only when its pixel count
// is inside the budget, and never for longer than OXIPNG_TIMEOUT. This is a best-effort size
// saving: declining it costs disk, while attempting it unbounded cost the whole container.
if mime_type == "image/png" {
if pixels <= CompressionWorker::OXIPNG_MAX_PIXELS {
let mut opts = oxipng::Options::from_preset(2);
opts.timeout = Some(CompressionWorker::OXIPNG_TIMEOUT);
let _ = oxipng::optimize(
&oxipng::InFile::Path(original.to_path_buf()),
&oxipng::OutFile::Path {
path: None,
preserve_attrs: true,
},
&opts,
);
} else {
tracing::info!(
%upload_id, pixels,
"skipping oxipng: above the pixel budget; the original is stored as uploaded"
);
}
}
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
/// Peak resident set of THIS process, in bytes, from `/proc/self/status`.
fn peak_rss_bytes() -> u64 {
let status = std::fs::read_to_string("/proc/self/status").expect("procfs");
let line = status
.lines()
.find(|l| l.starts_with("VmHWM:"))
.expect("VmHWM");
let kb: u64 = line
.split_whitespace()
.nth(1)
.and_then(|v| v.parse().ok())
.expect("VmHWM value");
kb * 1024
}
/// Reset the kernel's peak-RSS watermark so the measurement covers only what follows.
/// Linux 4.0+; writing "5" to `clear_refs` resets `VmHWM` to the current RSS.
fn reset_peak_rss() {
let _ = std::fs::write("/proc/self/clear_refs", "5");
}
/// The pixel gate has to sit below what the axis limits allow, or it can never fire.
#[test]
fn the_oxipng_gate_is_reachable_within_the_decode_limits() {
const _: () = {
// imaging::decode_limits permits 12_000 x 12_000 = 144 MP. A gate above that would
// never skip anything.
assert!(CompressionWorker::OXIPNG_MAX_PIXELS < 12_000 * 12_000);
// ...and it must stay above a 48 MP camera, so real photos still get optimised.
assert!(CompressionWorker::OXIPNG_MAX_PIXELS >= 8_000_000);
};
}
/// The heavy-image gate has to classify the two cases the way the sizing assumed:
/// an ordinary phone photo must NOT serialise, and the giant must.
#[test]
fn the_heavy_gate_separates_a_phone_photo_from_a_giant() {
let dir = std::env::temp_dir().join(format!("es-heavy-{}", std::process::id()));
std::fs::create_dir_all(&dir).unwrap();
// 12 MP, the shape of a default phone capture.
let ordinary = dir.join("ordinary.jpg");
image::RgbImage::new(4032, 3024).save(&ordinary).unwrap();
let ordinary_peak = crate::services::imaging::estimated_processing_peak_bytes(
&ordinary,
CompressionWorker::DISPLAY_MAX_EDGE,
)
.expect("header readable");
assert!(
ordinary_peak <= CompressionWorker::HEAVY_IMAGE_BYTES,
"a 12 MP photo estimated at {} MiB would serialise the common path",
ordinary_peak / 1048576
);
// The 64 MP RGBA case that measured ~516 MiB peak.
let giant = dir.join("giant.png");
image::RgbaImage::new(8000, 8000).save(&giant).unwrap();
let giant_peak = crate::services::imaging::estimated_processing_peak_bytes(
&giant,
CompressionWorker::DISPLAY_MAX_EDGE,
)
.expect("header readable");
assert!(
giant_peak > CompressionWorker::HEAVY_IMAGE_BYTES,
"an 8000x8000 RGBA original estimated at only {} MiB would be allowed to run \
concurrently with another one — 2x its real ~516 MiB peak does not fit in 1 GiB",
giant_peak / 1048576
);
// The estimate must also be in the right ballpark, not merely on the right side of the
// threshold: 244 MiB decode + 262 MiB f32 resize intermediate.
assert!(
(400..700).contains(&(giant_peak / 1048576)),
"estimate {} MiB is far from the measured ~516 MiB peak",
giant_peak / 1048576
);
let _ = std::fs::remove_dir_all(&dir);
}
/// The OOM that took the container down, measured rather than argued.
///
/// An 8000x8000 RGBA PNG passes admission: 256,000,000 bytes is just under the 256 MiB
/// `max_alloc`, and smooth content is a few MB on disk, far under any size cap. The old
/// code kept that ~244 MiB decode alive across an unbounded, multi-threaded oxipng run and
/// peaked at ~1250 MiB — inside a 1 GiB cgroup. Being SIGKILLed there is not a blip: the
/// row was already committed, so the boot backfill replayed the identical workload on every
/// restart.
///
/// `#[ignore]` because it allocates ~250 MiB and takes a few seconds. Run explicitly:
/// cargo test --release oom -- --ignored --nocapture --test-threads=1
/// It must run ALONE — `VmHWM` is per process, so a concurrent test would pollute it.
#[test]
#[ignore = "heavy: allocates ~250 MiB; run with --ignored --test-threads=1"]
fn a_large_png_stays_far_below_the_container_limit() {
const EDGE: u32 = 8_000;
let dir = std::env::temp_dir().join(format!("es-oom-{}", std::process::id()));
std::fs::create_dir_all(&dir).unwrap();
let original = dir.join("big.png");
// Smooth gradient: ~244 MiB decoded, a couple of MB on disk. That gap is the whole
// point — file size tells you nothing about what a PNG costs to process.
{
let mut buf = image::RgbaImage::new(EDGE, EDGE);
for (x, y, px) in buf.enumerate_pixels_mut() {
*px = image::Rgba([(x >> 5) as u8, (y >> 5) as u8, ((x + y) >> 6) as u8, 255]);
}
buf.save(&original).unwrap();
}
// Everything above is fixture setup, not the code under test.
reset_peak_rss();
let before = peak_rss_bytes();
write_image_derivatives(
Uuid::new_v4(),
&original,
"image/png",
&dir.join("preview.jpg"),
&dir.join("display.jpg"),
)
.expect("derivatives");
let peak = peak_rss_bytes();
let on_disk = std::fs::metadata(&original).unwrap().len();
eprintln!(
"input {:.2} MiB on disk ({EDGE}x{EDGE}); peak RSS {:.0} MiB (was {:.0} MiB before)",
on_disk as f64 / 1048576.0,
peak as f64 / 1048576.0,
before as f64 / 1048576.0
);
assert!(dir.join("preview.jpg").exists() && dir.join("display.jpg").exists());
// The container gets 1 GiB and runs two of these concurrently. 600 MiB is a generous
// ceiling that the old code (~1250 MiB) could not have met.
assert!(
peak < 600 * 1024 * 1024,
"peak RSS {} MiB — the decode is being held across oxipng again, or the pixel \
gate stopped firing",
peak / 1048576
);
let _ = std::fs::remove_dir_all(&dir);
}
}

34
backend/src/services/imaging.rs
View File

@@ -118,6 +118,40 @@ fn decoder_within_budget(path: &Path) -> Result<impl image::ImageDecoder> {
Ok(decoder)
}
/// Rough peak heap an image will cost to turn into derivatives, read from the HEADER only —
/// no pixels are decoded. `None` when the header can't be read or the image is over budget
/// (the caller is about to fail on it anyway).
///
/// Two terms, and the second is the one that surprises:
///
/// - the decoded buffer, `width * height * channels`; and
/// - the resize intermediate. `image`'s Lanczos3 path accumulates in `f32`, so the buffer
/// between the horizontal and vertical passes is `new_width * old_height * 4 channels * 4
/// bytes` — 16 bytes per pixel-row-slot, not the 4 the output uses. For an 8000x8000
/// original that is 262 MiB on top of a 244 MiB decode, measured. It is bigger than the
/// decode for any tall image, which is why "the decode is bounded by max_alloc" was never
/// the whole story.
///
/// Used to decide whether an image is heavy enough to need exclusive use of the box's memory
/// headroom, NOT to reject anything.
pub fn estimated_processing_peak_bytes(path: &Path, display_edge: u32) -> Option<u64> {
let decoder = decoder_within_budget(path).ok()?;
let (width, height) = decoder.dimensions();
let decoded = decoder.total_bytes();
// Aspect-preserving fit into `display_edge`, matching DynamicImage::resize. No downscale
// means no intermediate at all.
let intermediate = if width > display_edge || height > display_edge {
let ratio = f64::from(display_edge) / f64::from(width.max(height));
let new_width = (f64::from(width) * ratio).round().max(1.0) as u64;
new_width * u64::from(height) * 16
} else {
0
};
Some(decoded.saturating_add(intermediate))
}
/// Megapixels an image would decode to, or `None` if its header can't be read. Used only
/// to put a concrete number in the message the guest sees.
pub fn megapixels(path: &Path) -> Option<f64> {

140
backend/src/services/maintenance.rs
View File

@@ -120,6 +120,19 @@ pub async fn startup_recovery(pool: &PgPool) {
}
}
/// How long a file in `originals/` may exist without a database row before it is treated as
/// abandoned.
///
/// This window is the ONLY thing making the sweep safe, because the upload handler renames the
/// temp file into its final path BEFORE committing the row: for a short moment a perfectly
/// healthy upload legitimately looks exactly like an orphan. Six hours is far beyond any live
/// request (a 576 MiB body over a bad venue uplink is minutes, and the request itself is bounded
/// by the reverse proxy) while still reclaiming the leak inside a single event.
///
/// DO NOT SHORTEN THIS to make a test faster — a value below the longest possible in-flight
/// upload deletes photos out from under the request that is committing them.
const ORPHAN_UPLOAD_RETENTION_HOURS: u64 = 6;
/// Spawns a background task that periodically:
/// - deletes session rows whose `expires_at` is more than a day in the past
/// - prunes the in-memory rate-limiter HashMap of empty windows
@@ -176,6 +189,13 @@ async fn periodic_loop(
cleanup_sessions(&pool).await;
cleanup_deleted_media(&pool, &media_path).await;
sweep_orphan_upload_temps(&media_path).await;
// Runs AFTER the .tmp sweep, and covers the class that one structurally cannot see:
// an original that was renamed to its final name but whose transaction never
// committed. Those have no row, so `cleanup_deleted_media` (row-driven) can never
// find them, and `sweep_orphan_upload_temps` skips them because they no longer end
// in `.tmp` — they were permanently unowned, silently shrinking the free disk that
// `compute_storage_quota` divides among guests.
sweep_orphan_originals(&pool, &media_path).await;
rate_limiter.prune();
sse_tickets.prune();
}
@@ -379,6 +399,126 @@ async fn cleanup_sessions(pool: &PgPool) {
}
}
/// Reclaim files in `originals/` that no upload row references.
///
/// The backstop behind [`TempFileGuard`](crate::handlers::upload). The guard covers the
/// process that is running; this covers the process that was killed — a SIGKILL, an OOM, or a
/// power cut leaves whatever bytes had been written with no `Drop` to reclaim them, and those
/// files are then permanently invisible: they have no row, so `cleanup_deleted_media` (which is
/// row-driven) can never see them, and they are not counted against any quota while still
/// consuming the free disk that `compute_storage_quota` divides among guests. On a single box
/// where all three volumes share a filesystem, that ends with Postgres unable to write WAL.
///
/// Two classes:
/// - `*.tmp` — an upload that never got as far as being renamed. Always safe past the window.
/// - everything else — a final-named original whose commit never happened.
async fn sweep_orphan_originals(pool: &PgPool, media_path: &std::path::Path) {
let originals = media_path.join("originals");
let cutoff = Duration::from_secs(ORPHAN_UPLOAD_RETENTION_HOURS * 3600);
// originals/{event_slug}/{uuid}.{ext} — one level of per-event directories.
let mut event_dirs = match tokio::fs::read_dir(&originals).await {
Ok(rd) => rd,
// Nothing uploaded yet; the directory is created lazily by the upload handler.
Err(_) => return,
};
let mut candidates: Vec<(String, std::path::PathBuf)> = Vec::new();
let mut temps_removed = 0u32;
while let Ok(Some(event_dir)) = event_dirs.next_entry().await {
if !event_dir
.file_type()
.await
.map(|t| t.is_dir())
.unwrap_or(false)
{
continue;
}
let slug = event_dir.file_name().to_string_lossy().to_string();
let Ok(mut files) = tokio::fs::read_dir(event_dir.path()).await else {
continue;
};
while let Ok(Some(entry)) = files.next_entry().await {
let Ok(meta) = entry.metadata().await else {
continue;
};
if !meta.is_file() {
continue;
}
// Too young to judge: an upload committing RIGHT NOW is indistinguishable from an
// orphan, because the rename precedes the commit.
let recent = meta
.modified()
.ok()
.and_then(|m| m.elapsed().ok())
.is_none_or(|age| age < cutoff);
if recent {
continue;
}
let name = entry.file_name().to_string_lossy().to_string();
if name.ends_with(".tmp") {
// A `.tmp` never has a row by construction — no DB check needed.
if tokio::fs::remove_file(entry.path()).await.is_ok() {
temps_removed += 1;
}
continue;
}
candidates.push((format!("originals/{slug}/{name}"), entry.path()));
}
}
if temps_removed > 0 {
tracing::warn!(
"reclaimed {temps_removed} abandoned upload temp file(s) older than \
{ORPHAN_UPLOAD_RETENTION_HOURS}h"
);
}
if candidates.is_empty() {
return;
}
// One query per batch, not one per file: a backlog of thousands of orphans must not turn
// into thousands of round trips on an hourly timer.
let mut orphans_removed = 0u32;
for chunk in candidates.chunks(500) {
let paths: Vec<String> = chunk.iter().map(|(rel, _)| rel.clone()).collect();
// NO `deleted_at IS NULL` FILTER HERE. A soft-deleted row still points at its file
// during its retention window, and reclaiming that file is `cleanup_deleted_media`'s
// job — filtering here would race the two sweeps and destroy the exact files the
// recovery window exists to preserve.
let unreferenced: Result<Vec<(String,)>, _> = sqlx::query_as(
"SELECT p FROM unnest($1::text[]) AS p
WHERE NOT EXISTS (SELECT 1 FROM upload u WHERE u.original_path = p)",
)
.bind(&paths)
.fetch_all(pool)
.await;
let unreferenced = match unreferenced {
Ok(rows) => rows,
Err(e) => {
tracing::warn!(error = ?e, "orphan-original sweep query failed");
return;
}
};
for (rel,) in unreferenced {
if let Some((_, abs)) = chunk.iter().find(|(r, _)| *r == rel)
&& tokio::fs::remove_file(abs).await.is_ok()
{
orphans_removed += 1;
}
}
}
if orphans_removed > 0 {
tracing::warn!(
"reclaimed {orphans_removed} original(s) with no upload row, older than \
{ORPHAN_UPLOAD_RETENTION_HOURS}h"
);
}
}
#[cfg(test)]
mod tests {
use super::*;

157
backend/src/services/sse_tickets.rs
View File

@@ -13,6 +13,16 @@ use rand::Rng;
/// stream open. Tickets are consumed on use and expire after `TTL`.
const TTL: Duration = Duration::from_secs(30);
/// Ceiling on outstanding tickets across the whole process.
///
/// Not really about the bytes (~120 each) — about `issue` having had no bound of any kind.
/// Sized well above a real event: ~1000 concurrent clients each holding one live 30 s ticket.
const MAX_TICKETS: usize = 4096;
/// Live tickets one session may hold. Above 1 because two tabs sharing a token open their
/// EventSources concurrently; 4 absorbs that without letting a reconnect loop accumulate.
const MAX_TICKETS_PER_SESSION: usize = 4;
#[derive(Clone)]
pub struct SseTicketStore {
inner: Arc<Mutex<HashMap<String, Entry>>>,
@@ -39,9 +49,47 @@ impl SseTicketStore {
}
/// Mint a new ticket bound to the caller's session (identified by token hash).
pub fn issue(&self, token_hash: String) -> String {
///
/// `None` when the store is at capacity — the caller should answer 503, not evict.
///
/// Three bounds, because `issue` had none: no size cap, no per-caller cap, and no rate
/// limit on the endpoint, while `prune` ran only hourly against a 30-second TTL. So any
/// authenticated session could loop the endpoint and grow the map for an hour.
pub fn issue(&self, token_hash: String) -> Option<String> {
let ticket = random_ticket();
let mut map = self.inner.lock().unwrap();
// Prune on issue rather than only hourly. This alone changes the bound from "tickets
// minted since the last maintenance tick" to "tickets live at once", which is what the
// 30 s TTL was always meant to express.
map.retain(|_, e| e.issued_at.elapsed() <= TTL);
// Cap the caller's own outstanding tickets, evicting their oldest. NOT one-per-session:
// two tabs sharing a token open their EventSources concurrently, and having tab B
// invalidate tab A's unconsumed ticket looks exactly like a flaky SSE connection.
let mut mine: Vec<(String, Instant)> = map
.iter()
.filter(|(_, e)| e.token_hash == token_hash)
.map(|(k, e)| (k.clone(), e.issued_at))
.collect();
if mine.len() >= MAX_TICKETS_PER_SESSION {
mine.sort_by_key(|(_, issued)| *issued);
for (key, _) in mine.iter().take(mine.len() - MAX_TICKETS_PER_SESSION + 1) {
map.remove(key);
}
}
// At capacity, REFUSE — never evict a stranger's ticket. Evicting would let one
// misbehaving client deny SSE to the whole venue, which is worse than failing the
// request that hit the ceiling.
if map.len() >= MAX_TICKETS {
tracing::warn!(
outstanding = map.len(),
"SSE ticket store at capacity; refusing to mint"
);
return None;
}
map.insert(
ticket.clone(),
Entry {
@@ -49,7 +97,7 @@ impl SseTicketStore {
issued_at: Instant::now(),
},
);
ticket
Some(ticket)
}
/// Consume a ticket. Returns `Some(token_hash)` if the ticket exists and is
@@ -84,10 +132,16 @@ fn random_ticket() -> String {
mod tests {
use super::*;
/// `issue` now returns `Option`; in every test below the store is far from capacity, so an
/// `expect` here documents that refusing is exceptional rather than routine.
fn issue(store: &SseTicketStore, hash: &str) -> String {
store.issue(hash.into()).expect("store has capacity")
}
#[test]
fn issue_then_consume_returns_the_hash_exactly_once() {
let store = SseTicketStore::new();
let ticket = store.issue("hash-1".into());
let ticket = issue(&store, "hash-1");
assert_eq!(store.consume(&ticket).as_deref(), Some("hash-1"));
// Single-use: a replay of the same ticket is rejected.
assert_eq!(
@@ -106,8 +160,8 @@ mod tests {
#[test]
fn issued_tickets_are_unique_and_hex() {
let store = SseTicketStore::new();
let a = store.issue("h".into());
let b = store.issue("h".into());
let a = issue(&store, "h");
let b = issue(&store, "h");
assert_ne!(a, b, "each ticket must be unique");
assert_eq!(a.len(), 48, "24 random bytes → 48 hex chars");
assert!(a.chars().all(|c| c.is_ascii_hexdigit()));
@@ -116,29 +170,104 @@ mod tests {
#[test]
fn fresh_ticket_survives_prune() {
let store = SseTicketStore::new();
let ticket = store.issue("h".into());
let ticket = issue(&store, "h");
store.prune(); // not expired → kept
assert_eq!(store.consume(&ticket).as_deref(), Some("h"));
}
#[test]
fn expired_ticket_consumes_to_none() {
// Construct an entry that is already past the TTL and confirm consume() rejects it.
let store = SseTicketStore::new();
let stale = "stale-ticket".to_string();
/// Build an entry that is already past the TTL.
fn insert_stale(store: &SseTicketStore, key: &str, token_hash: &str) {
store.inner.lock().unwrap().insert(
stale.clone(),
key.to_string(),
Entry {
token_hash: "h".into(),
token_hash: token_hash.into(),
issued_at: Instant::now()
.checked_sub(TTL + Duration::from_secs(1))
.expect("host uptime should exceed the ticket TTL"),
},
);
}
#[test]
fn expired_ticket_consumes_to_none() {
let store = SseTicketStore::new();
insert_stale(&store, "stale-ticket", "h");
assert_eq!(
store.consume(&stale),
store.consume("stale-ticket"),
None,
"an expired ticket must not authenticate"
);
}
/// The TTL is 30 s but `prune` only ran hourly, so the map was really bounded by "tickets
/// minted in the last hour" — which is unbounded for a client in a loop.
#[test]
fn issuing_prunes_expired_entries() {
let store = SseTicketStore::new();
insert_stale(&store, "stale-a", "someone-else");
insert_stale(&store, "stale-b", "someone-else");
issue(&store, "h");
assert_eq!(
store.inner.lock().unwrap().len(),
1,
"issue must reclaim expired slots, not merely add to them"
);
}
/// Two tabs sharing a token is normal, so the per-session cap must be above 1 — but a
/// reconnect loop must not accumulate. The caller's OWN oldest is what gets evicted.
#[test]
fn a_session_is_capped_and_evicts_only_its_own_oldest() {
let store = SseTicketStore::new();
let stranger = issue(&store, "other-session");
let mut mine: Vec<String> = Vec::new();
for _ in 0..MAX_TICKETS_PER_SESSION + 2 {
mine.push(issue(&store, "mine"));
}
let live = mine
.iter()
.filter(|t| store.inner.lock().unwrap().contains_key(*t))
.count();
assert_eq!(live, MAX_TICKETS_PER_SESSION, "one session, bounded");
assert!(
store.inner.lock().unwrap().contains_key(&mine[mine.len() - 1]),
"the newest ticket is the one the caller is about to use"
);
assert_eq!(
store.consume(&stranger).as_deref(),
Some("other-session"),
"another session's ticket must survive — evicting it would let one client deny \
SSE to the venue"
);
}
/// At capacity the store REFUSES rather than evicting a stranger. Refusing fails the one
/// request that hit the ceiling; evicting would break an unrelated client's live stream.
#[test]
fn at_capacity_the_store_refuses_instead_of_evicting() {
let store = SseTicketStore::new();
{
let mut map = store.inner.lock().unwrap();
for i in 0..MAX_TICKETS {
map.insert(
format!("filler-{i}"),
Entry {
token_hash: format!("session-{i}"),
issued_at: Instant::now(),
},
);
}
}
assert_eq!(
store.issue("newcomer".into()),
None,
"a full store must refuse, so the caller can answer 503"
);
assert!(
store.inner.lock().unwrap().contains_key("filler-0"),
"no existing ticket may be sacrificed to make room"
);
}
}

113
backend/src/services/video.rs
View File

@@ -76,7 +76,7 @@ pub async fn extract_poster_frame(src: &Path, dest: &Path, width: u32) -> Result
/// Run one ffmpeg attempt. A non-zero exit is NOT an error here — the artifact check above is the
/// authority, and a corrupt input that fails at 1 s may still yield a frame at 0.
async fn run_ffmpeg(src: &Path, dest: &Path, width: u32, seek: &str) -> Result<()> {
let mut child = tokio::process::Command::new("ffmpeg")
let child = tokio::process::Command::new("ffmpeg")
.args([
// BEFORE -i: an input-side seek. See SEEK_POSITIONS.
"-ss",
@@ -90,24 +90,56 @@ async fn run_ffmpeg(src: &Path, dest: &Path, width: u32, seek: &str) -> Result<(
"-y",
dest.to_str().unwrap_or_default(),
])
.stdout(std::process::Stdio::piped())
// ffmpeg writes the poster to `dest` itself; nothing here ever reads stdout, so
// giving it a pipe only created something that could fill.
.stdout(std::process::Stdio::null())
// stderr IS piped — it is the only diagnostic when a clip yields no frame — but it
// must be DRAINED, which is the whole point of `wait_with_output` below.
.stderr(std::process::Stdio::piped())
.kill_on_drop(true)
.spawn()
.context("failed to spawn ffmpeg")?;
match tokio::time::timeout(FFMPEG_TIMEOUT, child.wait()).await {
Ok(res) => {
res.context("ffmpeg wait failed")?;
}
Err(_) => {
let _ = child.kill().await;
anyhow::bail!("ffmpeg timed out after {}s", FFMPEG_TIMEOUT.as_secs());
}
// `wait_with_output`, NOT `wait`. ffmpeg is verbose on stderr (banner, stream info,
// per-frame progress) and `wait()` reads neither pipe — so once the ~64 KiB pipe buffer
// filled, ffmpeg blocked writing, `wait()` never returned, and the call burned the full
// timeout. That is not merely slow: the timeout is an `Err`, so after 2 seek positions x
// 3 compression attempts the caller soft-deletes a perfectly playable video for a
// poster-frame failure. `wait_with_output` polls the pipe and the exit status together.
//
// It also CONSUMES the child, so the explicit `child.kill()` that used to sit on the
// timeout arm cannot exist here — and is not needed: `kill_on_drop(true)` is set above,
// and dropping the future on timeout drops the child with it.
let out = match tokio::time::timeout(FFMPEG_TIMEOUT, child.wait_with_output()).await {
Ok(res) => res.context("ffmpeg wait failed")?,
Err(_) => anyhow::bail!("ffmpeg timed out after {}s", FFMPEG_TIMEOUT.as_secs()),
};
// A non-zero exit is not an error (see the doc comment) — the artifact check in
// `extract_poster_frame` is the authority. Log the tail so a systematically failing
// format is diagnosable without turning it into data loss.
if !out.status.success() {
tracing::debug!(
seek,
status = ?out.status,
stderr = %tail_lines(&out.stderr, 10),
"ffmpeg exited non-zero; the artifact check decides"
);
}
Ok(())
}
/// Last `n` lines of a child's stderr, lossily decoded.
///
/// Bounded on purpose: ffmpeg's stderr is unbounded, and the reason we now drain it is that
/// unbounded output used to be a hazard. Emitting all of it into a log line — into container
/// logs that are themselves size-capped — would just move the problem.
fn tail_lines(bytes: &[u8], n: usize) -> String {
let text = String::from_utf8_lossy(bytes);
let lines: Vec<&str> = text.lines().filter(|l| !l.trim().is_empty()).collect();
lines[lines.len().saturating_sub(n)..].join(" | ")
}
#[cfg(test)]
mod tests {
use super::*;
@@ -167,4 +199,65 @@ mod tests {
);
let _ = std::fs::remove_dir_all(&dir);
}
#[test]
fn the_stderr_tail_is_bounded_and_survives_invalid_utf8() {
let noisy: Vec<u8> = (0..500)
.map(|i| format!("line {i}\n"))
.collect::<String>()
.into_bytes();
let got = tail_lines(&noisy, 3);
assert_eq!(got, "line 497 | line 498 | line 499");
// ffmpeg emits filenames verbatim, so its stderr is not guaranteed to be UTF-8.
assert_eq!(tail_lines(&[b'o', b'k', 0xff], 5), "ok\u{fffd}");
assert_eq!(tail_lines(b"", 5), "");
}
/// A real extraction must finish in a small fraction of `FFMPEG_TIMEOUT`.
///
/// Wall-clock is the ONLY observable of the bug this guards: piping stderr and then
/// calling `wait()` (which drains nothing) blocks ffmpeg on a full pipe buffer until the
/// timeout fires, and the timeout is an `Err`, so the upload is soft-deleted. The
/// assertion is deliberately on elapsed time, not on the exit status.
///
/// Honest limitation: our fixture is quiet enough not to fill a 64 KiB pipe on its own,
/// so this catches a regression to `wait()` only in combination with a verbose input. It
/// is still worth pinning — a reverted drain plus any chatty clip is data loss.
#[tokio::test]
async fn a_real_clip_yields_a_poster_well_inside_the_timeout() {
if tokio::process::Command::new("ffmpeg")
.arg("-version")
.stdout(std::process::Stdio::null())
.stderr(std::process::Stdio::null())
.status()
.await
.is_err()
{
eprintln!("skipping: ffmpeg not on PATH");
return;
}
let src = Path::new("../e2e/fixtures/media/sample.mp4");
if !src.exists() {
eprintln!("skipping: {} missing", src.display());
return;
}
let dir = std::env::temp_dir().join(format!("es-video-ok-{}", std::process::id()));
std::fs::create_dir_all(&dir).unwrap();
let dest = dir.join("poster.jpg");
let started = std::time::Instant::now();
let got = extract_poster_frame(src, &dest, 400).await;
let elapsed = started.elapsed();
assert!(matches!(got, Ok(true)), "expected a poster, got {got:?}");
assert!(dest.metadata().unwrap().len() > 0);
assert!(
elapsed < FFMPEG_TIMEOUT / 4,
"extraction took {elapsed:?}; a drained stderr finishes in well under \
{FFMPEG_TIMEOUT:?} — this is the pipe-deadlock regression guard"
);
let _ = std::fs::remove_dir_all(&dir);
}
}