2f952494c2
Reproduced live, by accident, while smoke-testing on a machine with no ffmpeg: the clip uploaded fine, returned 201, and roughly six seconds later had `deleted_at` set and was gone from the feed. The `Ok(None)` "this clip yields no frame" case was already handled — that fix landed when sub-second clips were being destroyed. But the `?` on the call itself still routed every OTHER failure into the same give-up path, which soft-deletes: ffmpeg missing from the image, ffmpeg hanging on a truncated `.mov` and tripping the timeout, an ENOSPC on `thumbnails/`, or a DB blip in `set_thumbnail_path`. None of those says anything about the video, and `get_original` serves the file byte-for-byte, so a post that merely lacks a poster is fully watchable. No failure in the video branch may fail the upload. iPhone `.mov` is exactly the input most likely to trip it, and a wedding clip is not retakeable. ENOSPC gets its own classifier. It was the one failure the retry loop actively made worse: a disk does not drain during six seconds of backoff, so all three attempts failed identically while holding a compression permit that photos were queued behind — and the give-up path then refunded the quota and soft-deleted the row while deliberately KEEPING the original. That freed nothing, removed the photo seconds after a 201, and handed the guest the allowance to upload it again into the same full disk. Now: no retry, no refund, no delete. The row stays live and the photo is served from its original, and `backfill_stale_derivatives` regenerates the derivatives on the next start once there is room. `is_storage_full_error` has to look inside `ImageError::IoError` as well as at bare io errors, because `image` wraps rather than sources it and a plain chain walk would miss every derivative-write failure. FFMPEG_TIMEOUT drops 120s -> 45s. It was never a budget for honest work — a poster from a phone clip takes well under a second, and `-ss` before `-i` means even a 500 MB file seeks rather than scans. It is the ceiling on how long a pathological input holds a permit that guests' photos are waiting behind, so it should be as tight as it can be without cutting off real work. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
416 lines
20 KiB
Rust
416 lines
20 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;
|
|
|
|
/// 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
|
|
.do_process(upload_id, &original_path, &mime_type)
|
|
.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:#}"
|
|
);
|
|
// Refund + soft-delete (one tx, so v_feed excludes it) so a failed
|
|
// transcode doesn't leave a permanently broken feed card or silently
|
|
// charge the uploader's quota. Then tell the uploader (upload-error
|
|
// toast) and evict the card everywhere (upload-deleted).
|
|
//
|
|
// The ORIGINAL IS DELIBERATELY KEPT. This path used to `remove_file` it
|
|
// unconditionally, which meant any transient error — a disk-full blip
|
|
// while saving a derivative, a pool hiccup, a panic in the image codec —
|
|
// irreversibly destroyed the guest's only copy of a photo they can never
|
|
// retake. The row is only soft-deleted, so keeping the bytes makes the
|
|
// upload fully recoverable; the file is orphaned rather than lost, and
|
|
// the path is logged so it can be found. `backfill_stale_derivatives`
|
|
// already refuses to destroy data on error for exactly this reason.
|
|
let _ = Upload::set_compression_status(&worker.pool, upload_id, "failed").await;
|
|
if let Err(del) = Upload::soft_delete(&worker.pool, upload_id).await {
|
|
tracing::warn!(error = ?del, %upload_id, "failed to soft-delete after compression failure");
|
|
}
|
|
tracing::warn!(
|
|
%upload_id,
|
|
path = %worker.media_path.join(&original_path).display(),
|
|
"original retained for recovery after compression failure"
|
|
);
|
|
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(),
|
|
});
|
|
let _ = worker.sse_tx.send(SseEvent {
|
|
event_type: "upload-deleted".to_string(),
|
|
data: serde_json::json!({ "upload_id": upload_id }).to_string(),
|
|
});
|
|
}
|
|
}
|
|
});
|
|
}
|
|
|
|
async fn do_process(
|
|
&self,
|
|
upload_id: Uuid,
|
|
original_path: &str,
|
|
mime_type: &str,
|
|
) -> 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/") {
|
|
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;
|
|
|
|
/// 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();
|
|
let preview_max = Self::PREVIEW_MAX_EDGE;
|
|
let display_max = Self::DISPLAY_MAX_EDGE;
|
|
|
|
// 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(())
|
|
})
|
|
.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.
|
|
pub async fn backfill_stale_derivatives(&self) {
|
|
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 (
|
|
(display_path IS NULL AND preview_path IS NOT NULL)
|
|
OR derivatives_rev < $1
|
|
)",
|
|
)
|
|
.bind(Self::DERIVATIVES_REV)
|
|
.fetch_all(&self.pool)
|
|
.await;
|
|
let rows = match rows {
|
|
Ok(r) => r,
|
|
Err(e) => {
|
|
tracing::warn!(error = ?e, "derivative backfill query failed");
|
|
return;
|
|
}
|
|
};
|
|
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 {
|
|
let _permit = worker.semaphore.acquire().await;
|
|
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;
|
|
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
|
|
// simply retried on the next start. The rev stays behind, which is the
|
|
// marker that it still needs doing.
|
|
tracing::warn!(error = ?e, %id, "derivative backfill failed; leaving as-is");
|
|
}
|
|
}
|
|
});
|
|
}
|
|
}
|
|
|
|
/// 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}")))
|
|
}
|
|
}
|