diff --git a/backend/migrations/021_derivative_attempts.down.sql b/backend/migrations/021_derivative_attempts.down.sql new file mode 100644 index 0000000..6870c02 --- /dev/null +++ b/backend/migrations/021_derivative_attempts.down.sql @@ -0,0 +1,3 @@ +DROP INDEX IF EXISTS idx_upload_derivative_backfill; +ALTER TABLE upload DROP COLUMN IF EXISTS derivative_last_error; +ALTER TABLE upload DROP COLUMN IF EXISTS derivative_attempts; diff --git a/backend/migrations/021_derivative_attempts.up.sql b/backend/migrations/021_derivative_attempts.up.sql new file mode 100644 index 0000000..b91a296 --- /dev/null +++ b/backend/migrations/021_derivative_attempts.up.sql @@ -0,0 +1,23 @@ +-- Bound how many times a permanently-failing upload can be re-processed. +-- +-- Without this, one poisoned row is an outage. The upload row is committed BEFORE compression +-- starts, `derivatives_rev` defaults to 0, and `set_derivatives_rev` only runs on success — so +-- a row whose processing kills the container survives at rev 0, the unconditional startup +-- backfill re-selects it on the next boot, and `restart: unless-stopped` turns that into an +-- infinite kill loop. Every restart also drops every SSE stream and truncates every in-flight +-- upload. That was reachable via a single large PNG (see services/compression.rs), but the +-- shape is general: any input that can kill or hang the worker repeats forever. +-- +-- The counter is incremented WRITE-AHEAD, before the work is attempted, because the failure +-- mode being defended against is a SIGKILL — no error is returned, no handler runs, no Drop +-- fires. A counter bumped in an error path increments zero times per crash and changes nothing. +ALTER TABLE upload ADD COLUMN IF NOT EXISTS derivative_attempts SMALLINT NOT NULL DEFAULT 0; + +-- Last failure text, so a row that has given up can be diagnosed without reproducing it. +-- Nothing reads this in code; it exists for the operator. +ALTER TABLE upload ADD COLUMN IF NOT EXISTS derivative_last_error TEXT; + +-- Serves the backfill selection, which now filters on both columns. +CREATE INDEX IF NOT EXISTS idx_upload_derivative_backfill + ON upload (derivatives_rev, derivative_attempts) + WHERE deleted_at IS NULL; diff --git a/backend/src/main.rs b/backend/src/main.rs index c52e8c1..ebd2614 100644 --- a/backend/src/main.rs +++ b/backend/src/main.rs @@ -58,6 +58,12 @@ async fn main() -> Result<()> { // originals are never touched, so a failure just retries on the next start. state.compression.backfill_stale_derivatives().await; + // Re-extract poster frames for videos a restart interrupted. `startup_recovery` above + // marks their compression `failed` but nothing re-enqueued them, so `thumbnail_path` + // stayed NULL for the rest of the event. Shares the attempt budget with the image + // backfill, so a clip that genuinely yields no frame stops being retried. + state.compression.backfill_video_posters().await; + // Re-spawn exports for events that were released but whose keepsake never finished // (crash mid-export). Needs the media/export paths + SSE sender, so it runs here // rather than inside `startup_recovery`. Fire-and-forget: the workers run in the diff --git a/backend/src/models/upload.rs b/backend/src/models/upload.rs index d27fbd7..0334132 100644 --- a/backend/src/models/upload.rs +++ b/backend/src/models/upload.rs @@ -150,10 +150,62 @@ impl Upload { /// Stamp which revision of the derivative pipeline produced this row's preview/display, /// so the startup backfill can find rows generated by an older one exactly once. + /// + /// Also clears the attempt counter: success is the only thing that resets it, and folding + /// the reset in here means both the live path and the backfill get it with no extra call + /// site to forget. pub async fn set_derivatives_rev(pool: &PgPool, id: Uuid, rev: i16) -> Result<(), sqlx::Error> { - sqlx::query("UPDATE upload SET derivatives_rev = $2 WHERE id = $1") + sqlx::query( + "UPDATE upload + SET derivatives_rev = $2, derivative_attempts = 0, derivative_last_error = NULL + WHERE id = $1", + ) + .bind(id) + .bind(rev) + .execute(pool) + .await?; + Ok(()) + } + + /// Record that derivative processing is ABOUT to be attempted, returning the new count. + /// + /// WRITE-AHEAD ON PURPOSE. The failure this bounds is a cgroup SIGKILL: the process + /// vanishes mid-work, so no `Err` is returned, no error handler runs and no `Drop` fires. + /// A counter incremented after a failure would increment zero times per crash and the + /// boot loop would be unchanged. Counting the ATTEMPT is the only thing that survives the + /// process dying. The cost is that a genuinely transient failure also burns an attempt — + /// acceptable, because the retry budget is per-boot-loop, not per-request, and success + /// resets it to zero. + /// `None` when the row no longer exists (hard-deleted, or an e2e TRUNCATE landed while the + /// task waited on the semaphore) — the caller should abandon quietly rather than treat a + /// missing row as a processing failure. + pub async fn begin_derivative_attempt( + pool: &PgPool, + id: Uuid, + ) -> Result, sqlx::Error> { + sqlx::query_scalar( + "UPDATE upload + SET derivative_attempts = derivative_attempts + 1 + WHERE id = $1 + RETURNING derivative_attempts", + ) + .bind(id) + .fetch_optional(pool) + .await + } + + /// Store why the last derivative attempt failed. Diagnostics only — nothing branches on it. + pub async fn record_derivative_failure( + pool: &PgPool, + id: Uuid, + error: &str, + ) -> Result<(), sqlx::Error> { + // Bounded: an anyhow chain can be long, and this is written on a failure path that may + // repeat across every row of a bad batch. + let truncated: String = error.chars().take(500).collect(); + sqlx::query("UPDATE upload SET derivative_last_error = $2 WHERE id = $1") .bind(id) - .bind(rev) + .bind(truncated) .execute(pool) .await?; Ok(()) diff --git a/backend/src/services/compression.rs b/backend/src/services/compression.rs index 1d462e1..051542a 100644 --- a/backend/src/services/compression.rs +++ b/backend/src/services/compression.rs @@ -62,6 +62,21 @@ impl CompressionWorker { /// next start. Rev 1 = EXIF orientation is applied. const DERIVATIVES_REV: i16 = 1; + /// How many times derivative generation may be ATTEMPTED for one upload before it is left + /// alone. Counted write-ahead and reset on success — see `Upload::begin_derivative_attempt`. + /// + /// This is what turns a fatal input from an outage into a blemish. The startup backfill + /// runs unconditionally on every boot, so before this bound a row whose processing killed + /// the process was re-selected and re-run forever, and `restart: unless-stopped` made that + /// an infinite loop that also dropped every SSE stream and truncated every in-flight + /// upload on each cycle. Three attempts absorbs genuinely transient infrastructure + /// failures (an ENOSPC spike, a pool blip) without ever becoming unbounded. + const MAX_DERIVATIVE_ATTEMPTS: i16 = 3; + + /// Rows regenerated per boot. Bounds both the query and the amount of work a single start + /// can queue; whatever is left is picked up on the next boot. + const BACKFILL_BATCH: i64 = 200; + /// Spawn a background task to process an uploaded file. pub fn process(&self, upload_id: Uuid, original_path: String, mime_type: String) { let worker = self.clone(); @@ -168,6 +183,21 @@ impl CompressionWorker { let original = self.media_path.join(original_path); if mime_type.starts_with("image/") { + // Count the attempt BEFORE doing the work — see `begin_derivative_attempt`. If this + // input is the one that kills the container, this write is the only record that + // survives, and it is what stops the boot backfill replaying it forever. + match Upload::begin_derivative_attempt(&self.pool, upload_id).await? { + Some(attempts) if attempts > Self::MAX_DERIVATIVE_ATTEMPTS => { + anyhow::bail!( + "derivative generation gave up after {} attempt(s)", + attempts - 1 + ); + } + Some(_) => {} + // The row vanished while this task waited on the semaphore. Nothing to do, and + // reporting a failure would broadcast into a stream that no longer has a card. + None => return Ok(()), + } let (preview_rel, display_rel) = self .generate_image_derivatives(upload_id, &original, mime_type) .await?; @@ -301,17 +331,29 @@ impl CompressionWorker { /// /// Unlike the failure path in `process`, a backfill error is logged and skipped — it must /// NEVER destroy or soft-delete an upload that already has a working preview. + /// + /// Bounded in three ways, all of them load-bearing on a box that restarts itself: + /// `derivative_attempts` stops a fatal row being replayed on every boot, `BACKFILL_BATCH` + /// stops one start queueing unbounded work, and the whole thing runs as ONE task walking + /// the rows sequentially rather than N tasks racing for the same semaphore. pub async fn backfill_stale_derivatives(&self) { + // `original_path IS NOT NULL` was dead — the column is NOT NULL. What actually needs + // excluding is the blanked path `cleanup_deleted_media` leaves behind. let rows = sqlx::query_as::<_, (Uuid, String, String)>( "SELECT id, original_path, mime_type FROM upload WHERE deleted_at IS NULL AND mime_type LIKE 'image/%' - AND original_path IS NOT NULL + AND original_path <> '' + AND derivative_attempts < $2 AND ( (display_path IS NULL AND preview_path IS NOT NULL) OR derivatives_rev < $1 - )", + ) + ORDER BY created_at DESC + LIMIT $3", ) .bind(Self::DERIVATIVES_REV) + .bind(Self::MAX_DERIVATIVE_ATTEMPTS) + .bind(Self::BACKFILL_BATCH) .fetch_all(&self.pool) .await; let rows = match rows { @@ -321,14 +363,32 @@ impl CompressionWorker { return; } }; + + self.report_exhausted_derivatives().await; + if rows.is_empty() { return; } tracing::info!("regenerating derivatives for {} upload(s)", rows.len()); - for (id, original_path, mime_type) in rows { - let worker = self.clone(); - tokio::spawn(async move { + + // ONE task for the whole batch. The previous shape spawned a task per row, so a large + // backlog created thousands of live tasks that each held a pool handle and queued on + // the same two semaphore permits, competing with live uploads for the entire boot. + let worker = self.clone(); + tokio::spawn(async move { + for (id, original_path, mime_type) in rows { let _permit = worker.semaphore.acquire().await; + // Write-ahead, exactly as in the live path: if this row is the one that kills + // the process, this increment is the only thing that outlives the SIGKILL. + match Upload::begin_derivative_attempt(&worker.pool, id).await { + Ok(Some(n)) if n > Self::MAX_DERIVATIVE_ATTEMPTS => continue, + Ok(Some(_)) => {} + Ok(None) => continue, + Err(e) => { + tracing::warn!(error = ?e, %id, "could not record a backfill attempt; skipping"); + continue; + } + } let original = worker.media_path.join(&original_path); match worker .generate_image_derivatives(id, &original, &mime_type) @@ -337,6 +397,8 @@ impl CompressionWorker { Ok((preview_rel, display_rel)) => { let _ = Upload::set_preview_path(&worker.pool, id, &preview_rel).await; let _ = Upload::set_display_path(&worker.pool, id, &display_rel).await; + // Clears derivative_attempts too, so a row that failed transiently is + // not one boot closer to being abandoned. let _ = Upload::set_derivatives_rev(&worker.pool, id, Self::DERIVATIVES_REV) .await; @@ -344,12 +406,133 @@ impl CompressionWorker { } Err(e) => { // Leave the existing derivatives and the original intact; this row is - // simply retried on the next start. The rev stays behind, which is the - // marker that it still needs doing. + // retried on the next start until its attempt budget runs out. The rev + // stays behind, which is the marker that it still needs doing. tracing::warn!(error = ?e, %id, "derivative backfill failed; leaving as-is"); + let _ = + Upload::record_derivative_failure(&worker.pool, id, &format!("{e:#}")) + .await; } } - }); + } + }); + } + + /// Re-extract poster frames for videos that never got one. + /// + /// A video interrupted by a restart is stranded: `startup_recovery` flips its + /// `compression_status` from `processing` to `failed` and nothing re-enqueues it, so + /// `thumbnail_path` stays NULL forever while the clip itself plays fine. The feed shows a + /// posterless tile for the rest of the event, and after + /// `FAILED_ORIGINAL_RETENTION_DAYS` the reclaim sweep is entitled to the original. + /// + /// Shares `derivative_attempts` with the image backfill on purpose. Note the consequence, + /// which is intended rather than a bug to fix later: `extract_poster_frame` returning + /// `Ok(false)` is a NORMAL, permanent outcome for a sub-second clip (Live Photos, + /// mis-taps), and since the counter is write-ahead and only cleared by a real success, + /// those clips stop being re-ffmpeg'd on every boot once the budget is spent. + pub async fn backfill_video_posters(&self) { + let rows = sqlx::query_as::<_, (Uuid, String)>( + "SELECT id, original_path FROM upload + WHERE deleted_at IS NULL AND mime_type LIKE 'video/%' + AND thumbnail_path IS NULL + AND original_path <> '' + AND derivative_attempts < $1 + ORDER BY created_at DESC + LIMIT $2", + ) + .bind(Self::MAX_DERIVATIVE_ATTEMPTS) + .bind(Self::BACKFILL_BATCH) + .fetch_all(&self.pool) + .await; + let rows = match rows { + Ok(r) => r, + Err(e) => { + tracing::warn!(error = ?e, "video poster backfill query failed"); + return; + } + }; + if rows.is_empty() { + return; + } + tracing::info!("re-extracting posters for {} video(s)", rows.len()); + + let worker = self.clone(); + tokio::spawn(async move { + for (id, original_path) in rows { + let _permit = worker.semaphore.acquire().await; + match Upload::begin_derivative_attempt(&worker.pool, id).await { + Ok(Some(n)) if n > Self::MAX_DERIVATIVE_ATTEMPTS => continue, + Ok(Some(_)) => {} + Ok(None) => continue, + Err(e) => { + tracing::warn!(error = ?e, %id, "could not record a poster attempt; skipping"); + continue; + } + } + let original = worker.media_path.join(&original_path); + match worker.generate_video_thumbnail(id, &original).await { + Ok(Some(thumb_rel)) => { + if Upload::set_thumbnail_path(&worker.pool, id, &thumb_rel) + .await + .is_ok() + { + // Clears the attempt counter: a video that eventually succeeded + // must not carry a budget scar into a future pipeline revision. + let _ = Upload::set_derivatives_rev( + &worker.pool, + id, + Self::DERIVATIVES_REV, + ) + .await; + tracing::info!("poster regenerated for upload {id}"); + } + } + // No frame at all — normal for a very short clip. The tile stays + // posterless and the attempt is spent, which is what stops the retry. + Ok(None) => { + tracing::debug!(%id, "still no poster frame; leaving the tile as-is"); + } + Err(e) => { + tracing::warn!(error = ?e, %id, "poster backfill failed; leaving as-is"); + let _ = + Upload::record_derivative_failure(&worker.pool, id, &format!("{e:#}")) + .await; + } + } + } + }); + } + + /// Say out loud, once per boot, that some uploads have stopped being retried. + /// + /// Without this the give-up is invisible: the loop stops (which is the point) but the + /// affected photos keep a stale or missing derivative forever with nothing to notice. The + /// originals are untouched, so this is recoverable once the cause is fixed — reset + /// `derivative_attempts` to 0 and restart. + async fn report_exhausted_derivatives(&self) { + let exhausted: Result = 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" + ); } }