Breadway/breadarr
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Fix real bugs found by an independent Opus review of the transcode feature

Browse source 
Requested and applied a full review of the AV1 transcode implementation.
Findings and fixes:

- Season-pack import bypassed upgrade_locked entirely: a season pack
  scored higher than a locally-transcoded file's stale release score
  would silently overwrite it. Fixed in import_season_pack_file (and
  mirrored into import_one for defense-in-depth) to check
  upgrade_locked before the score comparison, not after.
- The post-import enqueue hook only checked anime, silently skipping
  the HDR/2160p exclusions find_backlog_candidates applies to the
  backfill -- a freshly-grabbed HDR file would have gone through the
  unverified HDR-via-VAAPI path. Centralized all eligibility rules
  (anime, already-av1, HDR, height) into transcode::should_enqueue,
  used by both import_one and the newly-added season-pack enqueue
  hook (season packs previously had no transcode hook at all, despite
  being the actual headline use case -- 100GB+ season packs).
- find_backlog_candidates: a NULL-height row was included as a
  candidate but could never actually be claimed (claim_pending_jobs
  requires non-null height), permanently stuck pending. Fixed the
  WHERE clause.
- encode_and_verify now probes the real input file first and skips
  the encode entirely if it's already AV1 -- closes a narrow crash
  window where a rename-succeeded-but-DB-update-failed job would
  otherwise re-encode an already-transcoded file on retry.
- finalize_job's success path could leak a verified temp file and
  strand a job in 'running' forever if a filesystem operation failed
  partway through; now wrapped so any failure there cleans up and
  marks the job failed like every other error path.
- reset_orphaned_running_jobs now also sweeps each reset job's
  leftover temp file (job-id-scoped paths, so this lookup is
  unambiguous) rather than leaking them on a disk that's usually
  already tight on space.
- Added a unique index preventing two active jobs for the same file
  ever existing at once -- closes the remaining gap in last commit's
  concurrency fix (a daemon restart's reset could otherwise race a
  still-alive transcode-library backfill onto the same file).
- Made the VAAPI rate-control mode explicit (VBR) instead of
  driver-inferred.
- The daemon's transcode ticker awaited each cycle inline in the
  tokio::select! loop, blocking every other cycle (import, search,
  upgrade, reconcile) for the full multi-minute duration of an
  encode. Now spawns each cycle detached with a try-lock guard so
  overlapping ticks skip cleanly rather than stacking (the
  concurrency cap from the previous commit makes this safe).

Two items reviewed and deliberately left as documented, not fixed:
mp4 sources with mov_text subtitles will fail the encode cleanly (no
data loss, just no space saved) since Matroska can't hold that codec
via stream copy -- fixing this needs per-stream codec probing that
wasn't safe to add untested at this hour. Renaming an mp4 source's
extension to .mkv after transcoding is cosmetic (Jellyfin
content-sniffs fine) and left alone.
This commit is contained in:
Breadway 2026-07-25 00:38:51 +08:00
parent 576aad3bfe
commit 66d323b7f7
5 changed files with 395 additions and 65 deletions
Download patch file Download diff file Expand all files Collapse all files

8
breadarrd/src/db.rs
View file

