cacf616c2d
The compression job is handed to the worker by a bare `tokio::spawn` that runs AFTER `tx.commit()`, and the commit is a suspension point. If the guest walks out of range inside it, Postgres applies the COMMIT and axum drops the future before the spawn is reached: the row lands durably at `compression_status = 'pending'` with no job behind it. Nothing looked at it again. `startup_recovery` rescues only `'processing'`, and `backfill_stale_derivatives` — whose predicate WOULD match — runs once at boot and never on a timer. The photo keeps its feed entry, never gets a derivative, forces every viewer to pull the full original instead, and is skipped by the diashow for the rest of the event. Silent, and permanent. This is the same hazard `TempFileGuard` already covers for the file, reached one line later, so it gets the same answer: an `EnqueueGuard` armed before the commit and stood down only once the job is with the worker. `Drop` runs on cancellation, which closes the window — and it covers the indeterminate-commit error path too, which returns Err on a row that may well be live. Queueing a job for a row that did not commit is harmless: the status update matches zero rows and the task retires. A guard cannot survive a SIGKILL, so `requeue_stuck_pending` sweeps for the same state at boot and every ten minutes. The grace window is what makes it safe to re-enter the live path: a task flips the row to `'processing'` as its FIRST act after taking a permit, so a row still `'pending'` ten minutes on is lost rather than merely queued. The event simulation peaked at 120 queued with a p99 of 97s while being fed thirty times a real event's arrival rate, so the margin is wide. Hit 1 upload in 932 during that run. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
941 lines
46 KiB
Rust
941 lines
46 KiB
Rust
use std::path::{Path, PathBuf};
|
|
use std::sync::Arc;
|
|
use std::sync::atomic::{AtomicU64, Ordering};
|
|
|
|
use anyhow::{Context, Result};
|
|
use sqlx::PgPool;
|
|
use tokio::sync::{Semaphore, broadcast};
|
|
use uuid::Uuid;
|
|
|
|
use crate::models::upload::Upload;
|
|
use crate::state::SseEvent;
|
|
|
|
#[derive(Clone)]
|
|
pub struct CompressionWorker {
|
|
semaphore: Arc<Semaphore>,
|
|
pool: PgPool,
|
|
media_path: PathBuf,
|
|
sse_tx: broadcast::Sender<SseEvent>,
|
|
/// Bumped whenever the underlying data is reset out from under in-flight work (only the e2e
|
|
/// TRUNCATE does this today). A task captures the value at spawn and abandons itself if it has
|
|
/// changed by the time it runs — see `process`.
|
|
generation: Arc<AtomicU64>,
|
|
}
|
|
|
|
impl CompressionWorker {
|
|
pub fn new(
|
|
pool: PgPool,
|
|
media_path: PathBuf,
|
|
concurrency: usize,
|
|
sse_tx: broadcast::Sender<SseEvent>,
|
|
) -> Self {
|
|
Self {
|
|
semaphore: Arc::new(Semaphore::new(concurrency)),
|
|
pool,
|
|
media_path,
|
|
sse_tx,
|
|
generation: Arc::new(AtomicU64::new(0)),
|
|
}
|
|
}
|
|
|
|
/// Invalidate all in-flight and queued compression work. Called by the e2e TRUNCATE endpoint:
|
|
/// truncating deletes the upload rows and wipes `media/`, so a worker that was queued on the
|
|
/// semaphore when the wipe happened would otherwise wake in the NEXT test, fail to find its
|
|
/// file, and broadcast `upload-error` / `upload-deleted` into that test's live SSE stream —
|
|
/// corrupting any test that asserts on toasts or feed contents. Bumping the generation makes
|
|
/// those stale tasks return silently instead. A no-op in production (never called there).
|
|
pub fn bump_generation(&self) {
|
|
self.generation.fetch_add(1, Ordering::SeqCst);
|
|
}
|
|
|
|
/// How many times `do_process` is attempted before an upload is given up on. The
|
|
/// give-up path is user-visible (the photo disappears), so transient infrastructure
|
|
/// errors must not reach it.
|
|
const MAX_PROCESS_ATTEMPTS: u32 = 3;
|
|
|
|
/// Revision of the image-derivative pipeline. Bump this whenever a change makes existing
|
|
/// previews/displays wrong, so `backfill_stale_derivatives` regenerates them once on the
|
|
/// 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();
|
|
let born_at = worker.generation.load(Ordering::SeqCst);
|
|
tokio::spawn(async move {
|
|
let _permit = worker.semaphore.acquire().await;
|
|
// The data this task was queued against may have been reset while it waited for a permit
|
|
// (e2e TRUNCATE). If so, its file and row are gone; doing anything — including
|
|
// broadcasting a failure — would leak into an unrelated test. Abandon quietly.
|
|
if worker.generation.load(Ordering::SeqCst) != born_at {
|
|
return;
|
|
}
|
|
// Retry before giving up. Most failures here are transient and self-clearing —
|
|
// an ENOSPC spike while several guests upload at once, a momentary DB-pool
|
|
// exhaustion, a panic inside the image codec — and the give-up path is
|
|
// user-visible data loss, so it is worth a few seconds to avoid entering it.
|
|
//
|
|
// But only for failures that CAN clear. An image that exceeds the decode budget,
|
|
// is corrupt, or is in an unsupported format fails identically on every attempt,
|
|
// so retrying it just burns 2s + 4s of backoff and writes three near-identical
|
|
// warnings before reaching the same conclusion. Give up on those immediately.
|
|
let mut attempt = 1u32;
|
|
let outcome = loop {
|
|
match worker
|
|
// Charge the lifetime budget once per episode, on the first attempt only.
|
|
.do_process(upload_id, &original_path, &mime_type, attempt == 1)
|
|
.await
|
|
{
|
|
Ok(v) => break Ok(v),
|
|
Err(e)
|
|
if attempt < Self::MAX_PROCESS_ATTEMPTS
|
|
&& !crate::services::imaging::is_permanent_image_error(&e)
|
|
&& !crate::services::imaging::is_storage_full_error(&e) =>
|
|
{
|
|
tracing::warn!(
|
|
error = ?e, %upload_id, attempt,
|
|
"compression attempt failed; retrying"
|
|
);
|
|
tokio::time::sleep(std::time::Duration::from_secs(2u64.pow(attempt))).await;
|
|
attempt += 1;
|
|
// The data may have been reset while we slept (e2e TRUNCATE).
|
|
if worker.generation.load(Ordering::SeqCst) != born_at {
|
|
return;
|
|
}
|
|
}
|
|
Err(e) => break Err(e),
|
|
}
|
|
};
|
|
|
|
match outcome {
|
|
Ok(_) => {
|
|
tracing::info!("compression completed for upload {upload_id}");
|
|
let _ = worker.sse_tx.send(SseEvent {
|
|
event_type: "upload-processed".to_string(),
|
|
data: serde_json::json!({ "upload_id": upload_id }).to_string(),
|
|
});
|
|
}
|
|
Err(e) if crate::services::imaging::is_storage_full_error(&e) => {
|
|
// Out of disk. Keep the row AND the original — the opposite of the branch
|
|
// below, and for the same reason it retains the file: nothing here is the
|
|
// guest's fault and nothing about the photo is wrong.
|
|
//
|
|
// Soft-deleting on ENOSPC was strictly harmful. It refunded the quota while
|
|
// keeping the bytes, so it freed nothing, removed the photo from the feed
|
|
// seconds after a `201 Created`, and handed the guest the allowance to
|
|
// upload it again into the same full disk. Leaving the row live costs
|
|
// nothing instead: every client already falls back to the original when
|
|
// `preview_url` and `thumbnail_url` are NULL, so the photo stays visible —
|
|
// just uncompressed — and `backfill_stale_derivatives` regenerates the
|
|
// derivatives on the next start, once there is room for them.
|
|
tracing::error!(
|
|
%upload_id,
|
|
"compression failed: the media filesystem is out of space. The upload is \
|
|
kept and served from its original; free disk space and restart to \
|
|
regenerate derivatives: {e:#}"
|
|
);
|
|
let _ = Upload::set_compression_status(&worker.pool, upload_id, "failed").await;
|
|
// Not an "error" event: nothing was lost and there is nothing for the guest
|
|
// to act on. Clients treat this purely as "refetch me", which is what makes
|
|
// the card appear with its original as the image source.
|
|
let _ = worker.sse_tx.send(SseEvent {
|
|
event_type: "upload-processed".to_string(),
|
|
data: serde_json::json!({ "upload_id": upload_id }).to_string(),
|
|
});
|
|
}
|
|
Err(e) => {
|
|
tracing::error!(
|
|
"compression failed for upload {upload_id} after {attempt} attempt(s): {e:#}"
|
|
);
|
|
// KEEP THE ROW. This used to soft-delete, which made a derivative failure
|
|
// indistinguishable — to the guest — from their photo being deleted: they
|
|
// got a `201 Created`, watched the card appear, and then watched it vanish.
|
|
// The row left `v_feed`, `find_visible_media` and BOTH keepsake archives,
|
|
// so the photo was gone from the product's core promise while its bytes sat
|
|
// on disk for 14 days waiting for a `cleanup_deleted_media` that nothing
|
|
// told anyone about. There is no host or admin screen listing compression
|
|
// failures, so recovery meant hand-written SQL that also had to re-add the
|
|
// refunded quota bytes. Against "0 lost uploads", that was silent per-photo
|
|
// loss on any error the ENOSPC arm above doesn't catch — a HEIC that slipped
|
|
// the allowlist, a truncated frame, an ffmpeg hiccup, a pool blip.
|
|
//
|
|
// This is exactly what the ENOSPC arm already does and documents as correct:
|
|
// every client falls back to the original when `preview_url` and
|
|
// `thumbnail_url` are NULL, so the photo stays visible and downloadable —
|
|
// just uncompressed — and `backfill_stale_derivatives` retries it on the
|
|
// next boot, now bounded by `derivative_attempts` so a poisoned row cannot
|
|
// loop. The quota stays charged, which is correct: the bytes are still on
|
|
// disk and still the guest's.
|
|
let _ = Upload::set_compression_status(&worker.pool, upload_id, "failed").await;
|
|
tracing::warn!(
|
|
%upload_id,
|
|
path = %worker.media_path.join(&original_path).display(),
|
|
"derivatives failed; the upload is kept and served from its original"
|
|
);
|
|
// `upload-error` still fires so the uploader learns the photo will look
|
|
// uncompressed. `upload-deleted` deliberately does NOT — nothing was
|
|
// deleted, and evicting the card was the visible half of the data loss.
|
|
let _ = worker.sse_tx.send(SseEvent {
|
|
event_type: "upload-error".to_string(),
|
|
data: serde_json::json!({ "upload_id": upload_id, "error": e.to_string() })
|
|
.to_string(),
|
|
});
|
|
// Tell every client to refetch, so the card re-renders from the original
|
|
// instead of sitting on a stale "processing" placeholder forever.
|
|
let _ = worker.sse_tx.send(SseEvent {
|
|
event_type: "upload-processed".to_string(),
|
|
data: serde_json::json!({ "upload_id": upload_id }).to_string(),
|
|
});
|
|
}
|
|
}
|
|
});
|
|
}
|
|
|
|
/// `charge_lifetime_attempt` is true only for the FIRST `do_process` of a given
|
|
/// `process()` call, so the two budgets stay independent.
|
|
///
|
|
/// They were not. `MAX_PROCESS_ATTEMPTS` (in-request retries, 3) and
|
|
/// `MAX_DERIVATIVE_ATTEMPTS` (lifetime, 3) are equal, and every retry re-entered here and
|
|
/// charged the lifetime counter — so one request's three retries, six seconds apart,
|
|
/// exhausted the entire lifetime budget. A ten-second pool blip during the arrival burst
|
|
/// therefore stranded every photo whose worker was inside that window with no preview and no
|
|
/// display derivative, permanently, recoverable by nothing: the boot backfill re-selects them
|
|
/// and immediately gives up on the same exhausted counter.
|
|
///
|
|
/// The two exist to bound different things — "this request is flapping" versus "this INPUT is
|
|
/// poison" — and only the second should survive across requests.
|
|
async fn do_process(
|
|
&self,
|
|
upload_id: Uuid,
|
|
original_path: &str,
|
|
mime_type: &str,
|
|
charge_lifetime_attempt: bool,
|
|
) -> Result<()> {
|
|
Upload::set_compression_status(&self.pool, upload_id, "processing").await?;
|
|
|
|
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. Charging on
|
|
// the first attempt preserves that: a container-killing input never reaches a second.
|
|
let charged = if charge_lifetime_attempt {
|
|
Upload::begin_derivative_attempt(&self.pool, upload_id).await?
|
|
} else {
|
|
// Already charged for this episode. Re-read the row only to notice it vanished.
|
|
Upload::derivative_attempts(&self.pool, upload_id).await?
|
|
};
|
|
match charged {
|
|
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?;
|
|
Upload::set_preview_path(&self.pool, upload_id, &preview_rel).await?;
|
|
Upload::set_display_path(&self.pool, upload_id, &display_rel).await?;
|
|
Upload::set_derivatives_rev(&self.pool, upload_id, Self::DERIVATIVES_REV).await?;
|
|
tracing::info!("preview + display generated for upload {upload_id}");
|
|
} else if mime_type.starts_with("video/") {
|
|
// A missing poster must NOT fail the upload. `set_thumbnail_path` is only reached when
|
|
// a file really exists, so `thumbnail_path` stays NULL otherwise — which every consumer
|
|
// already handles (FeedListCard, VirtualFeed, LightboxModal are all null-safe).
|
|
//
|
|
// The `?` here used to hide the defect; making the check strict without also making
|
|
// this non-fatal would have been far worse than the bug. Every clip of a second or less
|
|
// would fail compression, exhaust its retries and be soft-deleted — a cosmetic defect
|
|
// turned into data loss, on exactly the mis-tap/Live-Photo clips guests produce most.
|
|
// Handling only the `Ok(None)` arm was not enough: the `?` on the call itself still
|
|
// routed every OTHER poster failure into the give-up path. `extract_poster_frame`
|
|
// returns `Err` when ffmpeg is missing from the image, when it hangs on a truncated
|
|
// `.mov` and trips FFMPEG_TIMEOUT, or when `thumbnails/` can't be created — and
|
|
// `set_thumbnail_path` returns `Err` on any DB blip. None of those say anything about
|
|
// the video itself, yet each one destroyed it. Confirmed live: on a box with no ffmpeg
|
|
// the spawn error propagated, exhausted all three attempts and soft-deleted the clip.
|
|
//
|
|
// Nothing about a video post depends on the poster — `get_original` serves the file
|
|
// byte-for-byte and the tile falls back to the video element — so no failure in this
|
|
// branch may fail the upload.
|
|
match self.generate_video_thumbnail(upload_id, &original).await {
|
|
Ok(Some(thumb_rel)) => {
|
|
match Upload::set_thumbnail_path(&self.pool, upload_id, &thumb_rel).await {
|
|
Ok(()) => tracing::info!("thumbnail generated for upload {upload_id}"),
|
|
Err(e) => tracing::warn!(
|
|
error = ?e, %upload_id,
|
|
"poster extracted but could not be recorded; the video keeps its own tile"
|
|
),
|
|
}
|
|
}
|
|
Ok(None) => {
|
|
tracing::warn!(
|
|
%upload_id,
|
|
"no poster frame could be extracted; the video keeps its own tile"
|
|
);
|
|
}
|
|
Err(e) => {
|
|
tracing::warn!(
|
|
error = ?e, %upload_id,
|
|
"poster extraction failed; the video keeps its own tile"
|
|
);
|
|
}
|
|
}
|
|
}
|
|
|
|
Upload::set_compression_status(&self.pool, upload_id, "done").await?;
|
|
Ok(())
|
|
}
|
|
|
|
/// Longest edge of the big-screen "display" derivative used by the diashow. Sized to be
|
|
/// sharp on 1080p/4K while staying bounded (a ~2048px JPEG decodes to ~16 MB — trivial
|
|
/// for any kiosk, unlike a raw multi-thousand-pixel original).
|
|
const DISPLAY_MAX_EDGE: u32 = 2048;
|
|
/// 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;
|
|
|
|
/// 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(
|
|
&self,
|
|
upload_id: Uuid,
|
|
original: &Path,
|
|
mime_type: &str,
|
|
) -> Result<(String, String)> {
|
|
let previews_dir = self.media_path.join("previews");
|
|
let displays_dir = self.media_path.join("displays");
|
|
tokio::fs::create_dir_all(&previews_dir).await?;
|
|
tokio::fs::create_dir_all(&displays_dir).await?;
|
|
|
|
let filename = format!("{upload_id}.jpg");
|
|
let preview_path = previews_dir.join(&filename);
|
|
let display_path = displays_dir.join(&filename);
|
|
let original = original.to_path_buf();
|
|
let mime_owned = mime_type.to_string();
|
|
|
|
// 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 > crate::services::imaging::HEAVY_IMAGE_BYTES => {
|
|
tracing::debug!(
|
|
%upload_id,
|
|
estimated_mib = bytes / (1024 * 1024),
|
|
"waiting for the heavy-image permit"
|
|
);
|
|
Some(
|
|
crate::services::imaging::HEAVY_IMAGE_PERMITS
|
|
.acquire()
|
|
.await,
|
|
)
|
|
}
|
|
_ => None,
|
|
};
|
|
|
|
// Run blocking image operations in a spawn_blocking task
|
|
tokio::task::spawn_blocking(move || {
|
|
write_image_derivatives(
|
|
upload_id,
|
|
&original,
|
|
&mime_owned,
|
|
&preview_path,
|
|
&display_path,
|
|
)
|
|
})
|
|
.await??;
|
|
|
|
Ok((
|
|
format!("previews/{filename}"),
|
|
format!("displays/{filename}"),
|
|
))
|
|
}
|
|
|
|
/// Regenerate image derivatives that an older pipeline produced. Fire-and-forget from
|
|
/// startup; picks up two cases, both of which leave the ORIGINAL untouched:
|
|
///
|
|
/// - uploads processed before the `display` derivative existed (preview but no
|
|
/// `display_path`), and
|
|
/// - uploads whose derivatives predate `DERIVATIVES_REV` — currently rev 1, which applies
|
|
/// the EXIF orientation. Everything generated before it is stored sideways for any
|
|
/// portrait phone photo.
|
|
///
|
|
/// 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.
|
|
/// How long an upload may sit at `'pending'` before the sweeper assumes its job was lost.
|
|
///
|
|
/// Must comfortably exceed the deepest real queue wait, or the sweeper re-enqueues photos
|
|
/// that are merely waiting their turn. The live path makes that easy to bound: a task's
|
|
/// FIRST action after taking a permit is to flip the row to `'processing'`, so a row still
|
|
/// `'pending'` after this long is either behind that much work or genuinely lost. The
|
|
/// event simulation peaked at 120 queued with a p99 of 97s while being fed thirty times a
|
|
/// real event's arrival rate, so ten minutes is a wide margin rather than a tight one.
|
|
const PENDING_GRACE: chrono::Duration = chrono::Duration::minutes(10);
|
|
|
|
/// Re-queue uploads whose compression job was lost between the commit and the spawn.
|
|
///
|
|
/// `EnqueueGuard` closes that window for a cancelled request, but nothing in-process
|
|
/// survives a SIGKILL or an OOM kill, and `startup_recovery` only rescues `'processing'`
|
|
/// — so a row orphaned at `'pending'` had no path back at all. Without this, the failure is
|
|
/// silent and permanent: the photo keeps its feed entry, never gets a derivative, and is
|
|
/// dropped from the diashow for the rest of the event.
|
|
///
|
|
/// Re-entering the live `process` path (rather than regenerating derivatives inline, as
|
|
/// `backfill_stale_derivatives` does) is deliberate: it reuses the status transitions, the
|
|
/// retry ladder and the SSE broadcast, so a rescued photo behaves exactly like one that was
|
|
/// never lost.
|
|
pub async fn requeue_stuck_pending(&self) {
|
|
let cutoff = chrono::Utc::now() - Self::PENDING_GRACE;
|
|
let rows = sqlx::query_as::<_, (Uuid, String, String)>(
|
|
"SELECT id, original_path, mime_type FROM upload
|
|
WHERE deleted_at IS NULL
|
|
AND compression_status = 'pending'
|
|
AND original_path <> ''
|
|
AND created_at < $1
|
|
AND derivative_attempts < $2
|
|
ORDER BY created_at
|
|
LIMIT $3",
|
|
)
|
|
.bind(cutoff)
|
|
.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, "stuck-pending sweep query failed");
|
|
return;
|
|
}
|
|
};
|
|
if rows.is_empty() {
|
|
return;
|
|
}
|
|
// WARN, not INFO: reaching this means the guard did not run, which is worth noticing
|
|
// even though the photo is being rescued.
|
|
tracing::warn!(
|
|
count = rows.len(),
|
|
"found upload(s) stuck at 'pending' with no compression job; re-queueing"
|
|
);
|
|
for (id, original_path, mime_type) in rows {
|
|
self.process(id, original_path, mime_type);
|
|
}
|
|
}
|
|
|
|
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 <> ''
|
|
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 {
|
|
Ok(r) => r,
|
|
Err(e) => {
|
|
tracing::warn!(error = ?e, "derivative backfill query failed");
|
|
return;
|
|
}
|
|
};
|
|
|
|
self.report_exhausted_derivatives().await;
|
|
|
|
if rows.is_empty() {
|
|
return;
|
|
}
|
|
tracing::info!("regenerating derivatives for {} upload(s)", rows.len());
|
|
|
|
// 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)
|
|
.await
|
|
{
|
|
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;
|
|
tracing::info!("derivatives regenerated for upload {id}");
|
|
}
|
|
Err(e) => {
|
|
// Leave the existing derivatives and the original intact; this row is
|
|
// 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"
|
|
);
|
|
}
|
|
}
|
|
|
|
/// Extract the feed poster for a video. `Ok(None)` when the clip yields no frame — see
|
|
/// [`crate::services::video::extract_poster_frame`], which owns the seek order, the timeout and
|
|
/// the artifact check that this function used to be missing.
|
|
async fn generate_video_thumbnail(
|
|
&self,
|
|
upload_id: Uuid,
|
|
original: &Path,
|
|
) -> Result<Option<String>> {
|
|
let thumbs_dir = self.media_path.join("thumbnails");
|
|
tokio::fs::create_dir_all(&thumbs_dir).await?;
|
|
|
|
let thumb_filename = format!("{upload_id}.jpg");
|
|
let thumb_path = thumbs_dir.join(&thumb_filename);
|
|
|
|
let produced =
|
|
crate::services::video::extract_poster_frame(original, &thumb_path, 800).await?;
|
|
|
|
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 <= crate::services::imaging::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 > crate::services::imaging::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);
|
|
}
|
|
}
|