From 576aad3bfecd3dcf376bea19d2da72c8c4fffefb Mon Sep 17 00:00:00 2001 From: Breadway Date: Sat, 25 Jul 2026 00:20:05 +0800 Subject: [PATCH] Fix a real production incident: unbounded concurrent transcode encodes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The daemon's steady-state transcode ticker and a manually-launched transcode-library backfill process were both independently claiming pending jobs from the same queue with no shared concurrency awareness, each capping only at its own parallelism_max. Combined with large files easily outlasting the 30s poll interval, this stacked to 20+ simultaneous GPU encode/decode sessions and triggered the kernel OOM killer on a shared 15GB host running a dozen+ other containers (confirmed via dmesg; no services were lost, no library files were touched — the original-file-safety design held under the crash). Fixes: claim_pending_jobs now treats its limit as a total concurrency cap (subtracting already-running jobs, wrapped in a BEGIN IMMEDIATE transaction so this is correct across concurrent processes touching the same database, not just within one). Added reset_orphaned_running_jobs, called on real daemon startup, so a crash never permanently strands job slots in 'running'. Lowered the default parallelism_max from 4 to 2 given the observed real-world memory pressure. --- breadarr-shared/src/config.rs | 8 +- breadarrd/src/main.rs | 15 +++ breadarrd/src/transcode/mod.rs | 196 ++++++++++++++++++++++++++++----- 3 files changed, 188 insertions(+), 31 deletions(-) diff --git a/breadarr-shared/src/config.rs b/breadarr-shared/src/config.rs index 684d6c7..5d0ee05 100644 --- a/breadarr-shared/src/config.rs +++ b/breadarr-shared/src/config.rs @@ -364,7 +364,13 @@ fn default_parallelism_min() -> usize { } fn default_parallelism_max() -> usize { - 4 + // Conservative on purpose: a real incident on a shared 15GB host + // running a dozen+ other containers showed concurrent 1080p/4K + // decode+encode sessions can push memory pressure into swap fast + // enough to trigger the OOM killer well before the GPU itself is the + // bottleneck. Raise this deliberately, per-deployment, once you've + // watched `free -h` under real load at the current setting. + 2 } fn default_reference_bitrate_kbps() -> u32 { diff --git a/breadarrd/src/main.rs b/breadarrd/src/main.rs index bd66aeb..9b170f9 100644 --- a/breadarrd/src/main.rs +++ b/breadarrd/src/main.rs @@ -142,6 +142,11 @@ async fn run_daemon(config: Config) -> Result<()> { let conn = Connection::open(config.db_path())?; db::init(&conn)?; info!(path = %config.db_path().display(), "database ready"); + match transcode::reset_orphaned_running_jobs(&conn) { + Ok(0) => {} + Ok(n) => info!(n, "reset orphaned 'running' transcode jobs left over from a previous crash"), + Err(e) => tracing::warn!(error = %e, "failed to reset orphaned transcode jobs"), + } // A second, independent connection for the HTTP API rather than sharing // `background_loop`'s. Both point at the same on-disk (WAL-mode) @@ -1209,6 +1214,16 @@ async fn probe_library_cmd(config: &Config) -> Result<()> { /// (`transcode::run_cycle`) until nothing is left pending. Shares that one /// code path deliberately — there is exactly one place that actually runs /// an encode, whether triggered by a backlog sweep or a fresh grab. +// Deliberately does NOT call `transcode::reset_orphaned_running_jobs` on +// startup the way `run_daemon` does — from this CLI's vantage point a +// `running` row could belong to the actual daemon's own ticker legitimately +// working on it right now (see the safety writeup on `claim_pending_jobs`: +// running this backfill alongside a live daemon with transcode enabled is +// intentionally supported, the two now share one atomic, count-aware +// concurrency cap), and there's no reliable way to tell that apart from a +// genuinely orphaned row from a ago-crashed run of this same command. +// If *this* command itself is killed mid-run, its claimed jobs stay +// `running` until the daemon is next restarted (which does its own reset). async fn transcode_library_cmd(config: &Config) -> Result<()> { if !config.transcode.enabled { bail!("transcode.enabled is false in config — enable it before running a backfill"); diff --git a/breadarrd/src/transcode/mod.rs b/breadarrd/src/transcode/mod.rs index 44414ae..cc8dd10 100644 --- a/breadarrd/src/transcode/mod.rs +++ b/breadarrd/src/transcode/mod.rs @@ -144,6 +144,23 @@ pub fn is_anime(conn: &Connection, media_item_id: i64) -> Result { Ok(result) } +/// Resets any `running` job back to `pending` — call once at process +/// startup (the daemon itself, and the `transcode-library` backfill), never +/// mid-run. `running` only ever means "some still-alive process is +/// currently encoding this" — a row left in that state at startup can only +/// mean the process that claimed it is gone (crashed, killed, power loss), +/// since a live process's own in-flight jobs aren't visible to it as +/// leftover state, they're just still running. Left unreset, a crash leaves +/// that job's slot permanently uncountable-but-also-unclaimable, quietly +/// shrinking real capacity forever instead of just costing one retry. +pub fn reset_orphaned_running_jobs(conn: &Connection) -> Result { + let reset = conn.execute( + "UPDATE transcode_job SET status = 'pending' WHERE status = 'running'", + [], + )?; + Ok(reset) +} + /// Queues one file for transcoding. `original_codec`/`original_bytes` are /// just recorded for the eventual report — not used for any decision. pub fn enqueue( @@ -226,39 +243,71 @@ struct ClaimedJob { /// needs. Only claims jobs whose file already has probe data — a job /// enqueued moments after import but before `ensure_probed` has run yet /// simply isn't claimed this tick, and picks up naturally on the next one. +/// +/// `limit` is treated as a *total* concurrency target, not "claim this many +/// more" — it's first reduced by however many jobs are already `running` +/// (set by anyone: this same daemon's previous still-in-flight cycle, or a +/// separately-invoked `transcode-library` backfill process hitting the same +/// database). Learned the hard way: without this, a long-running cycle +/// (large files easily take longer than `poll_interval_secs`) and a +/// concurrently-run backfill each independently claimed up to their own +/// `parallelism_max` with no awareness of the other, stacking to 20+ +/// simultaneous GPU encodes and OOMing the host. `status='running'` is +/// shared database state, not per-process, so counting it is what makes +/// this a real global cap no matter how many processes are hitting this +/// table at once — and wrapping the count+select+update in one +/// `BEGIN IMMEDIATE` transaction (rather than three separate statements) +/// is what stops two processes from both reading the same low count and +/// both claiming past the limit before either one's UPDATE lands. async fn claim_pending_jobs(conn: &Arc>, limit: usize) -> Result> { - let conn = conn.lock().await; - // Highest current bitrate first — the worst offenders (REMUX, huge - // season packs) free the most space per file transcoded, so they're - // worth reaching before smaller, already-reasonable files. - let mut stmt = conn.prepare( - "SELECT j.id, j.episode_file_id, ef.path, p.width, p.height, p.duration_secs - FROM transcode_job j - JOIN episode_file ef ON ef.id = j.episode_file_id - JOIN media_file_probe p ON p.episode_file_id = ef.id - WHERE j.status = 'pending' AND p.width IS NOT NULL AND p.height IS NOT NULL - ORDER BY (ef.size_bytes * 8.0 / NULLIF(p.duration_secs, 0)) DESC - LIMIT ?1", - )?; - let claimed = stmt - .query_map(params![limit as i64], |row| { - Ok(ClaimedJob { - job_id: row.get(0)?, - episode_file_id: row.get(1)?, - path: PathBuf::from(row.get::<_, String>(2)?), - width: row.get(3)?, - height: row.get(4)?, - duration_secs: row.get(5)?, - }) - })? - .collect::>>()?; + let mut conn = conn.lock().await; + let tx = conn.transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?; - for job in &claimed { - conn.execute( - "UPDATE transcode_job SET status = 'running' WHERE id = ?1", - params![job.job_id], + let running: i64 = tx.query_row( + "SELECT count(*) FROM transcode_job WHERE status = 'running'", + [], + |row| row.get(0), + )?; + let available = (limit as i64 - running).max(0); + + let claimed = if available == 0 { + Vec::new() + } else { + // Highest current bitrate first — the worst offenders (REMUX, huge + // season packs) free the most space per file transcoded, so + // they're worth reaching before smaller, already-reasonable files. + let mut stmt = tx.prepare( + "SELECT j.id, j.episode_file_id, ef.path, p.width, p.height, p.duration_secs + FROM transcode_job j + JOIN episode_file ef ON ef.id = j.episode_file_id + JOIN media_file_probe p ON p.episode_file_id = ef.id + WHERE j.status = 'pending' AND p.width IS NOT NULL AND p.height IS NOT NULL + ORDER BY (ef.size_bytes * 8.0 / NULLIF(p.duration_secs, 0)) DESC + LIMIT ?1", )?; - } + let claimed = stmt + .query_map(params![available], |row| { + Ok(ClaimedJob { + job_id: row.get(0)?, + episode_file_id: row.get(1)?, + path: PathBuf::from(row.get::<_, String>(2)?), + width: row.get(3)?, + height: row.get(4)?, + duration_secs: row.get(5)?, + }) + })? + .collect::>>()?; + + for job in &claimed { + tx.execute( + "UPDATE transcode_job SET status = 'running' WHERE id = ?1", + params![job.job_id], + )?; + } + claimed + }; + + tx.commit()?; Ok(claimed) } @@ -407,6 +456,93 @@ mod tests { } } + fn seed_file(conn: &Connection, id: i64, size_bytes: i64) { + conn.execute( + "INSERT INTO episode_file (id, episode_id, media_item_id, path, size_bytes) + VALUES (?1, NULL, NULL, ?2, ?3)", + params![id, format!("/tmp/f{id}.mkv"), size_bytes], + ) + .unwrap(); + conn.execute( + "INSERT INTO media_file_probe (episode_file_id, probed_at, probe_size_bytes, probe_mtime, + duration_secs, video_codec, width, height) + VALUES (?1, datetime('now'), ?2, 0, 1200.0, 'hevc', 1920, 1080)", + params![id, size_bytes], + ) + .unwrap(); + } + + // Regression test for a real incident: a long-running cycle (large + // files can easily outlast `poll_interval_secs`) and a separately + // invoked `transcode-library` backfill each independently claimed up to + // their own `parallelism_max` with no shared awareness, stacking to 20+ + // concurrent GPU encodes and OOMing the host. `claim_pending_jobs` must + // treat `limit` as a total cap, not "claim this many more". + #[tokio::test] + async fn claim_pending_jobs_respects_already_running_jobs_as_a_global_cap() { + let conn = Connection::open_in_memory().unwrap(); + crate::db::init(&conn).unwrap(); + for id in 1..=5 { + seed_file(&conn, id, 1_000_000_000); + } + // Two jobs already claimed by "someone else" (another process, or + // this same process's still-in-flight previous cycle). + conn.execute( + "INSERT INTO transcode_job (episode_file_id, status, queued_at) VALUES (1, 'running', datetime('now'))", + [], + ) + .unwrap(); + conn.execute( + "INSERT INTO transcode_job (episode_file_id, status, queued_at) VALUES (2, 'running', datetime('now'))", + [], + ) + .unwrap(); + for id in 3..=5 { + conn.execute( + "INSERT INTO transcode_job (episode_file_id, status, queued_at) VALUES (?1, 'pending', datetime('now'))", + params![id], + ) + .unwrap(); + } + + let conn = Arc::new(Mutex::new(conn)); + // Asking for a total of 3 concurrent, with 2 already running — + // should claim exactly 1 more, not 3 more. + let claimed = claim_pending_jobs(&conn, 3).await.unwrap(); + assert_eq!(claimed.len(), 1); + + let conn = conn.lock().await; + let running: i64 = conn + .query_row("SELECT count(*) FROM transcode_job WHERE status = 'running'", [], |r| r.get(0)) + .unwrap(); + assert_eq!(running, 3, "total in-flight jobs must never exceed the requested cap"); + } + + #[tokio::test] + async fn claim_pending_jobs_claims_nothing_when_already_at_the_cap() { + let conn = Connection::open_in_memory().unwrap(); + crate::db::init(&conn).unwrap(); + for id in 1..=3 { + seed_file(&conn, id, 1_000_000_000); + } + for id in 1..=2 { + conn.execute( + "INSERT INTO transcode_job (episode_file_id, status, queued_at) VALUES (?1, 'running', datetime('now'))", + params![id], + ) + .unwrap(); + } + conn.execute( + "INSERT INTO transcode_job (episode_file_id, status, queued_at) VALUES (3, 'pending', datetime('now'))", + [], + ) + .unwrap(); + + let conn = Arc::new(Mutex::new(conn)); + let claimed = claim_pending_jobs(&conn, 2).await.unwrap(); + assert!(claimed.is_empty(), "already at the cap — must not claim more"); + } + #[test] fn target_bitrate_matches_reference_at_reference_resolution() { let bitrate = target_bitrate_kbps(1920, 1080, &cfg());