@ -372,7 +372,13 @@ pub fn init(conn: &Connection) -> anyhow::Result<()> {
queued_at TEXT NOT NULL, queued_at TEXT NOT NULL,
finished_at TEXT 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 // Progress watermark for stalled-download detection (added after the

198
breadarrd/src/importer/mod.rs
View file

@ -1333,7 +1333,7 @@ pub async fn run_import_cycle(
category: &str, category: &str,
container_downloads_path: &str, container_downloads_path: &str,
host_downloads_path: &str, host_downloads_path: &str,
transcode_enabled: bool, transcode_cfg: Option<&breadarr_shared::config::TranscodeConfig>,
) -> Result<ImportStats> { ) -> Result<ImportStats> {
let pending = fetch_pending_grabs(conn)?; let pending = fetch_pending_grabs(conn)?;
if pending.is_empty() { if pending.is_empty() {
@ -1361,7 +1361,7 @@ pub async fn run_import_cycle(
&torrents, &torrents,
container_downloads_path, container_downloads_path,
host_downloads_path, host_downloads_path,
transcode_enabled, transcode_cfg,
)?; )?;
if stats.imported > 0 { if stats.imported > 0 {
@ -1387,7 +1387,7 @@ fn process_pending_grabs(
torrents: &[crate::qbit::TorrentInfo], torrents: &[crate::qbit::TorrentInfo],
container_downloads_path: &str, container_downloads_path: &str,
host_downloads_path: &str, host_downloads_path: &str,
transcode_enabled: bool, transcode_cfg: Option<&breadarr_shared::config::TranscodeConfig>,
) -> Result<ImportStats> { ) -> Result<ImportStats> {
let mut stats = ImportStats::default(); let mut stats = ImportStats::default();
@ -1447,6 +1447,7 @@ fn process_pending_grabs(
*season_number, *season_number,
root_folder, root_folder,
&content_path, &content_path,
transcode_cfg,
) { ) {
Ok(outcome) => { Ok(outcome) => {
stats.imported += outcome.episodes_imported; stats.imported += outcome.episodes_imported;
@ -1475,7 +1476,7 @@ fn process_pending_grabs(
continue; continue;
} }
match import_one(conn, grab, &content_path, transcode_enabled) { match import_one(conn, grab, &content_path, transcode_cfg) {
Ok(ImportOutcome::Imported { Ok(ImportOutcome::Imported {
remuxed, remuxed,
quality_flagged, quality_flagged,
@ -1526,11 +1527,43 @@ enum ImportOutcome {
SkippedAlreadyHaveBetter, 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<String>, i64, Option<i64>, 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( fn import_one(
conn: &Connection, conn: &Connection,
grab: &PendingGrab, grab: &PendingGrab,
content_path: &Path, content_path: &Path,
transcode_enabled: bool, transcode_cfg: Option<&breadarr_shared::config::TranscodeConfig>,
) -> Result<ImportOutcome> { ) -> Result<ImportOutcome> {
let source_path = locate_video_file(content_path)?; let source_path = locate_video_file(content_path)?;
let ext = source_path let ext = source_path
@ -1625,6 +1658,26 @@ fn import_one(
// quietly replace the better file already in the library. Checked here // quietly replace the better file already in the library. Checked here
// rather than left to the filesystem to decide by import order. // rather than left to the filesystem to decide by import order.
if dest.exists() { 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 let current_score: f32 = conn
.query_row( .query_row(
"SELECT score FROM release WHERE id = ?1", "SELECT score FROM release WHERE id = ?1",
@ -1777,29 +1830,9 @@ fn import_one(
// immediately, exactly as today; the transcode happens later in the // immediately, exactly as today; the transcode happens later in the
// background. Never blocks or fails the import itself: enqueue errors // background. Never blocks or fails the import itself: enqueue errors
// are logged and swallowed, same treatment as probing failures above. // are logged and swallowed, same treatment as probing failures above.
if transcode_enabled { if let Some(cfg) = transcode_cfg {
match crate::transcode::is_anime(conn, grab.media_item_id()) { if let Err(e) = maybe_enqueue_transcode(conn, grab.media_item_id(), episode_file_id, cfg) {
Ok(true) => {} // Anime excluded until its own encode tuning exists. tracing::warn!(episode_file_id, error = %e, "failed to check/enqueue transcode job");
Ok(false) => {
let original_codec: Option<String> = 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");
}
} }
} }
@ -1832,6 +1865,7 @@ struct SeasonPackImportOutcome {
/// double a season pack's import time. Any file that needs it is still /// double a season pack's import time. Any file that needs it is still
/// reachable afterward via `remux_backlog`, which sweeps the whole library /// reachable afterward via `remux_backlog`, which sweeps the whole library
/// including season-pack imports. /// including season-pack imports.
#[allow(clippy::too_many_arguments)]
fn import_season_pack( fn import_season_pack(
conn: &Connection, conn: &Connection,
release_id: i64, release_id: i64,
@ -1840,6 +1874,7 @@ fn import_season_pack(
season_number: u32, season_number: u32,
root_folder: &str, root_folder: &str,
content_path: &Path, content_path: &Path,
transcode_cfg: Option<&breadarr_shared::config::TranscodeConfig>,
) -> Result<SeasonPackImportOutcome> { ) -> Result<SeasonPackImportOutcome> {
let release_score: f32 = conn let release_score: f32 = conn
.query_row( .query_row(
@ -1919,6 +1954,7 @@ fn import_season_pack(
episode_number, episode_number,
root_folder, root_folder,
source_path, source_path,
transcode_cfg,
) { ) {
Ok(PackFileOutcome::Imported { quality_flagged }) => { Ok(PackFileOutcome::Imported { quality_flagged }) => {
outcome.episodes_imported += 1; outcome.episodes_imported += 1;
@ -1999,6 +2035,7 @@ fn import_season_pack_file(
episode_number: u32, episode_number: u32,
root_folder: &str, root_folder: &str,
source_path: &Path, source_path: &Path,
transcode_cfg: Option<&breadarr_shared::config::TranscodeConfig>,
) -> Result<PackFileOutcome> { ) -> Result<PackFileOutcome> {
let ext = source_path let ext = source_path
.extension() .extension()
@ -2027,6 +2064,23 @@ fn import_season_pack_file(
// already in place. // already in place.
let mut old_sibling: Option<PathBuf> = None; let mut old_sibling: Option<PathBuf> = None;
if dest.exists() { 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<f32> = conn.query_row( let existing_best: Option<f32> = conn.query_row(
"SELECT MAX(score) FROM release WHERE status = 'imported' AND id != ?1 AND episode_id = ?2", "SELECT MAX(score) FROM release WHERE status = 'imported' AND id != ?1 AND episode_id = ?2",
params![release_id, episode_id], 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 }) Ok(PackFileOutcome::Imported { quality_flagged })
} }
@ -3009,7 +3069,7 @@ mod tests {
// absent — simulating torrents qBit no longer knows about. // 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.imported, 1);
assert_eq!(stats.skipped_incomplete, 1); assert_eq!(stats.skipped_incomplete, 1);
@ -3071,7 +3131,7 @@ mod tests {
content_path: "/tmp/somewhere-mid-move".to_string(), 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.imported, 0);
assert_eq!(stats.skipped_incomplete, 1); assert_eq!(stats.skipped_incomplete, 1);
assert_eq!(stats.errors, 0); assert_eq!(stats.errors, 0);
@ -3128,7 +3188,7 @@ mod tests {
for i in 1..MAX_IMPORT_ERRORS { for i in 1..MAX_IMPORT_ERRORS {
let pending = fetch_pending_grabs(&conn).unwrap(); 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.errors, 1, "iteration {i}");
assert_eq!(stats.failed, 0, "iteration {i}"); assert_eq!(stats.failed, 0, "iteration {i}");
let status: String = conn let status: String = conn
@ -3139,7 +3199,7 @@ mod tests {
// The Nth failure crosses the threshold and gives up. // The Nth failure crosses the threshold and gives up.
let pending = fetch_pending_grabs(&conn).unwrap(); 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.failed, 1);
assert_eq!(stats.errors, 0); assert_eq!(stats.errors, 0);
let status: String = conn let status: String = conn
@ -3193,7 +3253,7 @@ mod tests {
root_folder: dest_root.to_string_lossy().to_string(), 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 { let ImportOutcome::Imported { remuxed, .. } = outcome else {
panic!("expected a real import, got a skip"); panic!("expected a real import, got a skip");
}; };
@ -3274,7 +3334,7 @@ mod tests {
root_folder: dest_root.to_string_lossy().to_string(), 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 let episode_file_id: i64 = conn
.query_row( .query_row(
@ -3350,7 +3410,7 @@ mod tests {
root_folder: dest_root.to_string_lossy().to_string(), 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 let job_count: i64 = conn
.query_row("SELECT count(*) FROM transcode_job", [], |r| r.get(0)) .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(), 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 { .. })); assert!(matches!(outcome, ImportOutcome::Imported { .. }));
let dest = dest_root.join("Season 03").join("Some Show - S03E02.mp4"); 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(), 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 { let ImportOutcome::Imported {
quality_flagged, .. quality_flagged, ..
} = outcome } = outcome
@ -3689,7 +3749,7 @@ mod tests {
root_folder: dest_root.to_string_lossy().to_string(), 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)); assert!(matches!(outcome, ImportOutcome::SkippedAlreadyHaveBetter));
// The existing better file must be untouched, not overwritten. // The existing better file must be untouched, not overwritten.
@ -3767,7 +3827,7 @@ mod tests {
root_folder: dest_root.to_string_lossy().to_string(), 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 { .. })); assert!(matches!(outcome, ImportOutcome::Imported { .. }));
// The new file's content landed at the shared deterministic path. // The new file's content landed at the shared deterministic path.
@ -3865,6 +3925,7 @@ mod tests {
1, 1,
&dest_root.to_string_lossy(), &dest_root.to_string_lossy(),
&pack_dir, &pack_dir,
None,
) )
.unwrap(); .unwrap();
@ -3941,6 +4002,7 @@ mod tests {
1, 1,
&dest_root.to_string_lossy(), &dest_root.to_string_lossy(),
&pack_dir, &pack_dir,
None,
) )
.unwrap(); .unwrap();
@ -3956,6 +4018,63 @@ mod tests {
std::fs::remove_dir_all(&dir).unwrap(); 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] #[test]
fn import_season_pack_fails_when_nothing_in_it_matches_a_tracked_episode() { fn import_season_pack_fails_when_nothing_in_it_matches_a_tracked_episode() {
let conn = seeded_season_pack_conn(2); let conn = seeded_season_pack_conn(2);
@ -3978,6 +4097,7 @@ mod tests {
1, 1,
&dest_root.to_string_lossy(), &dest_root.to_string_lossy(),
&pack_dir, &pack_dir,
None,
); );
assert!(result.is_err()); assert!(result.is_err());

1
breadarrd/src/jellyfin.rs
View file

@ -1,5 +1,6 @@
use anyhow::{bail, Context, Result}; use anyhow::{bail, Context, Result};
#[derive(Clone)]
pub struct JellyfinClient { pub struct JellyfinClient {
base_url: String, base_url: String,
api_key: String, api_key: String,

51
breadarrd/src/main.rs
View file

@ -358,6 +358,16 @@ async fn background_loop(
let mut upgrade_ticker = tokio::time::interval(std::time::Duration::from_secs( let mut upgrade_ticker = tokio::time::interval(std::time::Duration::from_secs(
config.sources.upgrade_poll_interval_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( let mut transcode_ticker = tokio::time::interval(std::time::Duration::from_secs(
config.transcode.poll_interval_secs, config.transcode.poll_interval_secs,
)); ));
@ -420,7 +430,7 @@ async fn background_loop(
_ = import_ticker.tick() => { _ = import_ticker.tick() => {
let result = { let result = {
let conn = conn.lock().await; 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, &notifier) { if let (Ok(stats), Some(n)) = (&result, &notifier) {
if stats.failed > 0 { if stats.failed > 0 {
@ -553,18 +563,33 @@ async fn background_loop(
if !config.transcode.enabled { if !config.transcode.enabled {
continue; continue;
} }
let result = transcode::run_cycle(conn.clone(), config.transcode.clone(), jellyfin.as_ref()).await; let Ok(busy_permit) = transcode_busy.clone().try_lock_owned() else {
let record = match &result { // A previous cycle is still running (this file's
Ok(stats) => { // encode took longer than one poll interval) — skip
info!(?stats, "transcode cycle complete"); // this tick rather than spawning a second overlapping
api::CycleRecord { at: chrono::Utc::now(), ok: true, detail: format!("{stats:?}") } // one; the still-running cycle will pick up any newly
} // pending jobs on its own next iteration anyway.
Err(e) => { continue;
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); 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() => { _ = reconcile_ticker.tick() => {
let result = { let result = {
@ -879,7 +904,7 @@ async fn debug_import_cycle(config: &Config) -> Result<()> {
&config.qbit.category, &config.qbit.category,
&config.qbit.container_downloads_path, &config.qbit.container_downloads_path,
&config.qbit.host_downloads_path, &config.qbit.host_downloads_path,
config.transcode.enabled, config.transcode.enabled.then_some(&config.transcode),
) )
.await?; .await?;
println!("{stats:?}"); println!("{stats:?}");

202
breadarrd/src/transcode/mod.rs
View file

@ -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 /// Blocking and slow by design (a real GPU encode, potentially minutes per
/// file) — callers must run this inside `tokio::task::spawn_blocking`, never /// file) — callers must run this inside `tokio::task::spawn_blocking`, never
/// directly on an async task, and never while holding the shared DB mutex. /// 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<()> { fn run_ffmpeg_encode(input: &Path, output: &Path, bitrate_kbps: u32, vaapi_device: &str) -> Result<()> {
let maxrate = bitrate_kbps * 3 / 2; let maxrate = bitrate_kbps * 3 / 2;
let bufsize = bitrate_kbps * 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) .arg(input)
.args(["-map", "0"]) .args(["-map", "0"])
.args(["-c:v", "av1_vaapi"]) .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(["-b:v", &format!("{bitrate_kbps}k")])
.args(["-maxrate", &format!("{maxrate}k")]) .args(["-maxrate", &format!("{maxrate}k")])
.args(["-bufsize", &format!("{bufsize}k")]) .args(["-bufsize", &format!("{bufsize}k")])
@ -68,6 +84,24 @@ fn run_ffmpeg_encode(input: &Path, output: &Path, bitrate_kbps: u32, vaapi_devic
Ok(()) 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 /// The blocking half of a transcode: encode to a temp file alongside the
/// original (same filesystem, required for the atomic rename-based swap /// original (same filesystem, required for the atomic rename-based swap
/// later — never a separate staging drive), then verify the result is /// 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 /// 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 /// lock); cleans it up itself on any failure, so the original is never at
/// risk regardless of what goes wrong here. /// 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( fn encode_and_verify(
input: PathBuf, input: PathBuf,
job_id: i64,
cfg: TranscodeConfig, cfg: TranscodeConfig,
width: i64, width: i64,
height: i64, height: i64,
original_duration: Option<f64>, original_duration: Option<f64>,
) -> Result<PathBuf> { ) -> Result<Option<PathBuf>> {
let tmp_path = input.with_extension("av1.mkv"); 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); let bitrate = target_bitrate_kbps(width, height, &cfg);
run_ffmpeg_encode(&input, &tmp_path, bitrate, &cfg.vaapi_device)?; 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 /// 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<bool> {
/// leftover state, they're just still running. Left unreset, a crash leaves /// leftover state, they're just still running. Left unreset, a crash leaves
/// that job's slot permanently uncountable-but-also-unclaimable, quietly /// that job's slot permanently uncountable-but-also-unclaimable, quietly
/// shrinking real capacity forever instead of just costing one retry. /// 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<usize> { pub fn reset_orphaned_running_jobs(conn: &Connection) -> Result<usize> {
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::<rusqlite::Result<Vec<_>>>()?
};
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( let reset = conn.execute(
"UPDATE transcode_job SET status = 'pending' WHERE status = 'running'", "UPDATE transcode_job SET status = 'pending' WHERE status = 'running'",
[], [],
@ -177,6 +254,38 @@ pub fn enqueue(
Ok(()) 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<i64>,
cfg: &TranscodeConfig,
) -> Result<bool> {
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: /// Every existing-library file eligible for the `transcode-library` backfill:
/// not already AV1, not anime, not HDR/2160p+ (per `cfg.exclude_hdr` / /// 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), /// `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' WHERE p.video_codec IS NOT NULL AND p.video_codec != 'av1'
AND (ef.upgrade_locked IS NULL OR ef.upgrade_locked = 0) AND (ef.upgrade_locked IS NULL OR ef.upgrade_locked = 0)
AND (?1 = 0 OR p.hdr = 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 ( AND NOT (
(m.tvdb_id IS NOT NULL AND m.tvdb_id IN (m.tvdb_id IS NOT NULL AND m.tvdb_id IN
(SELECT tvdb_id FROM anime_mapping WHERE tvdb_id IS NOT NULL)) (SELECT tvdb_id FROM anime_mapping WHERE tvdb_id IS NOT NULL))
@ -319,10 +428,31 @@ async fn claim_pending_jobs(conn: &Arc<Mutex<Connection>>, limit: usize) -> Resu
/// AV1 ground truth — same shape as `remux_one_backlog_file`. On failure, /// AV1 ground truth — same shape as `remux_one_backlog_file`. On failure,
/// the original is left completely untouched; the job is marked `failed` /// the original is left completely untouched; the job is marked `failed`
/// with the error recorded, no automatic retry. /// with the error recorded, no automatic retry.
async fn finalize_job(conn: &Arc<Mutex<Connection>>, job: ClaimedJob, encode_result: Result<PathBuf>) -> Result<TranscodeOutcome> { ///
/// `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<Mutex<Connection>>,
job: ClaimedJob,
encode_result: Result<Option<PathBuf>>,
) -> Result<TranscodeOutcome> {
let conn = conn.lock().await; let conn = conn.lock().await;
match encode_result { let tmp_path = match &encode_result {
Ok(tmp_path) => { Ok(Some(p)) => Some(p.clone()),
_ => None,
};
let swap: Result<TranscodeOutcome> = 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(); let original_bytes = std::fs::metadata(&job.path)?.len();
std::fs::rename(&tmp_path, &job.path)?; std::fs::rename(&tmp_path, &job.path)?;
let new_bytes = std::fs::metadata(&job.path)?.len(); let new_bytes = std::fs::metadata(&job.path)?.len();
@ -336,14 +466,23 @@ async fn finalize_job(conn: &Arc<Mutex<Connection>>, job: ClaimedJob, encode_res
params![job.episode_file_id], params![job.episode_file_id],
)?; )?;
importer::ensure_probed(&conn, job.episode_file_id, &job.path)?; 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( conn.execute(
"UPDATE transcode_job SET status = 'done', new_bytes = ?1, finished_at = datetime('now') WHERE id = ?2", "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(outcome)
Ok(TranscodeOutcome { original_bytes, new_bytes })
} }
Err(e) => { Err(e) => {
if let Some(tmp_path) = tmp_path {
let _ = std::fs::remove_file(&tmp_path);
}
conn.execute( conn.execute(
"UPDATE transcode_job SET status = 'failed', error = ?1, finished_at = datetime('now') WHERE id = ?2", "UPDATE transcode_job SET status = 'failed', error = ?1, finished_at = datetime('now') WHERE id = ?2",
params![e.to_string(), job.job_id], params![e.to_string(), job.job_id],
@ -410,9 +549,11 @@ pub async fn run_cycle(
for job in claimed { for job in claimed {
let cfg = cfg.clone(); let cfg = cfg.clone();
let path = job.path.clone(); let path = job.path.clone();
let job_id = job.job_id;
let (width, height, duration) = (job.width, job.height, job.duration_secs); let (width, height, duration) = (job.width, job.height, job.duration_secs);
let encode_handle = let encode_handle = tokio::task::spawn_blocking(move || {
tokio::task::spawn_blocking(move || encode_and_verify(path, cfg, width, height, duration)); encode_and_verify(path, job_id, cfg, width, height, duration)
});
handles.push((job, encode_handle)); handles.push((job, encode_handle));
} }
@ -543,6 +684,43 @@ mod tests {
assert!(claimed.is_empty(), "already at the cap — must not claim more"); 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] #[test]
fn target_bitrate_matches_reference_at_reference_resolution() { fn target_bitrate_matches_reference_at_reference_resolution() {
let bitrate = target_bitrate_kbps(1920, 1080, &cfg()); let bitrate = target_bitrate_kbps(1920, 1080, &cfg());