fix(compression): bound derivative retries so one bad upload can't loop forever

The OOM in the previous commit was survivable; what made it an outage was that it
repeated. 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 killed the container survived at rev 0, and backfill_stale_derivatives
(called unconditionally at every boot) re-selected it and re-ran the identical
workload. With restart: unless-stopped that is an infinite kill loop, and every
cycle also drops every SSE stream and truncates every in-flight upload.

Verified end to end against the real schema in a scratch database: with the new
guard the backfill selects the row on boots 1-3 and zero rows from boot 4 on,
and a later success resets the counter.

migration 021 adds derivative_attempts and derivative_last_error.

The counter is incremented WRITE-AHEAD, before the work is attempted. This is the
whole design: the failure being bounded is a cgroup SIGKILL, so no Err is
returned, no error handler runs and no Drop fires. A counter bumped in a failure
path increments zero times per crash and the loop would be unchanged.
set_derivatives_rev clears it, so success is the only reset and both the live
path and the backfill get it without a new call site to forget.

Also in the backfill:
- one task walking the rows sequentially instead of one task per row. A large
  backlog used to spawn thousands of tasks, each holding a pool handle and
  queueing on the same two permits, competing with live uploads for a whole boot.
- LIMIT 200 per boot, and original_path <> '' replacing an IS NOT NULL that was
  dead (the column is NOT NULL; cleanup_deleted_media blanks it instead).
- a once-per-boot error log naming how many uploads have given up. Without it the
  give-up is invisible — the loop stops, which is the point, but the photos keep
  a stale derivative forever with nothing to notice.

Adds backfill_video_posters for the mirror-image gap: a video interrupted by a
restart has its compression_status flipped processing -> failed by
startup_recovery and is never re-enqueued, so thumbnail_path stays NULL for the
rest of the event while the clip itself plays fine. It shares the same attempt
budget, which means a genuinely posterless sub-second clip (Live Photo, mis-tap)
stops being re-ffmpeg'd after three boots. That is intended, not a bug to fix
later — Ok(false) is a normal permanent outcome there.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
MechaCat02
2026-07-31 22:52:48 +02:00
parent a4bac03628
commit 5969ec74ea
5 changed files with 277 additions and 10 deletions

View File

@@ -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;

View File

@@ -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;

View File

@@ -58,6 +58,12 @@ async fn main() -> Result<()> {
// originals are never touched, so a failure just retries on the next start. // originals are never touched, so a failure just retries on the next start.
state.compression.backfill_stale_derivatives().await; 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 // 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 // (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 // rather than inside `startup_recovery`. Fire-and-forget: the workers run in the

View File

@@ -150,8 +150,16 @@ impl Upload {
/// Stamp which revision of the derivative pipeline produced this row's preview/display, /// 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. /// 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> { 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(id)
.bind(rev) .bind(rev)
.execute(pool) .execute(pool)
@@ -159,6 +167,50 @@ impl Upload {
Ok(()) 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<Option<i16>, 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(truncated)
.execute(pool)
.await?;
Ok(())
}
pub async fn set_thumbnail_path( pub async fn set_thumbnail_path(
pool: &PgPool, pool: &PgPool,
id: Uuid, id: Uuid,

View File

@@ -62,6 +62,21 @@ impl CompressionWorker {
/// next start. Rev 1 = EXIF orientation is applied. /// next start. Rev 1 = EXIF orientation is applied.
const DERIVATIVES_REV: i16 = 1; 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. /// Spawn a background task to process an uploaded file.
pub fn process(&self, upload_id: Uuid, original_path: String, mime_type: String) { pub fn process(&self, upload_id: Uuid, original_path: String, mime_type: String) {
let worker = self.clone(); let worker = self.clone();
@@ -168,6 +183,21 @@ impl CompressionWorker {
let original = self.media_path.join(original_path); let original = self.media_path.join(original_path);
if mime_type.starts_with("image/") { 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 let (preview_rel, display_rel) = self
.generate_image_derivatives(upload_id, &original, mime_type) .generate_image_derivatives(upload_id, &original, mime_type)
.await?; .await?;
@@ -301,17 +331,29 @@ impl CompressionWorker {
/// ///
/// Unlike the failure path in `process`, a backfill error is logged and skipped — it must /// 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. /// 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) { 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)>( let rows = sqlx::query_as::<_, (Uuid, String, String)>(
"SELECT id, original_path, mime_type FROM upload "SELECT id, original_path, mime_type FROM upload
WHERE deleted_at IS NULL AND mime_type LIKE 'image/%' WHERE deleted_at IS NULL AND mime_type LIKE 'image/%'
AND original_path IS NOT NULL AND original_path <> ''
AND derivative_attempts < $2
AND ( AND (
(display_path IS NULL AND preview_path IS NOT NULL) (display_path IS NULL AND preview_path IS NOT NULL)
OR derivatives_rev < $1 OR derivatives_rev < $1
)", )
ORDER BY created_at DESC
LIMIT $3",
) )
.bind(Self::DERIVATIVES_REV) .bind(Self::DERIVATIVES_REV)
.bind(Self::MAX_DERIVATIVE_ATTEMPTS)
.bind(Self::BACKFILL_BATCH)
.fetch_all(&self.pool) .fetch_all(&self.pool)
.await; .await;
let rows = match rows { let rows = match rows {
@@ -321,14 +363,32 @@ impl CompressionWorker {
return; return;
} }
}; };
self.report_exhausted_derivatives().await;
if rows.is_empty() { if rows.is_empty() {
return; return;
} }
tracing::info!("regenerating derivatives for {} upload(s)", rows.len()); tracing::info!("regenerating derivatives for {} upload(s)", rows.len());
for (id, original_path, mime_type) in rows {
// 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(); let worker = self.clone();
tokio::spawn(async move { tokio::spawn(async move {
for (id, original_path, mime_type) in rows {
let _permit = worker.semaphore.acquire().await; 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); let original = worker.media_path.join(&original_path);
match worker match worker
.generate_image_derivatives(id, &original, &mime_type) .generate_image_derivatives(id, &original, &mime_type)
@@ -337,6 +397,8 @@ impl CompressionWorker {
Ok((preview_rel, display_rel)) => { Ok((preview_rel, display_rel)) => {
let _ = Upload::set_preview_path(&worker.pool, id, &preview_rel).await; 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_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 _ = let _ =
Upload::set_derivatives_rev(&worker.pool, id, Self::DERIVATIVES_REV) Upload::set_derivatives_rev(&worker.pool, id, Self::DERIVATIVES_REV)
.await; .await;
@@ -344,13 +406,134 @@ impl CompressionWorker {
} }
Err(e) => { Err(e) => {
// Leave the existing derivatives and the original intact; this row is // Leave the existing derivatives and the original intact; this row is
// simply retried on the next start. The rev stays behind, which is the // retried on the next start until its attempt budget runs out. The rev
// marker that it still needs doing. // stays behind, which is the marker that it still needs doing.
tracing::warn!(error = ?e, %id, "derivative backfill failed; leaving as-is"); 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 /// Extract the feed poster for a video. `Ok(None)` when the clip yields no frame — see