From 0b37a245c72fec4fec83ee81c03db2cab20a270c Mon Sep 17 00:00:00 2001 From: houssemko Date: Fri, 18 Sep 2026 08:51:44 +0100 Subject: [PATCH 1/2] Merge via direct ffmpeg and spawn extraction without the crate executor --- src/video.rs | 201 +++++++++++++++++++++++++++++++++++++++++++-- src/video_tests.rs | 155 ++++++++++++++++++++++++++++++++++ 2 files changed, 347 insertions(+), 9 deletions(-) diff --git a/src/video.rs b/src/video.rs index 18c8a1df..3a5c50ac 100644 --- a/src/video.rs +++ b/src/video.rs @@ -1063,10 +1063,69 @@ async fn fetch_video_page( // or malformed URL can never be read as a flag. args.push("--".to_string()); args.push(url.to_string()); - let executor = yt_dlp::executor::Executor::new(youtube_bin, args, timeout); - let output = executor.execute().await.map_err(VideoError::fetch)?; + // Spawned directly (tokio + timeout) rather than through the + // crate's executor: same semantics — concurrent pipe drain, + // timeout kill, nonzero exit as error — with the failure detail + // taken from stderr instead of a wrapped crate error. + let mut cmd = tokio::process::Command::new(youtube_bin); + cmd.args(&args); + cmd.stdin(std::process::Stdio::null()) + .stdout(std::process::Stdio::piped()) + .stderr(std::process::Stdio::piped()); + #[cfg(unix)] + { + cmd.process_group(0); + } + let mut child = cmd.spawn().map_err(VideoError::fetch)?; + let stdout = child + .stdout + .take() + .ok_or_else(|| VideoError::fetch("yt-dlp gave no output pipe"))?; + let stderr = child + .stderr + .take() + .ok_or_else(|| VideoError::fetch("yt-dlp gave no log pipe"))?; + // Drain both pipes concurrently: `--dump-single-json` output is + // megabytes, and an unread pipe would stall yt-dlp once full. + fn drain( + stream: impl tokio::io::AsyncRead + Unpin + Send + 'static, + ) -> tokio::task::JoinHandle> { + tokio::spawn(async move { + use tokio::io::AsyncReadExt as _; + let mut buf = Vec::new(); + let mut reader = tokio::io::BufReader::new(stream); + reader.read_to_end(&mut buf).await.ok(); + buf + }) + } + let out_task = drain(stdout); + let err_task = drain(stderr); + let status = match tokio::time::timeout(timeout, child.wait()).await { + Ok(Ok(status)) => status, + Ok(Err(e)) => { + kill_tree(&mut child); + return Err(VideoError::fetch(&e)); + } + Err(_) => { + kill_tree(&mut child); + let _ = child.wait().await; + return Err(VideoError::fetch(gettext("the lookup timed out"))); + } + }; + let stdout = out_task.await.unwrap_or_default(); + let stderr = err_task.await.unwrap_or_default(); + if !status.success() { + let detail = String::from_utf8_lossy(&stderr) + .lines() + .rev() + .find(|l| !l.trim().is_empty()) + .unwrap_or("yt-dlp reported failure") + .trim() + .to_string(); + return Err(VideoError::fetch(detail)); + } let mut value: serde_json::Value = - serde_json::from_str(&output.stdout).map_err(VideoError::fetch)?; + serde_json::from_str(&String::from_utf8_lossy(&stdout)).map_err(VideoError::fetch)?; sanitize_video_json(&mut value); let mut video: Video = serde_json::from_value(value).map_err(VideoError::fetch)?; for format in &mut video.formats { @@ -2568,12 +2627,14 @@ pub async fn run_video_download( phase(gettext("Merging…")); } finish_merge( - &downloader, + &ffmpeg_bin, &staging, vpart.as_deref(), &apart, &job.dest, video_sel.as_ref().map(|s| s.ext.as_str()), + &video.title, + timeout, ) .await .map(Some) @@ -2591,21 +2652,143 @@ pub async fn run_video_download( /// Merge verified parts and rename the result into place. Audio-only rows /// adopt the audio part directly (no ffmpeg round-trip). +/// Audio codec for the merge: stream-copy when the container takes +/// the part's codec, AAC re-encode otherwise. Ported from the crate's +/// `audio_codec_for_mux` (whose `audio_codec_hint` was always `None` +/// at our call site), kept as a pure function so the matrix stays +/// unit-tested. +fn merge_audio_codec(audio_ext: &str, output_ext: &str) -> &'static str { + let is_aac = matches!(audio_ext, "m4a" | "aac"); + let is_opus = matches!(audio_ext, "webm" | "opus" | "ogg"); + match output_ext { + "mp4" | "m4a" | "mov" if is_aac => "copy", + "webm" if is_opus => "copy", + // Matroska supports any codec natively + "mkv" | "mka" => "copy", + _ => "aac", + } +} + +/// ffmpeg argv merging split parts: `-i` audio first (so `0:a` is the +/// audio part), stream-copy video, title embedded in the same pass. +/// Pure for tests: inputs, mapping and the end-of-options separator +/// are pinned here, not in assertion-hostile spawn code. +fn merge_argv(audio_part: &Path, video_part: &Path, out: &Path, title: &str) -> Vec { + let ext = |p: &Path| { + p.extension() + .and_then(|e| e.to_str()) + .unwrap_or("") + .to_lowercase() + }; + vec![ + "-hide_banner".to_string(), + "-loglevel".to_string(), + "warning".to_string(), + "-nostats".to_string(), + "-i".to_string(), + audio_part.to_string_lossy().into_owned(), + "-i".to_string(), + video_part.to_string_lossy().into_owned(), + "-map".to_string(), + "0:a".to_string(), + "-map".to_string(), + "1:v".to_string(), + "-c:v".to_string(), + "copy".to_string(), + "-c:a".to_string(), + merge_audio_codec(&ext(audio_part), &ext(out)).to_string(), + "-metadata".to_string(), + format!("title={title}"), + "--".to_string(), + out.to_string_lossy().into_owned(), + ] +} + +/// Run the merge through ffmpeg directly: same command the crate's +/// combine step built, minus the builder round-trip — and failures now +/// surface ffmpeg's own last line instead of a wrapped crate error. +async fn run_merge_ffmpeg( + ffmpeg_bin: &Path, + audio_part: &Path, + video_part: &Path, + out: &Path, + title: &str, + timeout: Duration, +) -> Result<(), VideoError> { + let mut cmd = tokio::process::Command::new(ffmpeg_bin); + cmd.args(merge_argv(audio_part, video_part, out, title)); + cmd.stdin(std::process::Stdio::null()) + .stdout(std::process::Stdio::piped()) + .stderr(std::process::Stdio::piped()); + #[cfg(unix)] + { + cmd.process_group(0); + } + let mut child = cmd.spawn().map_err(VideoError::runtime)?; + let stderr = child + .stderr + .take() + .ok_or_else(|| VideoError::runtime("ffmpeg gave no log pipe"))?; + let logs = tokio::spawn(async move { + use tokio::io::AsyncReadExt as _; + let mut reader = tokio::io::BufReader::new(stderr); + let mut tail = Vec::new(); + let mut buf = [0u8; 4096]; + loop { + match reader.read(&mut buf).await { + Ok(0) | Err(_) => break, + Ok(n) => { + tail.extend_from_slice(&buf[..n]); + if tail.len() > 8192 { + tail.drain(..tail.len() - 8192); + } + } + } + } + String::from_utf8_lossy(&tail).into_owned() + }); + let status = match tokio::time::timeout(timeout, child.wait()).await { + Ok(Ok(status)) => status, + Ok(Err(e)) => { + kill_tree(&mut child); + let _ = logs.await; + return Err(VideoError::runtime(&e)); + } + Err(_) => { + kill_tree(&mut child); + let _ = logs.await; + return Err(VideoError::part_failed("timed out merging")); + } + }; + let log_tail = logs.await.unwrap_or_default(); + if !status.success() { + let detail = log_tail + .lines() + .rev() + .find(|l| !l.trim().is_empty()) + .unwrap_or("ffmpeg reported failure") + .trim() + .to_string(); + return Err(VideoError::combine(detail)); + } + Ok(()) +} + +#[allow(clippy::too_many_arguments)] async fn finish_merge( - downloader: &Downloader, + ffmpeg_bin: &Path, staging: &Path, vpart: Option<&Path>, apart: &Path, dest: &Path, video_ext: Option<&str>, + title: &str, + timeout: Duration, ) -> Result { let final_tmp: PathBuf = match (vpart, video_ext) { (Some(v), Some(ext)) => { let muxed = staging.join(format!("muxed.{ext}")); - downloader - .combine_audio_and_video_to_path(apart, v, &muxed) - .await - .map_err(VideoError::combine)?; + run_merge_ffmpeg(ffmpeg_bin, apart, v, &muxed, title, timeout).await?; muxed } // Audio-only: the part already is the finished file (container diff --git a/src/video_tests.rs b/src/video_tests.rs index 419e4955..5828d9f2 100644 --- a/src/video_tests.rs +++ b/src/video_tests.rs @@ -2059,3 +2059,158 @@ fn hls_selection_ignores_extractor_order() { "https://cdn.example/h1080" ); } + +// ── direct ffmpeg merge ────────────────────────────────────────────── + +#[test] +fn merge_audio_codec_matrix() { + // Mirrors the crate's audio_codec_for_mux without its builder. + assert_eq!(merge_audio_codec("m4a", "mp4"), "copy"); + assert_eq!(merge_audio_codec("aac", "mp4"), "copy"); + assert_eq!(merge_audio_codec("webm", "webm"), "copy"); + assert_eq!(merge_audio_codec("opus", "webm"), "copy"); + assert_eq!(merge_audio_codec("mka", "mka"), "copy"); + assert_eq!(merge_audio_codec("webm", "mkv"), "copy"); + // Opus into MP4 (the YouTube split case) re-encodes to AAC. + assert_eq!(merge_audio_codec("webm", "mp4"), "aac"); + assert_eq!(merge_audio_codec("ogg", "mp4"), "aac"); +} + +#[test] +fn merge_argv_maps_audio_first() { + let argv = merge_argv( + std::path::Path::new("/tmp/st/audio.webm"), + std::path::Path::new("/tmp/st/video.mp4"), + std::path::Path::new("/tmp/st/muxed.mp4"), + "Some Title", + ); + let inputs: Vec<&str> = argv + .windows(2) + .filter(|w| w[0] == "-i") + .map(|w| w[1].as_str()) + .collect(); + assert_eq!(inputs, ["/tmp/st/audio.webm", "/tmp/st/video.mp4"]); + assert!(argv.windows(2).any(|w| w == ["-map", "0:a"])); + assert!(argv.windows(2).any(|w| w == ["-map", "1:v"])); + assert!(argv.windows(2).any(|w| w == ["-c:v", "copy"])); + assert!(argv.windows(2).any(|w| w == ["-c:a", "aac"])); + assert!(argv.contains(&"title=Some Title".to_string())); + assert_eq!(argv[argv.len() - 2], "--"); + assert_eq!(argv[argv.len() - 1], "/tmp/st/muxed.mp4"); +} + +fn fake_ffmpeg_merge(dir: &std::path::Path, fail: bool) -> std::path::PathBuf { + let bin = dir.join("fake-ffmpeg"); + std::fs::write( + &bin, + if fail { + r#"#!/bin/sh +echo "[out#0/mp4] Invalid data found when processing input" >&2 +exit 1 +"# + } else { + r#"#!/bin/sh +out="" +prev="" +for a in "$@"; do + if [ "$prev" = "--" ]; then out="$a"; fi + prev="$a" +done +echo "$@" >> "$out.argv.log" +printf 'merged' > "$out" +exit 0 +"# + }, + ) + .unwrap(); + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt as _; + std::fs::set_permissions(&bin, std::fs::Permissions::from_mode(0o755)).unwrap(); + } + bin +} + +#[test] +fn merge_binary_combines_and_reports_failure() { + let dir = std::env::temp_dir().join(format!("grab-fakemerge-{}", std::process::id())); + let _ = std::fs::remove_dir_all(&dir); + std::fs::create_dir_all(&dir).unwrap(); + let fake = fake_ffmpeg_merge(&dir, false); + let (apart, vpart, out) = (dir.join("a.webm"), dir.join("v.mp4"), dir.join("muxed.mp4")); + std::fs::write(&apart, b"a").unwrap(); + std::fs::write(&vpart, b"v").unwrap(); + let res = crate::download::tokio_rt().block_on(run_merge_ffmpeg( + &fake, + &apart, + &vpart, + &out, + "T", + std::time::Duration::from_secs(30), + )); + assert!(matches!(res, Ok(())), "got {res:?}"); + assert_eq!(std::fs::read(&out).unwrap(), b"merged"); + + let fail = fake_ffmpeg_merge(&dir, true); + let res = crate::download::tokio_rt().block_on(run_merge_ffmpeg( + &fail, + &apart, + &vpart, + &dir.join("muxed2.mp4"), + "T", + std::time::Duration::from_secs(30), + )); + match res { + Err(e) => assert!( + e.to_string().contains("Invalid data found"), + "real ffmpeg message surfaces: {e}" + ), + ok => panic!("expected combine error, got {ok:?}"), + } + let _ = std::fs::remove_dir_all(&dir); +} + +// ── extraction spawn (no crate executor) ───────────────────────────── + +#[test] +fn fetch_video_page_parses_dump_json() { + // Fake yt-dlp emitting --dump-single-json bytes: proves the direct + // spawn, concurrent drain and parse path without network. + let dir = std::env::temp_dir().join(format!("grab-fakeextract-{}", std::process::id())); + let _ = std::fs::remove_dir_all(&dir); + std::fs::create_dir_all(&dir).unwrap(); + std::fs::write( + dir.join("info.json"), + serde_json::json!({ + "id": "abc", + "title": "T", + "age_limit": 0, + "live_status": "not_live", + "playable_in_embed": true, + "extractor": "generic", + "extractor_key": "Generic", + "_version": {"version": "2026.08.19", "repository": "yt-dlp"}, + "formats": [], + }) + .to_string(), + ) + .unwrap(); + let bin = dir.join("fake-ytdlp"); + std::fs::write(&bin, "#!/bin/sh\ncat \"$(dirname \"$0\")/info.json\"\n").unwrap(); + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt as _; + std::fs::set_permissions(&bin, std::fs::Permissions::from_mode(0o755)).unwrap(); + } + let video = crate::download::tokio_rt() + .block_on(fetch_video_page( + &bin, + "https://example.com/v", + "none", + std::time::Duration::from_secs(30), + )) + .expect("fake extract parses"); + assert_eq!(video.id, "abc"); + assert!(video.formats.is_empty()); + let _ = std::fs::remove_dir_all(&dir); +} From f59462ade8ca66891d78db24ddb9b62ea8fb6a17 Mon Sep 17 00:00:00 2001 From: houssemko Date: Fri, 18 Sep 2026 09:04:33 +0100 Subject: [PATCH 2/2] Repair web-merge damage: restore eaten test tails, drop unused Downloader --- src/video.rs | 19 +++---------------- src/video_tests.rs | 13 +++++++++++++ 2 files changed, 16 insertions(+), 16 deletions(-) diff --git a/src/video.rs b/src/video.rs index 77dd8f5e..ac493a16 100644 --- a/src/video.rs +++ b/src/video.rs @@ -28,7 +28,6 @@ use std::sync::{ use std::time::{Duration, SystemTime, UNIX_EPOCH}; use thiserror::Error; use tokio::sync::oneshot; -use yt_dlp::Downloader; use yt_dlp::client::deps::{Libraries, LibraryInstaller}; use yt_dlp::model::format::{Extension, Format, FormatType, HttpHeaders, Protocol}; use yt_dlp::model::selector::{ @@ -2326,24 +2325,12 @@ pub async fn run_video_download( ffmpeg = %ff_version, "starting video attempt" ); - // The downloader timeout covers the ffmpeg merge: a full-length - // merge on a slow CPU dwarfs any network timeout, so never go below - // the crate default (the user's setting extends it). Metadata uses - // the crate's extractor timeout via [`fetch_video_page`]. + // Attempt timeout floor: a full-length merge on a slow CPU dwarfs + // any network timeout, so never go below the crate default (the + // user's setting extends it). let timeout = Duration::from_secs(job.timeout_secs.max(300)); let youtube_bin = libs.youtube.clone(); let ffmpeg_bin = libs.ffmpeg.clone(); - let mut builder = Downloader::builder(libs, staging.clone()).with_timeout(timeout); - if !job.user_agent.is_empty() { - builder = builder.with_user_agent(job.user_agent.clone()); - } - // Authenticated extraction for gated pages via the browser profile. - // (Part downloads go through the binary, which carries the - // extraction cookies and format headers itself.) - if let Some(spec) = cookies_browser_spec(&job.cookies_browser) { - builder = builder.with_cookies_from_browser(spec); - } - let downloader = builder.build().await.map_err(VideoError::fetch)?; let phase = |text: String| { tx.send(EngineMsg::Phase(text)).ok(); }; diff --git a/src/video_tests.rs b/src/video_tests.rs index 11bbebb4..ad39b745 100644 --- a/src/video_tests.rs +++ b/src/video_tests.rs @@ -2121,6 +2121,15 @@ printf 'merged' > "$out" exit 0 "# }, + ) + .unwrap(); + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt as _; + std::fs::set_permissions(&bin, std::fs::Permissions::from_mode(0o755)).unwrap(); + } + bin +} // ── binary part downloads ──────────────────────────────────────────── fn part_test_job() -> VideoJob { @@ -2314,6 +2323,10 @@ fn fetch_video_page_parses_dump_json() { .expect("fake extract parses"); assert_eq!(video.id, "abc"); assert!(video.formats.is_empty()); + let _ = std::fs::remove_dir_all(&dir); +} + +#[test] fn part_binary_download_reports_progress() { let dir = std::env::temp_dir().join(format!("grab-fakeyt-{}", std::process::id())); let _ = std::fs::remove_dir_all(&dir);