clusterfuck
/
crabidy
4
1
Fork 0
Code Issues 15 Pull Requests Actions Packages Projects 1 Releases 5 Wiki Activity

Fetch streams in bounded windows and restore yt-dlp for stream URLs 

YouTube caps tokenless stream URLs at exactly their leading 1 MiB:
plain, open-ended, and oversized requests get 403, and fresh URLs
refuse offset starts, so playback died mid-first-minute. PO tokens
would lift the cap but the token-capable Innertube clients need
signature deciphering that is broken in rustypipe upstream (botguard
was built and tested — ineffective through the iOS client).

The player now streams every http(s) source through a windowed
SourceStream (bounded ~1 MiB ranges, 200-body fallback, eager
seek/reconnect so rejected windows fail typed instead of retrying
forever) and the capture downloader windows the same way. Stream URLs
come from a minimal yt-dlp sidecar again — metadata stays on the
pure-Rust rustypipe extractor — whose cipher-solved URLs stream whole
files at a throttled ~32 KB/s; a missing binary degrades to 1 MiB
streams with a warning. botguard_bin is wired through so streams flip
back to pure Rust when upstream deciphering recovers. Live-verified on
the exact track from the failure log.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Test User 2026-07-21 18:18:40 +02:00
parent 6f787285bf
commit 973bb7bbc6
17 changed files with 1240 additions and 39 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
Cargo.lock generated
View File

@ -303,7 +303,10 @@ name = "audio-player"
version = "0.1.0"
dependencies = [
"anyhow",
"bytes",
"flume",
"futures",
"reqwest 0.13.1",
"rodio",
"stream-download",
"thiserror 2.0.19",

1
Cargo.toml
View File

@ -19,6 +19,7 @@ edition = "2021"
anyhow = "1"
async-trait = "0.1"
base64 = "0.22"
bytes = "1"
chrono = { version = "0.4", default-features = false, features = ["clock"] }
clap = { version = "4", features = ["derive"] }
clap-serde-derive = "0.2"

50
architecture/youtube-rustypipe.md
View File

@ -1,4 +1,9 @@
# YouTube provider on rustypipe (pure Rust)
# YouTube provider on rustypipe
> **Amended the same day** by D2-revised below: pure-Rust stream
> *fetching* turned out to be blocked by YouTube's PO-token
> enforcement, so stream URLs go through a minimal `yt-dlp` sidecar
> again while all metadata stays on rustypipe.
## Context and problem statement
@ -74,6 +79,49 @@ likely not decode locally, but the URL is still honest — e.g. a future
player may cope). Download captures inherit the same choice, so their
audio is `.m4a`.
## D2-revised — Stream *fetching* (2026-07-21, second round)
Real use immediately hit YouTube's tokenless-fetch enforcement. Live
findings against googlevideo (all reproduced, same day):
- Every rustypipe (iOS-client) stream URL serves **exactly its leading
~1 MiB**, measured to the byte (403 at cumulative 1 048 576): plain
GETs 403, open-ended ranges 403, bounded ranges work only until the
budget is spent, and a **fresh URL refuses to start at an offset** —
so chaining fresh URLs per window is impossible.
- PO tokens would lift the cap, but rustypipe attaches them only to
web-family clients, whose **signature deciphering is broken upstream
right now** ("could not extract sig fn name", also on git master) —
`rustypipe-botguard` was built and tested; ineffective through the
iOS client (no `pot` parameter).
- `yt-dlp` (2026.07.04) still solves the cipher challenges; its URLs
accept plain and ranged GETs for the whole file, **throttled to
~32 KB/s** — twice the itag-140 audio bitrate, so playback holds and
captures are merely slow. Its TV/Android clients are SABR-blocked
(no URLs at all), so this is the state of the art everywhere.
**Decisions:**
1. **Windowed HTTP fetching everywhere.** The player streams through a
`WindowedHttpStream` (audio-player, a `stream-download`
`SourceStream`) and the capture `Downloader` downloads in the same
bounded ~1 MiB `Range` windows — the request pattern real players
produce, correct for every provider (servers that ignore `Range`
degrade to one 200 body; rejected windows fail typed, never loop).
2. **`yt-dlp` returns as a stream-URL-only sidecar.** All metadata
(search, playlists, details) stays on the pure-Rust rustypipe
extractor; only `get_urls_for_track` consults `yt-dlp` (argv-only,
bounded, stderr-summarized). A missing binary degrades with a
warning — streams then stop after their first ~1 MiB instead of
failing entirely.
3. **The pure-Rust path stays wired.** `botguard_bin` is configurable
(and PATH-auto-detected by rustypipe); when upstream deciphering
recovers, PO-token'd rustypipe URLs make the sidecar unnecessary
without code changes beyond removing the fallback preference.
The per-track capture deadline was raised to 30 minutes — at 32 KB/s a
long track legitimately takes that.
## D3 — Login and playlists
The `cookies` setting keeps its meaning: a path to a Netscape

3
audio-player/Cargo.toml
View File

@ -5,7 +5,10 @@ edition.workspace = true
[dependencies]
anyhow.workspace = true
bytes.workspace = true
flume.workspace = true
futures.workspace = true
reqwest.workspace = true
rodio.workspace = true
stream-download.workspace = true
thiserror.workspace = true

1
audio-player/src/lib.rs
View File

@ -1,5 +1,6 @@
mod player;
mod player_engine;
pub mod windowed_http;
pub use player::{Player, PlayerError};
pub use player_engine::{MediaInfo, PlayerMessage};

17
audio-player/src/player_engine.rs
View File

