diff --git a/breadarrd/src/db.rs b/breadarrd/src/db.rs index d356d88..0c37c80 100644 --- a/breadarrd/src/db.rs +++ b/breadarrd/src/db.rs @@ -372,7 +372,13 @@ pub fn init(conn: &Connection) -> anyhow::Result<()> { queued_at TEXT NOT NULL, finished_at TEXT ); - CREATE INDEX IF NOT EXISTS idx_transcode_job_status ON transcode_job(status);", + CREATE INDEX IF NOT EXISTS idx_transcode_job_status ON transcode_job(status); + -- Prevents two jobs for the same file ever being active at once — + -- closes the door on the same file getting encoded twice + -- concurrently regardless of how it happened (a stray duplicate + -- enqueue, a daemon-restart reset racing a still-alive backfill). + CREATE UNIQUE INDEX IF NOT EXISTS idx_transcode_job_active_episode_file + ON transcode_job(episode_file_id) WHERE status IN ('pending','running');", )?; // Progress watermark for stalled-download detection (added after the diff --git a/breadarrd/src/importer/mod.rs b/breadarrd/src/importer/mod.rs index 1aa86df..8379fe4 100644 --- a/breadarrd/src/importer/mod.rs +++ b/breadarrd/src/importer/mod.rs @@ -1333,7 +1333,7 @@ pub async fn run_import_cycle( category: &str, container_downloads_path: &str, host_downloads_path: &str, - transcode_enabled: bool, + transcode_cfg: Option<&breadarr_shared::config::TranscodeConfig>, ) -> Result { let pending = fetch_pending_grabs(conn)?; if pending.is_empty() { @@ -1361,7 +1361,7 @@ pub async fn run_import_cycle( &torrents, container_downloads_path, host_downloads_path, - transcode_enabled, + transcode_cfg, )?; if stats.imported > 0 { @@ -1387,7 +1387,7 @@ fn process_pending_grabs( torrents: &[crate::qbit::TorrentInfo], container_downloads_path: &str, host_downloads_path: &str, - transcode_enabled: bool, + transcode_cfg: Option<&breadarr_shared::config::TranscodeConfig>, ) -> Result { let mut stats = ImportStats::default(); @@ -1447,6 +1447,7 @@ fn process_pending_grabs( *season_number, root_folder, &content_path, + transcode_cfg, ) { Ok(outcome) => { stats.imported += outcome.episodes_imported; @@ -1475,7 +1476,7 @@ fn process_pending_grabs( continue; } - match import_one(conn, grab, &content_path, transcode_enabled) { + match import_one(conn, grab, &content_path, transcode_cfg) { Ok(ImportOutcome::Imported { remuxed, quality_flagged, @@ -1526,11 +1527,43 @@ enum ImportOutcome { SkippedAlreadyHaveBetter, } +/// Shared by every place a freshly-imported file becomes eligible for the +/// async transcode queue (`import_one`, season-pack import) — reads the +/// probe data `ensure_probed` just wrote, delegates the actual eligibility +/// call to `transcode::should_enqueue` (the one place anime/HDR/height/ +/// already-AV1 rules live), and enqueues only if all of that says yes. +/// Never propagates a failure into the caller's import result — same +/// "never fail an otherwise-successful import" treatment as probing. +fn maybe_enqueue_transcode( + conn: &Connection, + media_item_id: i64, + episode_file_id: i64, + cfg: &breadarr_shared::config::TranscodeConfig, +) -> Result<()> { + let probe: Option<(Option, i64, Option, i64)> = conn + .query_row( + "SELECT video_codec, hdr, height, probe_size_bytes FROM media_file_probe WHERE episode_file_id = ?1", + params![episode_file_id], + |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?)), + ) + .optional()?; + let Some((video_codec, hdr, height, size_bytes)) = probe else { + // Probing itself failed (recorded as `probe_failed`) — nothing + // reliable to decide eligibility from yet; skip rather than guess. + return Ok(()); + }; + + if crate::transcode::should_enqueue(conn, media_item_id, video_codec.as_deref(), hdr != 0, height, cfg)? { + crate::transcode::enqueue(conn, episode_file_id, video_codec.as_deref(), size_bytes)?; + } + Ok(()) +} + fn import_one( conn: &Connection, grab: &PendingGrab, content_path: &Path, - transcode_enabled: bool, + transcode_cfg: Option<&breadarr_shared::config::TranscodeConfig>, ) -> Result { let source_path = locate_video_file(content_path)?; let ext = source_path @@ -1625,6 +1658,26 @@ fn import_one( // quietly replace the better file already in the library. Checked here // rather than left to the filesystem to decide by import order. if dest.exists() { + // Belt-and-suspenders: the missing-content/upgrade search paths + // already refuse to re-grab an `upgrade_locked` file (it isn't + // "missing" and the upgrade loop skips it), so this shouldn't be + // reachable for one today — but a locally AV1-transcoded file + // should never be silently overwritten on score alone regardless + // of which path produced the incoming grab, same guard as + // `import_season_pack_file`'s matching check. + let upgrade_locked: i64 = conn + .query_row( + "SELECT upgrade_locked FROM episode_file WHERE path = ?1", + params![dest.to_string_lossy()], + |row| row.get(0), + ) + .unwrap_or(0); + if upgrade_locked != 0 { + if remuxed { + std::fs::remove_file(&working_path).ok(); + } + return Ok(ImportOutcome::SkippedAlreadyHaveBetter); + } let current_score: f32 = conn .query_row( "SELECT score FROM release WHERE id = ?1", @@ -1777,29 +1830,9 @@ fn import_one( // immediately, exactly as today; the transcode happens later in the // background. Never blocks or fails the import itself: enqueue errors // are logged and swallowed, same treatment as probing failures above. - if transcode_enabled { - match crate::transcode::is_anime(conn, grab.media_item_id()) { - Ok(true) => {} // Anime excluded until its own encode tuning exists. - Ok(false) => { - let original_codec: Option = conn - .query_row( - "SELECT video_codec FROM media_file_probe WHERE episode_file_id = ?1", - params![episode_file_id], - |row| row.get(0), - ) - .optional()? - .flatten(); - if original_codec.as_deref() != Some("av1") { - if let Err(e) = - crate::transcode::enqueue(conn, episode_file_id, original_codec.as_deref(), size_bytes as i64) - { - tracing::warn!(episode_file_id, error = %e, "failed to enqueue transcode job"); - } - } - } - Err(e) => { - tracing::warn!(episode_file_id, error = %e, "failed to check anime status for transcode enqueue"); - } + if let Some(cfg) = transcode_cfg { + if let Err(e) = maybe_enqueue_transcode(conn, grab.media_item_id(), episode_file_id, cfg) { + tracing::warn!(episode_file_id, error = %e, "failed to check/enqueue transcode job"); } } @@ -1832,6 +1865,7 @@ struct SeasonPackImportOutcome { /// double a season pack's import time. Any file that needs it is still /// reachable afterward via `remux_backlog`, which sweeps the whole library /// including season-pack imports. +#[allow(clippy::too_many_arguments)] fn import_season_pack( conn: &Connection, release_id: i64, @@ -1840,6 +1874,7 @@ fn import_season_pack( season_number: u32, root_folder: &str, content_path: &Path, + transcode_cfg: Option<&breadarr_shared::config::TranscodeConfig>, ) -> Result { let release_score: f32 = conn .query_row( @@ -1919,6 +1954,7 @@ fn import_season_pack( episode_number, root_folder, source_path, + transcode_cfg, ) { Ok(PackFileOutcome::Imported { quality_flagged }) => { outcome.episodes_imported += 1; @@ -1999,6 +2035,7 @@ fn import_season_pack_file( episode_number: u32, root_folder: &str, source_path: &Path, + transcode_cfg: Option<&breadarr_shared::config::TranscodeConfig>, ) -> Result { let ext = source_path .extension() @@ -2027,6 +2064,23 @@ fn import_season_pack_file( // already in place. let mut old_sibling: Option = None; if dest.exists() { + // A locally AV1-transcoded file is a deliberate shrink, not + // something a season pack (scored against its own, much larger, + // original release) should be allowed to silently overwrite just + // because its release score happens to be higher — that score was + // never computed against a transcoded file's actual size/quality + // tradeoff. Checked before the score comparison, same reasoning as + // `movie_eligible_for_upgrade`/`enumerate_upgrade_targets`. + let upgrade_locked: i64 = conn + .query_row( + "SELECT upgrade_locked FROM episode_file WHERE path = ?1", + params![dest.to_string_lossy()], + |row| row.get(0), + ) + .unwrap_or(0); + if upgrade_locked != 0 { + return Ok(PackFileOutcome::SkippedAlreadyHaveBetter); + } let existing_best: Option = conn.query_row( "SELECT MAX(score) FROM release WHERE status = 'imported' AND id != ?1 AND episode_id = ?2", params![release_id, episode_id], @@ -2092,6 +2146,12 @@ fn import_season_pack_file( } }; + if let Some(cfg) = transcode_cfg { + if let Err(e) = maybe_enqueue_transcode(conn, media_item_id, episode_file_id, cfg) { + tracing::warn!(episode_file_id, error = %e, "failed to check/enqueue transcode job"); + } + } + Ok(PackFileOutcome::Imported { quality_flagged }) } @@ -3009,7 +3069,7 @@ mod tests { // absent — simulating torrents qBit no longer knows about. ]; - let stats = process_pending_grabs(&conn, &pending, &torrents, "", "", false).unwrap(); + let stats = process_pending_grabs(&conn, &pending, &torrents, "", "", None).unwrap(); assert_eq!(stats.imported, 1); assert_eq!(stats.skipped_incomplete, 1); @@ -3071,7 +3131,7 @@ mod tests { content_path: "/tmp/somewhere-mid-move".to_string(), }]; - let stats = process_pending_grabs(&conn, &pending, &torrents, "", "", false).unwrap(); + let stats = process_pending_grabs(&conn, &pending, &torrents, "", "", None).unwrap(); assert_eq!(stats.imported, 0); assert_eq!(stats.skipped_incomplete, 1); assert_eq!(stats.errors, 0); @@ -3128,7 +3188,7 @@ mod tests { for i in 1..MAX_IMPORT_ERRORS { let pending = fetch_pending_grabs(&conn).unwrap(); - let stats = process_pending_grabs(&conn, &pending, &torrents, "", "", false).unwrap(); + let stats = process_pending_grabs(&conn, &pending, &torrents, "", "", None).unwrap(); assert_eq!(stats.errors, 1, "iteration {i}"); assert_eq!(stats.failed, 0, "iteration {i}"); let status: String = conn @@ -3139,7 +3199,7 @@ mod tests { // The Nth failure crosses the threshold and gives up. let pending = fetch_pending_grabs(&conn).unwrap(); - let stats = process_pending_grabs(&conn, &pending, &torrents, "", "", false).unwrap(); + let stats = process_pending_grabs(&conn, &pending, &torrents, "", "", None).unwrap(); assert_eq!(stats.failed, 1); assert_eq!(stats.errors, 0); let status: String = conn @@ -3193,7 +3253,7 @@ mod tests { root_folder: dest_root.to_string_lossy().to_string(), }; - let outcome = import_one(&conn, &grab, &content, false).unwrap(); + let outcome = import_one(&conn, &grab, &content, None).unwrap(); let ImportOutcome::Imported { remuxed, .. } = outcome else { panic!("expected a real import, got a skip"); }; @@ -3274,7 +3334,7 @@ mod tests { root_folder: dest_root.to_string_lossy().to_string(), }; - import_one(&conn, &grab, &content, true).unwrap(); + import_one(&conn, &grab, &content, Some(&breadarr_shared::config::TranscodeConfig::default())).unwrap(); let episode_file_id: i64 = conn .query_row( @@ -3350,7 +3410,7 @@ mod tests { root_folder: dest_root.to_string_lossy().to_string(), }; - import_one(&conn, &grab, &content, true).unwrap(); + import_one(&conn, &grab, &content, Some(&breadarr_shared::config::TranscodeConfig::default())).unwrap(); let job_count: i64 = conn .query_row("SELECT count(*) FROM transcode_job", [], |r| r.get(0)) @@ -3415,7 +3475,7 @@ mod tests { root_folder: dest_root.to_string_lossy().to_string(), }; - let outcome = import_one(&conn, &grab, &content, false).unwrap(); + let outcome = import_one(&conn, &grab, &content, None).unwrap(); assert!(matches!(outcome, ImportOutcome::Imported { .. })); let dest = dest_root.join("Season 03").join("Some Show - S03E02.mp4"); @@ -3479,7 +3539,7 @@ mod tests { root_folder: dest_root.to_string_lossy().to_string(), }; - let outcome = import_one(&conn, &grab, &content, false).unwrap(); + let outcome = import_one(&conn, &grab, &content, None).unwrap(); let ImportOutcome::Imported { quality_flagged, .. } = outcome @@ -3689,7 +3749,7 @@ mod tests { root_folder: dest_root.to_string_lossy().to_string(), }; - let outcome = import_one(&conn, &grab, &content, false).unwrap(); + let outcome = import_one(&conn, &grab, &content, None).unwrap(); assert!(matches!(outcome, ImportOutcome::SkippedAlreadyHaveBetter)); // The existing better file must be untouched, not overwritten. @@ -3767,7 +3827,7 @@ mod tests { root_folder: dest_root.to_string_lossy().to_string(), }; - let outcome = import_one(&conn, &grab, &content, false).unwrap(); + let outcome = import_one(&conn, &grab, &content, None).unwrap(); assert!(matches!(outcome, ImportOutcome::Imported { .. })); // The new file's content landed at the shared deterministic path. @@ -3865,6 +3925,7 @@ mod tests { 1, &dest_root.to_string_lossy(), &pack_dir, + None, ) .unwrap(); @@ -3941,6 +4002,7 @@ mod tests { 1, &dest_root.to_string_lossy(), &pack_dir, + None, ) .unwrap(); @@ -3956,6 +4018,63 @@ mod tests { std::fs::remove_dir_all(&dir).unwrap(); } + #[test] + fn import_season_pack_skips_an_upgrade_locked_episode_even_with_a_lower_scoring_existing_release() { + let conn = seeded_season_pack_conn(2); + let dir = std::env::temp_dir().join(format!( + "breadarr-season-pack-locked-{}", + std::process::id() + )); + let pack_dir = dir.join("pack"); + std::fs::create_dir_all(&pack_dir).unwrap(); + std::fs::write(pack_dir.join("Show.S01E01.mkv"), b"new e01").unwrap(); + std::fs::write(pack_dir.join("Show.S01E02.mkv"), b"new e02").unwrap(); + let dest_root = dir.join("library"); + let season_dir = dest_root.join("Season 01"); + std::fs::create_dir_all(&season_dir).unwrap(); + + // Episode 1's existing release scores *lower* than the incoming + // pack (5.0 vs the pack's 15.0) — on score alone this would be + // overwritten. It's also `upgrade_locked` (a completed local AV1 + // transcode), which must take priority over the score comparison. + conn.execute( + "INSERT INTO release (id, media_item_id, episode_id, raw_title, source_id, guid, score, status, torrent_hash, grabbed_at) + VALUES (2, 1, 1, 'Some Show S01E01 1080p', 1, 'guid-2', 5.0, 'imported', 'bbbb', datetime('now'))", + [], + ) + .unwrap(); + let existing = season_dir.join("Some Show - S01E01.mkv"); + std::fs::write(&existing, b"locally-transcoded e01").unwrap(); + conn.execute( + "INSERT INTO episode_file (episode_id, media_item_id, path, size_bytes, subtitle_status, upgrade_locked) + VALUES (1, NULL, ?1, 23, 'none', 1)", + params![existing.to_string_lossy()], + ) + .unwrap(); + + let outcome = import_season_pack( + &conn, + 1, + 1, + "Some Show", + 1, + &dest_root.to_string_lossy(), + &pack_dir, + None, + ) + .unwrap(); + + assert_eq!(outcome.episodes_imported, 1); // only episode 2 + assert_eq!(outcome.episodes_already_had_better, 1); // episode 1 skipped despite the lower score + assert_eq!( + std::fs::read(&existing).unwrap(), + b"locally-transcoded e01", + "the locked, locally-transcoded file must survive untouched regardless of score" + ); + + std::fs::remove_dir_all(&dir).unwrap(); + } + #[test] fn import_season_pack_fails_when_nothing_in_it_matches_a_tracked_episode() { let conn = seeded_season_pack_conn(2); @@ -3978,6 +4097,7 @@ mod tests { 1, &dest_root.to_string_lossy(), &pack_dir, + None, ); assert!(result.is_err()); diff --git a/breadarrd/src/jellyfin.rs b/breadarrd/src/jellyfin.rs index e70e194..1c2883a 100644 --- a/breadarrd/src/jellyfin.rs +++ b/breadarrd/src/jellyfin.rs @@ -1,5 +1,6 @@ use anyhow::{bail, Context, Result}; +#[derive(Clone)] pub struct JellyfinClient { base_url: String, api_key: String, diff --git a/breadarrd/src/main.rs b/breadarrd/src/main.rs index 9b170f9..be4abec 100644 --- a/breadarrd/src/main.rs +++ b/breadarrd/src/main.rs @@ -358,6 +358,16 @@ async fn background_loop( let mut upgrade_ticker = tokio::time::interval(std::time::Duration::from_secs( config.sources.upgrade_poll_interval_secs, )); + // Guards against the transcode ticker itself blocking every other cycle + // (import, search, upgrade, reconcile) for the full multi-minute + // duration of an encode — the ticker below spawns each cycle detached + // rather than awaiting it inline, and this is what stops two spawned + // cycles from running at once if one is still going when the next tick + // fires (a real possibility: files easily take longer to encode than + // `poll_interval_secs`). `claim_pending_jobs`'s own concurrency cap + // already makes overlap *safe*; this just keeps it from happening + // pointlessly. + let transcode_busy = std::sync::Arc::new(tokio::sync::Mutex::new(())); let mut transcode_ticker = tokio::time::interval(std::time::Duration::from_secs( config.transcode.poll_interval_secs, )); @@ -420,7 +430,7 @@ async fn background_loop( _ = import_ticker.tick() => { let result = { let conn = conn.lock().await; - importer::run_import_cycle(&conn, &qbit, jellyfin.as_ref(), &config.qbit.category, &config.qbit.container_downloads_path, &config.qbit.host_downloads_path, config.transcode.enabled).await + importer::run_import_cycle(&conn, &qbit, jellyfin.as_ref(), &config.qbit.category, &config.qbit.container_downloads_path, &config.qbit.host_downloads_path, config.transcode.enabled.then_some(&config.transcode)).await }; if let (Ok(stats), Some(n)) = (&result, ¬ifier) { if stats.failed > 0 { @@ -553,18 +563,33 @@ async fn background_loop( if !config.transcode.enabled { continue; } - let result = transcode::run_cycle(conn.clone(), config.transcode.clone(), jellyfin.as_ref()).await; - let record = match &result { - Ok(stats) => { - info!(?stats, "transcode cycle complete"); - api::CycleRecord { at: chrono::Utc::now(), ok: true, detail: format!("{stats:?}") } - } - Err(e) => { - error!(error = %e, "transcode cycle failed"); - api::CycleRecord { at: chrono::Utc::now(), ok: false, detail: e.to_string() } - } + let Ok(busy_permit) = transcode_busy.clone().try_lock_owned() else { + // A previous cycle is still running (this file's + // encode took longer than one poll interval) — skip + // this tick rather than spawning a second overlapping + // one; the still-running cycle will pick up any newly + // pending jobs on its own next iteration anyway. + continue; }; - cycle_status.lock().expect("cycle_status poisoned").last_transcode = Some(record); + let conn = conn.clone(); + let cfg = config.transcode.clone(); + let jellyfin = jellyfin.clone(); + let cycle_status = cycle_status.clone(); + tokio::spawn(async move { + let _permit = busy_permit; + let result = transcode::run_cycle(conn, cfg, jellyfin.as_ref()).await; + let record = match &result { + Ok(stats) => { + info!(?stats, "transcode cycle complete"); + api::CycleRecord { at: chrono::Utc::now(), ok: true, detail: format!("{stats:?}") } + } + Err(e) => { + error!(error = %e, "transcode cycle failed"); + api::CycleRecord { at: chrono::Utc::now(), ok: false, detail: e.to_string() } + } + }; + cycle_status.lock().expect("cycle_status poisoned").last_transcode = Some(record); + }); } _ = reconcile_ticker.tick() => { let result = { @@ -879,7 +904,7 @@ async fn debug_import_cycle(config: &Config) -> Result<()> { &config.qbit.category, &config.qbit.container_downloads_path, &config.qbit.host_downloads_path, - config.transcode.enabled, + config.transcode.enabled.then_some(&config.transcode), ) .await?; println!("{stats:?}"); diff --git a/breadarrd/src/transcode/mod.rs b/breadarrd/src/transcode/mod.rs index cc8dd10..a7f0e28 100644 --- a/breadarrd/src/transcode/mod.rs +++ b/breadarrd/src/transcode/mod.rs @@ -35,6 +35,17 @@ pub fn target_bitrate_kbps(width: i64, height: i64, cfg: &TranscodeConfig) -> u3 /// Blocking and slow by design (a real GPU encode, potentially minutes per /// file) — callers must run this inside `tokio::task::spawn_blocking`, never /// directly on an async task, and never while holding the shared DB mutex. +/// +/// Known limitation, deliberately not worked around yet: the output +/// container is always Matroska, and `-c:s copy`/`-c:d copy` stream-copy +/// whatever subtitle/data tracks the source has. mp4 sources using +/// `mov_text` subtitles (or certain data streams) aren't valid inside +/// Matroska and will make the whole `ffmpeg` invocation fail — safely +/// (the original file is never touched; the job just lands in `failed` +/// with no space saved for that file), but not silently worked around by +/// re-encoding or dropping the offending stream. Fixing this properly needs +/// probing the source's subtitle/data codecs first and choosing per-stream +/// handling, which wasn't done here rather than risk an unverified fix. fn run_ffmpeg_encode(input: &Path, output: &Path, bitrate_kbps: u32, vaapi_device: &str) -> Result<()> { let maxrate = bitrate_kbps * 3 / 2; let bufsize = bitrate_kbps * 2; @@ -48,6 +59,11 @@ fn run_ffmpeg_encode(input: &Path, output: &Path, bitrate_kbps: u32, vaapi_devic .arg(input) .args(["-map", "0"]) .args(["-c:v", "av1_vaapi"]) + // Explicit rather than relying on the driver's "auto" inference + // from the bitrate flags present — VBR here spends less on simple + // scenes and more on complex ones within the maxrate/bufsize + // envelope below, rather than a flat per-frame target. + .args(["-rc_mode", "VBR"]) .args(["-b:v", &format!("{bitrate_kbps}k")]) .args(["-maxrate", &format!("{maxrate}k")]) .args(["-bufsize", &format!("{bufsize}k")]) @@ -68,6 +84,24 @@ fn run_ffmpeg_encode(input: &Path, output: &Path, bitrate_kbps: u32, vaapi_devic Ok(()) } +/// The temp path a given job's encode writes to — job-id-scoped (not just +/// derived from the input filename) so two jobs can never collide on the +/// same temp file even if something ends up processing the same +/// `episode_file` path twice (e.g. a daemon restart racing a still-live +/// `transcode-library` backfill process). Also what `reset_orphaned_ +/// running_jobs` uses to find and clean up a crashed job's leftover partial. +/// +/// Known cosmetic limitation: `finalize_job` renames this over the +/// original path verbatim, keeping whatever extension the source had — an +/// `.mp4` source ends up as Matroska bytes at an `.mp4` path (Jellyfin +/// content-sniffs, so playback isn't affected, but `episode_file.path`'s +/// extension no longer matches the real container). Not fixed here; would +/// need the rename plus an `episode_file.path` update done together in +/// `finalize_job`'s transaction. +fn temp_path_for(input: &Path, job_id: i64) -> PathBuf { + input.with_extension(format!("job{job_id}.av1.mkv")) +} + /// The blocking half of a transcode: encode to a temp file alongside the /// original (same filesystem, required for the atomic rename-based swap /// later — never a separate staging drive), then verify the result is @@ -75,14 +109,32 @@ fn run_ffmpeg_encode(input: &Path, output: &Path, bitrate_kbps: u32, vaapi_devic /// file on disk on success (the caller finalizes the swap under the DB /// lock); cleans it up itself on any failure, so the original is never at /// risk regardless of what goes wrong here. +/// +/// Returns `Ok(None)` (no encode run at all) if the input turns out to +/// already be AV1 — checked fresh against the real file here, not trusted +/// from whatever `media_file_probe` said when the job was claimed. This is +/// what makes a job that's re-run after a crash between the file-swap +/// rename and the DB bookkeeping (a narrow but real window — see +/// `finalize_job`) self-correct into a no-op instead of re-encoding an +/// already-AV1 file a second time. fn encode_and_verify( input: PathBuf, + job_id: i64, cfg: TranscodeConfig, width: i64, height: i64, original_duration: Option, -) -> Result { - let tmp_path = input.with_extension("av1.mkv"); +) -> Result> { + match ffprobe::probe(&input) { + Ok(p) if p.video_codec.as_deref() == Some("av1") => return Ok(None), + // Any other outcome (a different codec, or the probe itself + // failing) falls through to a real encode attempt as normal — + // this check exists only to short-circuit the one case where + // there's provably nothing to do. + _ => {} + } + + let tmp_path = temp_path_for(&input, job_id); let bitrate = target_bitrate_kbps(width, height, &cfg); run_ffmpeg_encode(&input, &tmp_path, bitrate, &cfg.vaapi_device)?; @@ -118,7 +170,7 @@ fn encode_and_verify( } } - Ok(tmp_path) + Ok(Some(tmp_path)) } /// Whether `media_item_id` is anime — same two membership checks @@ -153,7 +205,32 @@ pub fn is_anime(conn: &Connection, media_item_id: i64) -> Result { /// 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. +/// +/// Also sweeps each reset job's expected temp file (`temp_path_for`, which +/// is job-id-scoped precisely so this lookup is unambiguous) — a killed +/// encode leaves a partial `.jobN.av1.mkv` behind that nothing else will +/// ever clean up, and on a library disk that's usually already tight on +/// space these accumulate for real over a big backfill. pub fn reset_orphaned_running_jobs(conn: &Connection) -> Result { + let orphaned: Vec<(i64, String)> = { + let mut stmt = conn.prepare( + "SELECT j.id, ef.path FROM transcode_job j + JOIN episode_file ef ON ef.id = j.episode_file_id + WHERE j.status = 'running'", + )?; + let rows = stmt.query_map([], |row| Ok((row.get(0)?, row.get(1)?)))?; + rows.collect::>>()? + }; + + for (job_id, path) in &orphaned { + let tmp = temp_path_for(Path::new(path), *job_id); + if tmp.exists() { + if let Err(e) = std::fs::remove_file(&tmp) { + tracing::warn!(job_id, path = %tmp.display(), error = %e, "failed to remove orphaned transcode temp file"); + } + } + } + let reset = conn.execute( "UPDATE transcode_job SET status = 'pending' WHERE status = 'running'", [], @@ -177,6 +254,38 @@ pub fn enqueue( Ok(()) } +/// The single source of truth for "should this freshly-imported file be +/// queued for transcoding at all" — every enqueue site (`import_one`, +/// season-pack import) must go through this rather than re-deriving its own +/// subset of the rules, which is exactly how the post-import hook +/// originally ended up checking anime but silently skipping the HDR/height +/// exclusions `find_backlog_candidates` applies to the backfill. Called +/// after `ensure_probed`, once real ffprobe data (codec/hdr/height) exists +/// for the file — the codec/hdr/height parameters come from that probe, not +/// from anything parsed off the release title. +pub fn should_enqueue( + conn: &Connection, + media_item_id: i64, + video_codec: Option<&str>, + hdr: bool, + height: Option, + cfg: &TranscodeConfig, +) -> Result { + if video_codec == Some("av1") { + return Ok(false); + } + if cfg.exclude_hdr && hdr { + return Ok(false); + } + if height.is_some_and(|h| h >= cfg.exclude_min_height as i64) { + return Ok(false); + } + if is_anime(conn, media_item_id)? { + return Ok(false); + } + Ok(true) +} + /// Every existing-library file eligible for the `transcode-library` backfill: /// not already AV1, not anime, not HDR/2160p+ (per `cfg.exclude_hdr` / /// `cfg.exclude_min_height` — the first pass is scoped to SDR 1080p/720p), @@ -193,7 +302,7 @@ pub fn find_backlog_candidates(conn: &Connection, cfg: &TranscodeConfig) -> Resu WHERE p.video_codec IS NOT NULL AND p.video_codec != 'av1' AND (ef.upgrade_locked IS NULL OR ef.upgrade_locked = 0) AND (?1 = 0 OR p.hdr = 0) - AND (p.height IS NULL OR p.height < ?2) + AND (p.height IS NOT NULL AND p.height < ?2) AND NOT ( (m.tvdb_id IS NOT NULL AND m.tvdb_id IN (SELECT tvdb_id FROM anime_mapping WHERE tvdb_id IS NOT NULL)) @@ -319,10 +428,31 @@ async fn claim_pending_jobs(conn: &Arc>, limit: usize) -> Resu /// AV1 ground truth — same shape as `remux_one_backlog_file`. On failure, /// the original is left completely untouched; the job is marked `failed` /// with the error recorded, no automatic retry. -async fn finalize_job(conn: &Arc>, job: ClaimedJob, encode_result: Result) -> Result { +/// +/// `Ok(None)` from the encode step means `encode_and_verify` found the +/// input already AV1 and skipped the encode entirely — marked `done` with +/// no byte-count change, not `failed`. +/// +/// The whole swap-and-bookkeeping sequence runs inside a closure so any +/// failure partway through (a deleted-out-from-under-it original, a +/// transient I/O error) is caught in one place: the job is marked `failed` +/// and the temp file is cleaned up, rather than an early `?` propagating +/// out and silently leaving the job stuck `running` with a leaked temp file +/// on a disk that's usually already tight on space. +async fn finalize_job( + conn: &Arc>, + job: ClaimedJob, + encode_result: Result>, +) -> Result { let conn = conn.lock().await; - match encode_result { - Ok(tmp_path) => { + let tmp_path = match &encode_result { + Ok(Some(p)) => Some(p.clone()), + _ => None, + }; + + let swap: Result = match encode_result { + Ok(None) => Ok(TranscodeOutcome { original_bytes: 0, new_bytes: 0 }), + Ok(Some(tmp_path)) => (|| { let original_bytes = std::fs::metadata(&job.path)?.len(); std::fs::rename(&tmp_path, &job.path)?; let new_bytes = std::fs::metadata(&job.path)?.len(); @@ -336,14 +466,23 @@ async fn finalize_job(conn: &Arc>, job: ClaimedJob, encode_res params![job.episode_file_id], )?; importer::ensure_probed(&conn, job.episode_file_id, &job.path)?; + Ok(TranscodeOutcome { original_bytes, new_bytes }) + })(), + Err(e) => Err(e), + }; + + match swap { + Ok(outcome) => { conn.execute( "UPDATE transcode_job SET status = 'done', new_bytes = ?1, finished_at = datetime('now') WHERE id = ?2", - params![new_bytes as i64, job.job_id], + params![outcome.new_bytes as i64, job.job_id], )?; - - Ok(TranscodeOutcome { original_bytes, new_bytes }) + Ok(outcome) } Err(e) => { + if let Some(tmp_path) = tmp_path { + let _ = std::fs::remove_file(&tmp_path); + } conn.execute( "UPDATE transcode_job SET status = 'failed', error = ?1, finished_at = datetime('now') WHERE id = ?2", params![e.to_string(), job.job_id], @@ -410,9 +549,11 @@ pub async fn run_cycle( for job in claimed { let cfg = cfg.clone(); let path = job.path.clone(); + let job_id = job.job_id; let (width, height, duration) = (job.width, job.height, job.duration_secs); - let encode_handle = - tokio::task::spawn_blocking(move || encode_and_verify(path, cfg, width, height, duration)); + let encode_handle = tokio::task::spawn_blocking(move || { + encode_and_verify(path, job_id, cfg, width, height, duration) + }); handles.push((job, encode_handle)); } @@ -543,6 +684,43 @@ mod tests { assert!(claimed.is_empty(), "already at the cap — must not claim more"); } + fn generate_test_clip(dir: &Path, codec: &str) -> PathBuf { + let path = dir.join(format!("clip_{codec}.mkv")); + let status = std::process::Command::new("ffmpeg") + .args(["-y", "-f", "lavfi", "-i", "testsrc=size=320x240:duration=1:rate=1"]) + .args(["-c:v", codec]) + .arg(&path) + .output() + .expect("failed to run ffmpeg to generate a test clip"); + assert!( + status.status.success(), + "ffmpeg failed to generate a {codec} test clip: {}", + String::from_utf8_lossy(&status.stderr) + ); + path + } + + // Regression test for a real gap found in review: a crash between the + // file-swap rename and the DB bookkeeping in `finalize_job` leaves a job + // `running` with the file already AV1 but `media_file_probe` still + // stale. On retry, `encode_and_verify` must notice the *actual* file is + // already AV1 and skip re-encoding it, rather than trusting the stale + // DB probe data it was claimed with. + #[test] + fn encode_and_verify_skips_a_file_that_is_already_av1() { + let dir = std::env::temp_dir().join(format!( + "breadarr-transcode-already-av1-{}", + std::process::id() + )); + std::fs::create_dir_all(&dir).unwrap(); + let clip = generate_test_clip(&dir, "libsvtav1"); + + let result = encode_and_verify(clip.clone(), 999, cfg(), 320, 240, Some(1.0)).unwrap(); + assert!(result.is_none(), "an already-AV1 file must not be re-encoded"); + + std::fs::remove_dir_all(&dir).unwrap(); + } + #[test] fn target_bitrate_matches_reference_at_reference_resolution() { let bitrate = target_bitrate_kbps(1920, 1080, &cfg());