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, pool: PgPool, media_path: PathBuf, sse_tx: broadcast::Sender, /// 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, } impl CompressionWorker { pub fn new( pool: PgPool, media_path: PathBuf, concurrency: usize, sse_tx: broadcast::Sender, ) -> 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. 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 = 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> { 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); } }