@ -10,6 +10,8 @@ use rodio::stream::{DeviceSinkBuilder, MixerDeviceSink};
use rodio::{Decoder, Source};
use stream_download::storage::temp::TempStorageProvider;
use stream_download::{Settings, StreamDownload};
use crate::windowed_http::{WindowedHttpParams, WindowedHttpStream};
use thiserror::Error;
use tracing::{debug, info, instrument, trace, warn};
use url::Url;
@ -78,6 +80,8 @@ pub struct PlayerEngine {
// Present when the engine had to bring its own runtime because the
// creating thread was not inside a tokio context.
_owned_runtime: Option<tokio::runtime::Runtime>,
/// Shared client for windowed network streams.
http: reqwest::Client,
}
impl PlayerEngine {
@ -101,6 +105,10 @@ impl PlayerEngine {
}
};
info!("audio output device opened");
let http = reqwest::Client::builder()
.connect_timeout(Duration::from_secs(30))
.build()
.context("failed to build the http client")?;
Ok(Self {
current_source: None,
media_info: None,
@ -111,6 +119,7 @@ impl PlayerEngine {
tx_player,
runtime,
_owned_runtime: owned_runtime,
http,
})
}
@ -195,11 +204,15 @@ impl PlayerEngine {
let duration = match Url::parse(source_str) {
Ok(url) if matches!(url.scheme(), "http" | "https") => {
trace!(%url, "opening network stream");
// Windowed fetching: some CDNs (googlevideo) reject plain
// and open-ended requests with 403 and only serve bounded
// ranges (see audio-player/src/windowed_http.rs).
let params = WindowedHttpParams::new(url.clone(), self.http.clone());
let reader = self.runtime.block_on(async {
tokio::time::timeout(
STREAM_OPEN_TIMEOUT,
StreamDownload::new_http(
url.clone(),
StreamDownload::new::<WindowedHttpStream>(
params,
TempStorageProvider::new(),
Settings::default(),
),

547
audio-player/src/windowed_http.rs Normal file
View File

@ -0,0 +1,547 @@
//! A windowed HTTP [`SourceStream`]: fetches media in small **bounded**
//! `Range` requests instead of one open-ended GET.
//!
//! Some CDNs (notably googlevideo, see
//! `architecture/youtube-rustypipe.md`) reject plain and open-ended
//! requests from unattested clients with `403 Forbidden` and only serve
//! bounded ranges of about a megabyte — the request pattern real players
//! produce. This stream chains such windows transparently; servers that
//! ignore the `Range` header (plain `200`) degrade to one continuous
//! body without windowing (and without seek support).
//!
//! Error messages never include the URL — stream URLs may embed tokens.
use std::io;
use std::pin::Pin;
use std::task::{Context, Poll};
use bytes::Bytes;
use futures::{Future, Stream};
use stream_download::source::{DecodeError, SourceStream};
use tracing::{debug, trace, warn};
use url::Url;
/// Bytes per request window. Verified against googlevideo: 1 MiB windows
/// are served, 8 MiB and open-ended requests are rejected.
pub const WINDOW_SIZE: u64 = 1024 * 1024;
/// Parameters for [`WindowedHttpStream::create`].
#[derive(Clone, Debug)]
pub struct WindowedHttpParams {
pub url: Url,
pub client: reqwest::Client,
/// Window size in bytes; [`WINDOW_SIZE`] outside of tests.
pub window: u64,
}
impl WindowedHttpParams {
pub fn new(url: Url, client: reqwest::Client) -> Self {
Self {
url,
client,
window: WINDOW_SIZE,
}
}
}
/// Error creating the stream (first window request failed).
#[derive(Debug, thiserror::Error)]
#[error("{0}")]
pub struct WindowedHttpError(String);
impl DecodeError for WindowedHttpError {}
type BytesStream = Pin<Box<dyn Stream<Item = Result<Bytes, reqwest::Error>> + Send + Sync>>;
type WindowFuture = Pin<Box<dyn Future<Output = io::Result<Window>> + Send + Sync>>;
/// One server response being drained.
struct Window {
stream: BytesStream,
/// Absolute end (exclusive) of the bytes this response carries;
/// `u64::MAX` for an un-ranged whole-body response.
end: u64,
/// Total resource size, when the response revealed it.
total: Option<u64>,
/// The server honored the `Range` header (`206`).
ranged: bool,
}
enum State {
/// Draining the current response.
Streaming {
stream: BytesStream,
end: u64,
},
/// Waiting for the next window request.
Requesting(WindowFuture),
Finished,
}
/// See the module docs.
pub struct WindowedHttpStream {
// (Debug impl below — the state machine holds unnameable futures.)
client: reqwest::Client,
url: Url,
window: u64,
/// Total resource size (from `Content-Range`); `None` when unknown.
content_length: Option<u64>,
/// The server honors ranges — windowing and seeking are available.
ranged: bool,
/// Absolute offset of the next byte to hand out.
position: u64,
/// Exclusive end requested via `seek_range`; `None` = to the end.
limit: Option<u64>,
/// Bytes to silently drop before yielding (un-ranged reconnect
/// catch-up).
discard: u64,
state: State,
}
/// Requests one bounded window `[start, end_exclusive)`.
///
/// `206` yields a ranged window (end and total parsed from
/// `Content-Range`, falling back to `Content-Length`); `200` means the
/// server ignored the header and sent the whole body; `416` past the end
/// yields an empty terminal window. Anything else is an error carrying
/// the status only.
async fn request_window(
client: reqwest::Client,
url: Url,
start: u64,
end_exclusive: u64,
) -> io::Result<Window> {
let range = format!("bytes={start}-{}", end_exclusive.saturating_sub(1));
trace!(range, "requesting window");
let response = client
.get(url)
.header(reqwest::header::RANGE, range)
.send()
.await
.map_err(|err| io::Error::other(format!("window request failed: {}", err.without_url())))?;
match response.status() {
reqwest::StatusCode::PARTIAL_CONTENT => {
// `Content-Range: bytes <start>-<end>/<total|*>`
let content_range = response
.headers()
.get(reqwest::header::CONTENT_RANGE)
.and_then(|value| value.to_str().ok())
.map(str::to_string);
let (end, total) = match content_range.as_deref().and_then(parse_content_range) {
Some((_, range_end, total)) => (range_end + 1, total),
None => {
// No usable Content-Range: derive the window end from
// the body length.
let len = response.content_length().unwrap_or(0);
(start + len, None)
}
};
Ok(Window {
stream: Box::pin(response.bytes_stream()),
end,
total,
ranged: true,
})
}
reqwest::StatusCode::OK => {
let total = response.content_length();
Ok(Window {
stream: Box::pin(response.bytes_stream()),
end: u64::MAX,
total,
ranged: false,
})
}
reqwest::StatusCode::RANGE_NOT_SATISFIABLE => Ok(Window {
stream: Box::pin(futures::stream::empty()),
end: start,
total: None,
ranged: true,
}),
status => Err(io::Error::other(format!(
"window request rejected: {status}"
))),
}
}
/// Parses `bytes <start>-<end>/<total|*>` into `(start, end, total)`.
fn parse_content_range(value: &str) -> Option<(u64, u64, Option<u64>)> {
let rest = value.trim().strip_prefix("bytes ")?;
let (range, total) = rest.split_once('/')?;
let (start, end) = range.split_once('-')?;
let total = match total.trim() {
"*" => None,
n => Some(n.parse().ok()?),
};
Some((start.trim().parse().ok()?, end.trim().parse().ok()?, total))
}
impl std::fmt::Debug for WindowedHttpStream {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("WindowedHttpStream")
.field("content_length", &self.content_length)
.field("ranged", &self.ranged)
.field("position", &self.position)
.finish_non_exhaustive()
}
}
impl WindowedHttpStream {
/// The exclusive end the consumer currently wants: the seek limit
/// clamped to the known size.
fn effective_end(&self) -> Option<u64> {
match (self.limit, self.content_length) {
(Some(limit), Some(len)) => Some(limit.min(len)),
(Some(limit), None) => Some(limit),
(None, len) => len,
}
}
/// Schedules the request for the window starting at `start`, or
/// finishes when nothing is left to fetch.
fn schedule_window(&mut self, start: u64) {
let end = match self.window_end(start) {
Some(end) => end,
None => {
self.state = State::Finished;
return;
}
};
self.state = State::Requesting(Box::pin(request_window(
self.client.clone(),
self.url.clone(),
start,
end,
)));
}
/// The exclusive end of the window starting at `start`; `None` when
/// nothing is left to fetch.
fn window_end(&self, start: u64) -> Option<u64> {
match self.effective_end() {
Some(effective) if start >= effective => None,
Some(effective) => Some((start + self.window).min(effective)),
None => Some(start + self.window),
}
}
/// Eagerly opens the window at `start` (seeks and reconnects), so
/// request failures surface to the caller instead of re-arising on
/// every poll.
async fn open_window(&mut self, start: u64) -> io::Result<()> {
match self.window_end(start) {
None => {
self.state = State::Finished;
Ok(())
}
Some(end) => {
let window =
request_window(self.client.clone(), self.url.clone(), start, end).await?;
self.state = State::Streaming {
stream: window.stream,
end: window.end,
};
Ok(())
}
}
}
}
impl Stream for WindowedHttpStream {
type Item = io::Result<Bytes>;
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
let this = &mut *self;
loop {
match &mut this.state {
State::Finished => return Poll::Ready(None),
State::Requesting(future) => match future.as_mut().poll(cx) {
Poll::Pending => return Poll::Pending,
Poll::Ready(Err(err)) => {
this.state = State::Finished;
return Poll::Ready(Some(Err(err)));
}
Poll::Ready(Ok(window)) => {
this.state = State::Streaming {
stream: window.stream,
end: window.end,
};
}
},
State::Streaming { stream, end } => match stream.as_mut().poll_next(cx) {
Poll::Pending => return Poll::Pending,
Poll::Ready(Some(Err(err))) => {
return Poll::Ready(Some(Err(io::Error::other(format!(
"stream body failed: {}",
err.without_url()
)))));
}
Poll::Ready(Some(Ok(mut bytes))) => {
// Un-ranged reconnect catch-up: drop the prefix the
// consumer already has.
if this.discard > 0 {
let drop_now = this.discard.min(bytes.len() as u64) as usize;
let _ = bytes.split_to(drop_now);
this.discard -= drop_now as u64;
if bytes.is_empty() {
continue;
}
}
this.position += bytes.len() as u64;
return Poll::Ready(Some(Ok(bytes)));
}
Poll::Ready(None) => {
let window_end = *end;
if !this.ranged {
// One whole-body response: its end is the end.
this.state = State::Finished;
return Poll::Ready(None);
}
if let Some(effective) = this.effective_end() {
if this.position >= effective {
this.state = State::Finished;
return Poll::Ready(None);
}
}
if this.position < window_end && this.content_length.is_none() {
// A short window with no known total: the
// resource ended early.
this.state = State::Finished;
return Poll::Ready(None);
}
let next = this.position;
this.schedule_window(next);
}
},
}
}
}
}
impl SourceStream for WindowedHttpStream {
type Params = WindowedHttpParams;
type StreamCreationError = WindowedHttpError;
async fn create(params: Self::Params) -> Result<Self, Self::StreamCreationError> {
let window = request_window(params.client.clone(), params.url.clone(), 0, params.window)
.await
.map_err(|err| WindowedHttpError(err.to_string()))?;
let content_length = window.total;
let ranged = window.ranged;
debug!(?content_length, ranged, "windowed http stream open");
Ok(Self {
client: params.client,
url: params.url,
window: params.window,
content_length,
ranged,
position: 0,
limit: None,
discard: 0,
state: State::Streaming {
stream: window.stream,
end: window.end,
},
})
}
fn content_length(&self) -> Option<u64> {
self.content_length
}
async fn seek_range(&mut self, start: u64, end: Option<u64>) -> io::Result<()> {
trace!(start, ?end, "seek");
self.position = start;
self.limit = end;
self.discard = 0;
// Eager: the request happens *here*, so a rejected window is an
// error the retry logic can time out on — a lazily scheduled
// request that keeps failing would look like a successful
// reconnect every time and retry forever.
self.open_window(start).await
}
async fn reconnect(&mut self, current_position: u64) -> io::Result<()> {
if self.ranged {
self.position = current_position;
self.discard = 0;
return self.open_window(current_position).await;
}
// The server does not honor ranges: refetch from the start and
// drop what the consumer already has.
warn!(
current_position,
"reconnecting to a server without range support"
);
self.position = current_position;
self.discard = current_position;
let window = request_window(self.client.clone(), self.url.clone(), 0, u64::MAX).await?;
self.state = State::Streaming {
stream: window.stream,
end: window.end,
};
Ok(())
}
fn supports_seek(&self) -> bool {
self.ranged
}
}
#[cfg(test)]
mod tests {
use super::*;
use futures::StreamExt;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
/// A minimal HTTP server for one fixed body. With `ranged`, bounded
/// `Range` requests get `206` + `Content-Range` slices — and, like
/// googlevideo, open-ended or oversized ranges get `403`. Without,
/// every request gets the whole body as `200`.
async fn serve(body: Vec<u8>, ranged: bool, max_window: u64) -> Url {
let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
.await
.expect("bind test server");
let addr = listener.local_addr().expect("addr");
tokio::spawn(async move {
while let Ok((mut sock, _)) = listener.accept().await {
let body = body.clone();
tokio::spawn(async move {
let mut buf = Vec::new();
let mut chunk = [0u8; 1024];
loop {
match sock.read(&mut chunk).await {
Ok(0) | Err(_) => break,
Ok(n) => {
buf.extend_from_slice(&chunk[..n]);
if buf.windows(4).any(|w| w == b"\r\n\r\n") {
break;
}
}
}
}
let request = String::from_utf8_lossy(&buf).to_lowercase();
let range = request
.lines()
.find_map(|line| line.strip_prefix("range: bytes="))
.and_then(|spec| {
let (start, end) = spec.trim().split_once('-')?;
let start: u64 = start.parse().ok()?;
let end: Option<u64> = end.parse().ok();
Some((start, end))
});
let total = body.len() as u64;
let response = match (ranged, range) {
(true, Some((start, Some(end))))
if start < total && end - start < max_window =>
{
let end = end.min(total - 1);
let slice = &body[start as usize..=end as usize];
let mut head = format!(
"HTTP/1.1 206 Partial Content\r\nContent-Range: bytes {start}-{end}/{total}\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
slice.len()
)
.into_bytes();
head.extend_from_slice(slice);
head
}
(true, Some((start, _))) if start >= total => {
format!("HTTP/1.1 416 Range Not Satisfiable\r\nContent-Range: bytes */{total}\r\nContent-Length: 0\r\nConnection: close\r\n\r\n")
.into_bytes()
}
(true, _) => {
// Open-ended or oversized: rejected, like
// googlevideo without a PO token.
b"HTTP/1.1 403 Forbidden\r\nContent-Length: 0\r\nConnection: close\r\n\r\n".to_vec()
}
(false, _) => {
let mut head = format!(
"HTTP/1.1 200 OK\r\nContent-Length: {total}\r\nConnection: close\r\n\r\n"
)
.into_bytes();
head.extend_from_slice(&body);
head
}
};
let _ = sock.write_all(&response).await;
let _ = sock.shutdown().await;
});
}
});
Url::parse(&format!("http://{addr}/stream")).expect("url")
}
fn params(url: Url, window: u64) -> WindowedHttpParams {
WindowedHttpParams {
url,
client: reqwest::Client::new(),
window,
}
}
async fn read_all(stream: &mut WindowedHttpStream) -> Vec<u8> {
let mut out = Vec::new();
while let Some(chunk) = stream.next().await {
out.extend_from_slice(&chunk.expect("chunk"));
}
out
}
#[tokio::test]
async fn chains_windows_over_a_range_only_server() {
// 10 windows of 16 bytes against a server that 403s anything
// bigger — exactly the googlevideo behavior.
let body: Vec<u8> = (0..160u32).map(|i| i as u8).collect();
let url = serve(body.clone(), true, 64).await;
let mut stream = WindowedHttpStream::create(params(url, 16))
.await
.expect("create");
assert_eq!(stream.content_length(), Some(160));
assert!(stream.supports_seek());
assert_eq!(read_all(&mut stream).await, body);
}
#[tokio::test]
async fn seeks_restart_the_window_chain() {
let body: Vec<u8> = (0..160u32).map(|i| i as u8).collect();
let url = serve(body.clone(), true, 64).await;
let mut stream = WindowedHttpStream::create(params(url, 16))
.await
.expect("create");
stream.seek_range(100, None).await.expect("seek");
assert_eq!(read_all(&mut stream).await, body[100..]);
}
#[tokio::test]
async fn plain_servers_stream_one_body_without_seek() {
let body: Vec<u8> = (0..100u32).map(|i| i as u8).collect();
let url = serve(body.clone(), false, 0).await;
let mut stream = WindowedHttpStream::create(params(url, 16))
.await
.expect("create");
assert_eq!(stream.content_length(), Some(100));
assert!(!stream.supports_seek());
assert_eq!(read_all(&mut stream).await, body);
}
#[tokio::test]
async fn rejections_surface_as_errors_without_the_url() {
// A ranged server with max_window 0 rejects everything.
let url = serve(vec![0; 10], true, 0).await;
let err = WindowedHttpStream::create(params(url, 16))
.await
.expect_err("403 fails creation");
let message = err.to_string();
assert!(message.contains("403"), "{message}");
assert!(!message.contains("127.0.0.1"), "no url: {message}");
}
#[test]
fn content_range_parses_totals_and_wildcards() {
assert_eq!(
parse_content_range("bytes 0-1023/7831134"),
Some((0, 1023, Some(7831134)))
);
assert_eq!(parse_content_range("bytes 5-9/*"), Some((5, 9, None)));
assert_eq!(parse_content_range("garbage"), None);
}
}

132
crabidy-server/src/capture.rs
View File

@ -27,9 +27,19 @@ use tracing::warn;
/// Connect timeout for download requests.
pub const DOWNLOAD_CONNECT_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30);
/// Bytes per download request window. Some CDNs (googlevideo) reject
/// plain and open-ended requests from unattested clients with 403 and
/// only serve bounded ranges around this size — the same windowing the
/// player uses (audio-player/src/windowed_http.rs). Servers that ignore
/// the `Range` header answer 200 with the whole body, which is handled
/// as a single window.
pub const DOWNLOAD_WINDOW: u64 = 1024 * 1024;
/// Total per-track deadline: URL fetch, request, and streaming the whole
/// body. A stalled transfer aborts the capture instead of hanging it.
pub const DOWNLOAD_TRACK_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(600);
/// Generous: tokenless YouTube URLs are throttled to ~32 KB/s, so a long
/// track legitimately takes many minutes.
pub const DOWNLOAD_TRACK_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(1800);
/// Size limits for one capture. The walk aborts with
/// [`CaptureError::TooLarge`] when a limit trips — a runaway provider tree
@ -209,21 +219,29 @@ struct TrackEntry {
///
/// One shared HTTP client with a connect timeout; each track is bounded by
/// [`DOWNLOAD_TRACK_TIMEOUT`] end to end and never retried (a capture is
/// re-runnable; resuming re-attempts what is missing). Bodies are streamed
/// to disk against the capture's remaining byte budget.
/// re-runnable; resuming re-attempts what is missing). Bodies are fetched
/// in bounded [`DOWNLOAD_WINDOW`] ranges and streamed to disk against the
/// capture's remaining byte budget.
#[derive(Debug)]
pub struct Downloader {
http: reqwest::Client,
window: u64,
}
impl Downloader {
/// Builds the shared HTTP client. Fails only when the TLS backend
/// cannot initialize.
pub fn new() -> Result<Self, reqwest::Error> {
Self::with_window(DOWNLOAD_WINDOW)
}
/// [`Self::new`] with an explicit window size — the seam the
/// window-chaining tests use.
pub(crate) fn with_window(window: u64) -> Result<Self, reqwest::Error> {
let http = reqwest::Client::builder()
.connect_timeout(DOWNLOAD_CONNECT_TIMEOUT)
.build()?;
Ok(Self { http })
Ok(Self { http, window })
}
/// Settles one track: audio file first, then the toml pointing at
@ -312,14 +330,30 @@ impl Downloader {
let download_err = |err: reqwest::Error| {
CaptureError::Download(format!("{track_path}: {}", err.without_url()))
};
// First bounded window; its status decides the mode. Some CDNs
// (googlevideo) 403 plain and open-ended requests, so every
// request carries a bounded range; servers that ignore the header
// answer 200 with the whole body.
let mut start = 0u64;
let mut response = self
.http
.get(url)
.header(
reqwest::header::RANGE,
format!("bytes=0-{}", self.window - 1),
)
.send()
.await
.map_err(download_err)?
.error_for_status()
.map_err(download_err)?;
let windowed = match response.status() {
reqwest::StatusCode::PARTIAL_CONTENT => true,
reqwest::StatusCode::OK => false,
status => {
return Err(CaptureError::Download(format!(
"{track_path}: HTTP status {status}"
)))
}
};
let content_type = response
.headers()
.get(reqwest::header::CONTENT_TYPE)
@ -328,13 +362,65 @@ impl Downloader {
let ext = extension_for(content_type.as_deref(), url);
let audio_name = audio_file_name(index, &track.title, &ext);
let mut audio = tokio::fs::File::create(dir.join(&audio_name)).await?;
while let Some(chunk) = response.chunk().await.map_err(download_err)? {
let len = chunk.len() as u64;
if len > *bytes_left {
return Err(CaptureError::TooLarge("download budget exhausted"));
loop {
// The window's extent and the resource total, from
// `Content-Range: bytes <a>-<b>/<total>` (Content-Length
// fallback for the extent).
let (window_end, total) = if windowed {
let content_range = response
.headers()
.get(reqwest::header::CONTENT_RANGE)
.and_then(|value| value.to_str().ok())
.and_then(parse_content_range);
match content_range {
Some((_, range_end, total)) => (range_end + 1, total),
None => (start + response.content_length().unwrap_or(0), None),
}
} else {
(u64::MAX, None)
};
let mut received = 0u64;
while let Some(chunk) = response.chunk().await.map_err(download_err)? {
let len = chunk.len() as u64;
if len > *bytes_left {
return Err(CaptureError::TooLarge("download budget exhausted"));
}
*bytes_left -= len;
received += len;
tokio::io::AsyncWriteExt::write_all(&mut audio, &chunk).await?;
}
if !windowed {
break;
}
start = window_end.max(start + received);
match total {
Some(total) if start >= total => break,
// A short or empty window with no known total: the
// resource ended early.
_ if received == 0 => break,
None if received < self.window => break,
_ => {}
}
response = self
.http
.get(url)
.header(
reqwest::header::RANGE,
format!("bytes={start}-{}", start + self.window - 1),
)
.send()
.await
.map_err(download_err)?;
match response.status() {
reqwest::StatusCode::PARTIAL_CONTENT => {}
// Past the end: everything is on disk.
reqwest::StatusCode::RANGE_NOT_SATISFIABLE => break,
status => {
return Err(CaptureError::Download(format!(
"{track_path}: HTTP status {status}"
)))
}
}
*bytes_left -= len;
tokio::io::AsyncWriteExt::write_all(&mut audio, &chunk).await?;
}
tokio::io::AsyncWriteExt::flush(&mut audio).await?;
drop(audio);
@ -346,6 +432,18 @@ impl Downloader {
}
}
/// Parses `bytes <start>-<end>/<total|*>` into `(start, end, total)`.
fn parse_content_range(value: &str) -> Option<(u64, u64, Option<u64>)> {
let rest = value.trim().strip_prefix("bytes ")?;
let (range, total) = rest.split_once('/')?;
let (start, end) = range.split_once('-')?;
let total = match total.trim() {
"*" => None,
n => Some(n.parse().ok()?),
};
Some((start.trim().parse().ok()?, end.trim().parse().ok()?, total))
}
/// Writes the skipped toml for `track` at listing position `index`,
/// overwriting whatever was there.
async fn write_skipped(
@ -631,6 +729,16 @@ mod tests {
assert_eq!(extension_for(None, "not a url"), "bin");
}
#[test]
fn content_range_parses_totals_and_wildcards() {
assert_eq!(
parse_content_range("bytes 0-1023/7831134"),
Some((0, 1023, Some(7831134)))
);
assert_eq!(parse_content_range("bytes 5-9/*"), Some((5, 9, None)));
assert_eq!(parse_content_range("garbage"), None);
}
#[test]
fn audio_files_pair_with_their_toml_names() {
// Same prefix and sanitized stem as `fsdy::track_file_name`, so the

93
crabidy-server/src/capture_store.rs
View File

@ -209,6 +209,74 @@ mod tests {
format!("http://{addr}/stream")
}
/// Like [`serve`], but the server enforces bounded ranges the way
/// googlevideo does: ranged requests up to `max_window` bytes get
/// `206` + `Content-Range` slices, anything else (plain, open-ended,
/// oversized) gets `403`.
async fn serve_ranged(content_type: &'static str, body: Vec<u8>, max_window: u64) -> String {
let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
.await
.expect("bind test server");
let addr = listener.local_addr().expect("test server addr");
tokio::spawn(async move {
while let Ok((mut sock, _)) = listener.accept().await {
let body = body.clone();
tokio::spawn(async move {
let mut buf = Vec::new();
let mut chunk = [0u8; 1024];
loop {
match sock.read(&mut chunk).await {
Ok(0) | Err(_) => break,
Ok(n) => {
buf.extend_from_slice(&chunk[..n]);
if buf.windows(4).any(|w| w == b"\r\n\r\n") {
break;
}
}
}
}
let request = String::from_utf8_lossy(&buf).to_lowercase();
let range = request
.lines()
.find_map(|line| line.strip_prefix("range: bytes="))
.and_then(|spec| {
let (start, end) = spec.trim().split_once('-')?;
let start: u64 = start.parse().ok()?;
let end: u64 = end.parse().ok()?;
Some((start, end))
});
let total = body.len() as u64;
let response = match range {
Some((start, end)) if start < total && end - start < max_window => {
let end = end.min(total - 1);
let slice = &body[start as usize..=end as usize];
let mut head = format!(
"HTTP/1.1 206 Partial Content\r\nContent-Type: {content_type}\r\n\
Content-Range: bytes {start}-{end}/{total}\r\n\
Content-Length: {}\r\nConnection: close\r\n\r\n",
slice.len()
)
.into_bytes();
head.extend_from_slice(slice);
head
}
Some((start, _)) if start >= total => format!(
"HTTP/1.1 416 Range Not Satisfiable\r\n\
Content-Range: bytes */{total}\r\nContent-Length: 0\r\n\
Connection: close\r\n\r\n"
)
.into_bytes(),
_ => b"HTTP/1.1 403 Forbidden\r\nContent-Length: 0\r\nConnection: close\r\n\r\n"
.to_vec(),
};
let _ = sock.write_all(&response).await;
let _ = sock.shutdown().await;
});
}
});
format!("http://{addr}/stream")
}
/// A provider with one album (`/mock/a`: tracks `one`, `two`) whose
/// stream URLs are the test server's, and a node-level download
/// blessing toggle.
@ -626,6 +694,31 @@ mod tests {
);
}
#[tokio::test]
async fn downloads_chain_bounded_windows_on_strict_cdns() {
// A googlevideo-style server: only bounded ranges under 16 bytes
// are served; plain or oversized requests are 403. A downloader
// with a 10-byte window must fetch the 100-byte body completely.
let body: Vec<u8> = (0..100u32).map(|i| i as u8).collect();
let url = serve_ranged("audio/flac", body.clone(), 16).await;
let mock = MockProvider::new(&[("/mock/a/1", &url)], true);
let dir = TempDir::new().expect("store tempdir");
let store_dir = dir.path().join("captures");
tokio::fs::create_dir_all(&store_dir).await.expect("mkdir");
let store = CaptureStore {
dir: store_dir,
sink: crate::capture::Sink::Download(
crate::capture::Downloader::with_window(10).expect("downloader"),
),
};
store
.capture(&mock, "/mock/a/1", "windowed", &silent())
.await
.expect("windowed capture");
let audio = fs::read(store.dir().join("windowed/0001 one.flac")).expect("audio");
assert_eq!(audio, body, "all windows stitched in order");
}
#[tokio::test]
async fn capture_reports_progress_totals_and_skips() {
let url = serve("200 OK", "audio/flac", b"x".to_vec()).await;

4
devenv.nix
View File

@ -14,6 +14,10 @@ let
d2
pkg-config
protobuf
# Stream-URL sidecar for the ytdy provider: YouTube caps tokenless
# stream URLs at ~1 MiB and yt-dlp is the only maintained cipher
# solver (architecture/youtube-rustypipe.md, D2-revised).
yt-dlp
];
in
{

30
plan/summary.md
View File

@ -1,5 +1,35 @@
# Implementation summaries
## youtube stream fetching (2026-07-21, follow-up)
The rustypipe swap fixed the decode problem but real playback then hit
YouTube's tokenless-fetch enforcement, measured live: every stream URL
serves exactly its leading 1 MiB (403 beyond — plain, open-ended, and
oversized requests are rejected outright, and fresh URLs refuse offset
starts, killing URL-per-window chaining). PO tokens would lift the cap,
but rustypipe only attaches them to web clients whose signature
deciphering is currently broken upstream (verified on git master;
`rustypipe-botguard` built and tested — ineffective through the iOS
client). `yt-dlp` still solves the ciphers; its URLs stream the whole
file at a throttled ~32 KB/s — double the audio bitrate.
Shipped: (1) **windowed HTTP fetching** everywhere — a
`WindowedHttpStream` `SourceStream` in audio-player (bounded ~1 MiB
ranges, 200-body fallback for range-ignoring servers, eager
seek/reconnect so rejected windows fail typed instead of retrying
forever, URLs never in errors) and the same windowing in the capture
downloader (strict-CDN test stitches windows byte-exact); (2) `yt-dlp`
back as a **stream-URL-only sidecar** — all metadata stays on
rustypipe; a missing binary degrades to 1 MiB streams with a warning,
a failing call falls back to the rustypipe URL; (3) `botguard_bin`
config passthrough so streams flip back to pure Rust when upstream
deciphering recovers; (4) capture per-track deadline raised to 30 min
for the throttle. Live-verified end to end on the exact track from the
user's log: sidecar URL in 3 s, windowed stream + rodio decode
producing samples at 4.3 s. 179 workspace tests green (12 new);
`quality/youtube-rustypipe.md` gained a checked "Stream fetching"
section.
## youtube-rustypipe (2026-07-21)
Built per `plan/youtube-rustypipe.md`: the ytdy provider's `yt-dlp`

9
plan/youtube-rustypipe.md
View File

@ -27,3 +27,12 @@ Ordered tasks; each names its verification (ytdy tests and/or gates in
URL fetch, decode, download capture — removed after passing.
- [x] **T7 — Docs.** `plan/summary.md` section incl. deviations;
reconcile the architecture docs.
- [x] **T8 — Same-day fix: windowed fetching + stream sidecar.**
YouTube caps tokenless URLs at their leading ~1 MiB (measured; offset
starts refused; upstream web-client deciphering broken; botguard
ineffective through iOS). `WindowedHttpStream` in audio-player,
windowed capture downloads, `yt-dlp` back as a stream-URL-only
sidecar with degrade-and-fallback, `botguard_bin` passthrough,
capture deadline 30 min. Verifies: audio-player windowed tests,
capture window-chaining test, ytdy sidecar tests, gates "Stream
fetching"; live probe on the failing track.

20
quality/youtube-rustypipe.md
View File

@ -47,3 +47,23 @@ probe).
- [x] Live probe (removed after passing): search + stream URL +
download capture through the real provider; stream decodes with the
player's decoder stack.
## Stream fetching (D2-revised, same-day fix)
- [x] The player streams every http(s) source through the windowed
`SourceStream` (bounded ~1 MiB ranges); servers that ignore `Range`
degrade to one 200 body; rejected windows are typed errors (eager
seek/reconnect — no infinite retry loops), and error messages never
carry URLs.
- [x] The capture downloader uses the same bounded windows with a 200
whole-body fallback; existing capture tests pass unchanged and a
googlevideo-style strict server test stitches windows correctly.
- [x] `yt-dlp` is consulted **only** for stream URLs (argv-only,
bounded, stderr summarized ≤200 chars); a missing binary degrades
with a warning, never a failed init; a failing call falls back to the
rustypipe URL.
- [x] `botguard_bin` is configurable and forwarded to rustypipe (the
future pure-Rust exit).
- [x] Live probe (removed after passing): the user's exact failing
track resolved via the sidecar and decoded through the real
player path (windowed StreamDownload → rodio) in ~4 s.

1
ytdy/Cargo.toml
View File

@ -10,6 +10,7 @@ dirs.workspace = true
rustypipe.workspace = true
serde.workspace = true
thiserror.workspace = true
tokio = { workspace = true, features = ["process", "time", "io-util"] }
toml.workspace = true
tracing.workspace = true

18
ytdy/src/extract.rs
View File

@ -125,12 +125,22 @@ impl RustyPipeExtractor {
/// Builds the client. `storage_dir` holds rustypipe's cache file
/// (client state and the rotated auth cookie — as secret as the
/// cookies file itself); `timeout` bounds every YouTube request.
pub fn new(storage_dir: PathBuf, timeout: Duration) -> Result<Self, ExtractError> {
/// `botguard_bin` optionally names a `rustypipe-botguard` binary for
/// PO-token attestation (rustypipe also auto-detects it on PATH).
pub fn new(
storage_dir: PathBuf,
timeout: Duration,
botguard_bin: Option<&Path>,
) -> Result<Self, ExtractError> {
std::fs::create_dir_all(&storage_dir)
.map_err(|err| ExtractError::Client(format!("storage directory: {err}")))?;
let rp = RustyPipe::builder()
let mut builder = RustyPipe::builder()
.storage_dir(storage_dir)
.timeout(timeout)
.timeout(timeout);
if let Some(bin) = botguard_bin {
builder = builder.botguard_bin(bin.as_os_str().to_owned());
}
let rp = builder
.build()
.map_err(|err| ExtractError::Client(err.to_string()))?;
Ok(Self { rp })
@ -297,7 +307,7 @@ mod tests {
// missing file short-circuit before any request.
let dir = tempfile::TempDir::new().expect("tempdir");
let extractor =
RustyPipeExtractor::new(dir.path().join("rustypipe"), Duration::from_secs(5))
RustyPipeExtractor::new(dir.path().join("rustypipe"), Duration::from_secs(5), None)
.expect("client builds");
assert!(!extractor.login(None).await);
assert!(!extractor.login(Some(&dir.path().join("gone.txt"))).await);

149
ytdy/src/lib.rs
View File

@ -1,6 +1,10 @@
//! YouTube media provider, backed by the pure-Rust `rustypipe`
//! Innertube client (see `architecture/youtube-provider.md` for the
//! tree shape and `architecture/youtube-rustypipe.md` for the engine).
//! YouTube media provider: metadata (search, playlists, video details)
//! comes from the pure-Rust `rustypipe` Innertube client; **stream
//! URLs** go through a minimal `yt-dlp` sidecar when available, because
//! YouTube currently caps tokenless URLs at their leading ~1 MiB (see
//! `architecture/youtube-provider.md` for the tree shape and
//! `architecture/youtube-rustypipe.md` for the engine and its
//! constraints).
//!
//! Mounted at [`PROVIDER_ROOT`]. Search works without login: creatable
//! search-term nodes exactly like `/tidal/search` (in-memory terms,
@ -19,7 +23,9 @@ use serde::{Deserialize, Serialize};
use tracing::{debug, warn};
pub mod extract;
pub mod ytdlp;
use extract::{Extract, ExtractError, RustyPipeExtractor, VideoEntry};
use ytdlp::YtDlp;
/// First path segment owned by this provider.
pub const PROVIDER_ROOT: &str = "/youtube";
@ -49,6 +55,19 @@ pub struct Settings {
/// Per-request timeout in seconds. Default
/// [`DEFAULT_CALL_TIMEOUT_SECS`].
pub call_timeout_secs: Option<u64>,
/// The `yt-dlp` binary used **only** to resolve stream URLs (a bare
/// name resolves via PATH; default `yt-dlp`). YouTube currently
/// caps tokenless URLs at their leading ~1 MiB, and yt-dlp is the
/// only maintained cipher solver — without it, playback of a
/// YouTube track stops after roughly a minute (see
/// architecture/youtube-rustypipe.md).
pub binary: Option<PathBuf>,
/// Optional `rustypipe-botguard` binary for PO-token attestation
/// (also auto-detected on PATH). Currently ineffective — the
/// token-capable Innertube clients need signature deciphering that
/// is broken in rustypipe upstream — but wired so streams flip back
/// to pure Rust when upstream recovers.
pub botguard_bin: Option<PathBuf>,
}
/// A parsed `/youtube/...` path.
@ -118,6 +137,9 @@ fn entry_to_track(entry: &VideoEntry, node_path: &str) -> Track {
#[derive(Debug)]
pub struct Client {
extractor: Box<dyn Extract>,
/// The stream-URL sidecar; `None` degrades to rustypipe URLs (which
/// YouTube currently caps at their leading ~1 MiB).
ytdlp: Option<YtDlp>,
settings: Settings,
/// The login attempt at init succeeded — gates the playlists
/// subtree.
@ -133,6 +155,7 @@ impl Client {
fn with_extractor(extractor: Box<dyn Extract>, settings: Settings, logged_in: bool) -> Self {
Self {
extractor,
ytdlp: None,
settings,
logged_in,
search_terms: std::sync::RwLock::new(Vec::new()),
@ -263,10 +286,11 @@ impl Client {
#[async_trait]
impl ProviderClient for Client {
/// Builds the rustypipe client (state under
/// `<config>/crabidy/rustypipe/`) and attempts the cookie login. A
/// client that cannot be built fails init (the orchestrator disables
/// the provider non-fatally); a failed login degrades to logged-out
/// with a warning, never an error.
/// `<config>/crabidy/rustypipe/`), attempts the cookie login, and
/// probes the `yt-dlp` stream-URL sidecar. A rustypipe client that
/// cannot be built fails init (the orchestrator disables the
/// provider non-fatally); a failed login or a missing sidecar
/// degrades with a warning, never an error.
async fn init(raw_toml_settings: &str) -> Result<Self, ProviderError> {
let settings: Settings = toml::from_str(raw_toml_settings).unwrap_or_else(|_| {
warn!("could not parse toml settings, using defaults");
@ -280,17 +304,36 @@ impl ProviderClient for Client {
let storage_dir = dirs::config_dir()
.map(|dir| dir.join("crabidy").join("rustypipe"))
.ok_or_else(|| ProviderError::Config("no config directory".to_string()))?;
let extractor = RustyPipeExtractor::new(storage_dir, timeout).map_err(|err| {
warn!("cannot build the youtube client: {err}");
ProviderError::Config(err.to_string())
})?;
let extractor =
RustyPipeExtractor::new(storage_dir, timeout, settings.botguard_bin.as_deref())
.map_err(|err| {
warn!("cannot build the youtube client: {err}");
ProviderError::Config(err.to_string())
})?;
let logged_in = extractor.login(settings.cookies.as_deref()).await;
let binary = settings
.binary
.clone()
.unwrap_or_else(|| PathBuf::from("yt-dlp"));
let ytdlp = YtDlp::new(binary, timeout.max(Duration::from_secs(60)));
let ytdlp = match ytdlp.probe().await {
Ok(version) => {
debug!(version, "yt-dlp stream sidecar ready");
Some(ytdlp)
}
Err(err) => {
warn!(
"yt-dlp not usable ({err}); youtube streams will stop after \
their first ~1 MiB (see architecture/youtube-rustypipe.md)"
);
None
}
};
debug!(logged_in, "youtube extractor ready");
Ok(Self::with_extractor(
Box::new(extractor),
settings,
logged_in,
))
Ok(Self {
ytdlp,
..Self::with_extractor(Box::new(extractor), settings, logged_in)
})
}
fn settings(&self) -> String {
@ -307,11 +350,32 @@ impl ProviderClient for Client {
/// The video's audio stream URL — `audio/mp4` (AAC) preferred, the
/// format the local player decodes
/// (architecture/youtube-rustypipe.md D2).
///
/// Resolved through the `yt-dlp` sidecar when available (its cipher
/// solving yields fully streamable URLs); rustypipe is the fallback,
/// whose tokenless URLs YouTube currently caps at ~1 MiB.
async fn get_urls_for_track(&self, track_path: &str) -> Result<Vec<String>, ProviderError> {
let video = match parse_path(track_path)? {
YtPath::SearchTrack { video, .. } | YtPath::PlaylistTrack { video, .. } => video,
_ => return Err(ProviderError::MalformedPath),
};
if let Some(ytdlp) = &self.ytdlp {
let watch_url = format!("https://www.youtube.com/watch?v={video}");
match ytdlp.stream_url(&watch_url).await {
Ok(url) => return Ok(vec![url]),
Err(err) => {
warn!(
video,
"yt-dlp stream resolution failed, trying rustypipe: {err}"
);
}
}
} else {
warn!(
video,
"no yt-dlp sidecar; this stream will stop after its first ~1 MiB"
);
}
let url = self
.extractor
.audio_stream_url(video)
@ -769,11 +833,56 @@ mod tests {
}
#[test]
fn settings_tolerate_the_retired_binary_key() {
// Old ytdy.toml files carry `binary = "yt-dlp"`; parsing must
// ignore it instead of resetting the whole config.
fn settings_round_trip_old_and_new_keys() {
// Pre-rustypipe ytdy.toml files carry `binary`; it kept its
// meaning (now stream-URL-only). New keys parse alongside.
let settings: Settings =
toml::from_str("binary = \"yt-dlp\"\nsearch_results = 7\n").expect("parses");
toml::from_str("binary = \"yt-dlp\"\nsearch_results = 7\nbotguard_bin = \"/opt/bg\"\n")
.expect("parses");
assert_eq!(settings.search_results, Some(7));
assert_eq!(settings.binary, Some(PathBuf::from("yt-dlp")));
assert_eq!(settings.botguard_bin, Some(PathBuf::from("/opt/bg")));
}
#[tokio::test]
async fn stream_urls_prefer_the_ytdlp_sidecar() {
use std::os::unix::fs::PermissionsExt;
let dir = tempfile::TempDir::new().expect("tempdir");
let script = dir.path().join("fake-yt-dlp");
std::fs::write(
&script,
"#!/bin/sh\ncase \"$*\" in\n*watch?v=vid1*) echo \"https://sidecar.test/a.m4a\"; exit 0 ;;\n*) exit 1 ;;\nesac\n",
)
.expect("write");
let mut perms = std::fs::metadata(&script).expect("meta").permissions();
perms.set_mode(0o755);
std::fs::set_permissions(&script, perms).expect("chmod");
let mut client = client();
client.ytdlp = Some(YtDlp::new(script, Duration::from_secs(30)));
// The sidecar wins over the fake extractor's URL. (Retried: under
// parallel test load, exec of a just-written script can race a
// concurrent fork and fall back spuriously.)
let mut urls = Vec::new();
for _ in 0..5 {
urls = client
.get_urls_for_track("/youtube/search/lofi/vid1")
.await
.expect("stream url");
if urls == vec!["https://sidecar.test/a.m4a".to_string()] {
break;
}
}
assert_eq!(urls, vec!["https://sidecar.test/a.m4a".to_string()]);
// ...and a failing sidecar falls back to the extractor's answer.
client.ytdlp = Some(YtDlp::new(
dir.path().join("missing-binary"),
Duration::from_secs(5),
));
let urls = client
.get_urls_for_track("/youtube/search/lofi/vid1")
.await
.expect("fallback");
assert_eq!(urls, vec!["https://example.test/a.m4a".to_string()]);
}
}

201
ytdy/src/ytdlp.rs Normal file
View File

@ -0,0 +1,201 @@
//! Minimal `yt-dlp` sidecar for **stream URLs only**
//! (see `architecture/youtube-rustypipe.md` D2-revised).
//!
//! All metadata comes from the pure-Rust rustypipe extractor; this
//! module exists because YouTube currently rejects sustained tokenless
//! stream fetching (every URL serves exactly its leading ~1 MiB), and
//! the token-capable Innertube clients need signature deciphering that
//! is broken in rustypipe upstream right now. `yt-dlp` solves the
//! cipher challenges and its URLs stream fully (throttled, but well
//! above audio bitrate). When upstream recovers, this sidecar can be
//! dropped without touching the provider: it is only consulted by
//! `get_urls_for_track`.
//!
//! Subprocess discipline: argv-only (never a shell), one bounded call
//! (`kill_on_drop`), stdout capped, stderr summarized to a short
//! suffix. Errors carry the video id and exit status — never stream
//! URLs (they may embed tokens).
use std::path::PathBuf;
use std::process::Stdio;
use std::time::Duration;
use tracing::debug;
/// Cap on captured stdout; `-g` prints a handful of URLs.
const MAX_STDOUT_BYTES: usize = 512 * 1024;
/// Errors from the sidecar.
#[derive(Debug, thiserror::Error)]
pub enum YtDlpError {
#[error("cannot run yt-dlp: {0}")]
Spawn(String),
#[error("yt-dlp timed out after {0:?}")]
Timeout(Duration),
#[error("yt-dlp failed ({status}): {stderr}")]
Failed { status: String, stderr: String },
#[error("yt-dlp printed no stream url")]
NoUrl,
}
/// A configured `yt-dlp` binary used for stream URL resolution.
#[derive(Clone, Debug)]
pub struct YtDlp {
binary: PathBuf,
timeout: Duration,
}
impl YtDlp {
pub fn new(binary: PathBuf, timeout: Duration) -> Self {
Self { binary, timeout }
}
/// Probes `--version`; used at init to decide whether the sidecar
/// is available (absence degrades to rustypipe URLs, never fails
/// the provider).
pub async fn probe(&self) -> Result<String, YtDlpError> {
let stdout = self.run(&["--version"]).await?;
Ok(stdout.trim().to_string())
}
/// The audio stream URL for a watch URL: `audio/mp4` (AAC)
/// preferred — the format the local player decodes.
pub async fn stream_url(&self, watch_url: &str) -> Result<String, YtDlpError> {
let stdout = self
.run(&[
"-f",
"bestaudio[ext=m4a]/bestaudio/best",
"-g",
"--no-playlist",
watch_url,
])
.await?;
stdout
.lines()
.map(str::trim)
.find(|line| !line.is_empty())
.map(str::to_string)
.ok_or(YtDlpError::NoUrl)
}
/// Runs one bounded call and returns its stdout.
async fn run(&self, args: &[&str]) -> Result<String, YtDlpError> {
debug!(binary = %self.binary.display(), ?args, "running yt-dlp");
let mut command = tokio::process::Command::new(&self.binary);
command
.args(["--no-warnings"])
.args(args)
.stdin(Stdio::null())
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.kill_on_drop(true);
let child = command
.spawn()
.map_err(|err| YtDlpError::Spawn(err.to_string()))?;
let output = tokio::time::timeout(self.timeout, child.wait_with_output())
.await
.map_err(|_| YtDlpError::Timeout(self.timeout))?
.map_err(|err| YtDlpError::Spawn(err.to_string()))?;
if !output.status.success() {
let stderr = String::from_utf8_lossy(&output.stderr);
let summary: String = stderr.trim().chars().take(200).collect();
return Err(YtDlpError::Failed {
status: output.status.to_string(),
stderr: summary,
});
}
let mut stdout = output.stdout;
stdout.truncate(MAX_STDOUT_BYTES);
Ok(String::from_utf8_lossy(&stdout).into_owned())
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::fs;
use std::os::unix::fs::PermissionsExt;
use std::path::Path;
/// A fake yt-dlp: a script dispatching on its argv.
fn fake_binary(dir: &Path, body: &str) -> PathBuf {
let path = dir.join("fake-yt-dlp");
fs::write(&path, format!("#!/bin/sh\n{body}\n")).expect("write fake binary");
let mut perms = fs::metadata(&path).expect("metadata").permissions();
perms.set_mode(0o755);
fs::set_permissions(&path, perms).expect("chmod");
path
}
/// Retries a call while exec races a concurrent fork (ETXTBSY under
/// parallel test load).
async fn retry_spawn<T, F, Fut>(mut call: F) -> Result<T, YtDlpError>
where
F: FnMut() -> Fut,
Fut: std::future::Future<Output = Result<T, YtDlpError>>,
{
let mut last = call().await;
for _ in 0..4 {
if !matches!(last, Err(YtDlpError::Spawn(_))) {
break;
}
last = call().await;
}
last
}
#[tokio::test]
async fn resolves_stream_urls_and_probes() {
let dir = tempfile::TempDir::new().expect("tempdir");
let binary = fake_binary(
dir.path(),
r#"
case "$*" in
*--version*) echo "2026.07.04"; exit 0 ;;
*watch?v=vid1*) printf '%s\n' "https://example.test/a.m4a"; exit 0 ;;
*) echo boom >&2; exit 1 ;;
esac"#,
);
let ytdlp = YtDlp::new(binary, Duration::from_secs(30));
assert_eq!(
retry_spawn(|| ytdlp.probe()).await.expect("probe"),
"2026.07.04"
);
let url = retry_spawn(|| ytdlp.stream_url("https://www.youtube.com/watch?v=vid1"))
.await
.expect("url");
assert_eq!(url, "https://example.test/a.m4a");
// Failures are typed and carry a bounded stderr summary.
let err = ytdlp
.stream_url("https://www.youtube.com/watch?v=nope")
.await
.expect_err("fails");
assert!(err.to_string().contains("boom"), "{err}");
// A missing binary is a typed spawn error.
let gone = YtDlp::new(dir.path().join("gone"), Duration::from_secs(5));
assert!(matches!(gone.probe().await, Err(YtDlpError::Spawn(_))));
}
#[tokio::test]
async fn hung_calls_hit_the_timeout() {
let dir = tempfile::TempDir::new().expect("tempdir");
let binary = fake_binary(dir.path(), "sleep 30");
let ytdlp = YtDlp::new(binary, Duration::from_millis(200));
let started = std::time::Instant::now();
let mut saw_timeout = false;
for _ in 0..5 {
match ytdlp.probe().await {
Err(YtDlpError::Timeout(_)) => {
saw_timeout = true;
break;
}
// Under parallel test load, exec of a just-written
// script can race a concurrent fork (ETXTBSY) — retry.
Err(YtDlpError::Spawn(_)) => continue,
other => panic!("expected a timeout, got {other:?}"),
}
}
assert!(saw_timeout, "no timeout within 5 attempts");
assert!(started.elapsed() < Duration::from_secs(10));
}
}