feat(moq-transcode): decode once per source, GPU resize fanout, and a moq transcode verb

Cargo.toml

@@ -83,11 +83,13 @@ moq-rtc = { version = "0.1.2", path = "rs/moq-rtc" }
 moq-rtmp = { version = "0.1.2", path = "rs/moq-rtmp" }
 moq-srt = { version = "0.1.1", path = "rs/moq-srt" }
 moq-token = { version = "0.6", path = "rs/moq-token" }
+# default-features off (the hardware codecs) on moq-transcode and moq-video so
+# each consumer opts in: binaries (libmoq, moq-boy, moq-cli) enable them, but a
+# self-compiler can leave them off to drop the CUDA / libva deps. Both crates
+# still default them on when depended on directly.
+moq-transcode = { version = "0.0.1", path = "rs/moq-transcode", default-features = false }
 # Standalone crate (moq-dev/vaapi); vendored from cros-libva + cros-codecs.
 moq-vaapi = "0.0.2"
-# default-features off (the `nvenc` / `vaapi` hardware encoders) so each consumer
-# opts in: binaries (libmoq, moq-boy, moq-cli) enable them, but a self-compiler can
-# leave them off to drop the CUDA / libva deps. moq-video itself still defaults them on.
 moq-video = { version = "0.0.6", path = "rs/moq-video", default-features = false }
 qmux = { version = "0.3", default-features = false }
 serde = { version = "1", features = ["derive"] }

doc/bin/cli.md

@@ -160,6 +160,35 @@ ffmpeg -f v4l2 -i /dev/video0 -f mpegts - | \
     moq --client-connect https://relay.example.com/anon --broadcast webcam.hang import ts
 ```
 
+### Transcode a Broadcast
+
+The `transcode` verb consumes a broadcast from the relay and publishes a
+just-in-time transcoded ladder next to it. The derivative catalog references the
+source renditions directly and adds the lower rungs, which are only decoded and
+encoded while someone actually watches (or fetches) them. It is gated behind the
+`transcode` feature:
+
+```bash
+cargo build --release -p moq-cli --features transcode
+
+# Publish `cam.hang/transcode.hang` with the default ladder (1080p..240p,
+# filtered to rungs strictly below the source):
+moq --client-connect https://relay.example.com/anon --broadcast cam.hang transcode
+
+# Pick the ladder (height:bitrate in bits per second) and pin the codecs:
+moq --client-connect https://relay.example.com/anon --broadcast cam.hang     transcode --rung 720:2500000 --rung 360:600000 --encoder nvenc --decoder nvdec
+
+# Publish the derivative somewhere else (the catalog then omits the relative
+# source references):
+moq --client-connect https://relay.example.com/anon --broadcast cam.hang transcode --output ladder.hang
+```
+
+On an NVIDIA GPU the pipeline is fully GPU-resident: one shared NVDEC session
+decodes the source for all active rungs, a CUDA kernel resizes each rung's copy,
+and NVENC encodes it in place, with no CPU copies. Without a GPU it falls back
+to openh264 and CPU scaling. Like `capture`, the hardware backends are on by
+default; drop the default features for a software-only build.
+
 ### Play a Broadcast
 
 Pull a broadcast back out and play it:

rs/CLAUDE.md

@@ -27,12 +27,12 @@ Layered roughly transport -> container/format -> media -> apps/bindings.
 - `moq-mux` (lib): the conversion layer. File/stream formats (`container/`: fmp4, flv, mkv, ts, loc) and codec parsers (`codec/`: h264, h265, av1, vp8/9, opus, aac, ...) <-> hang broadcasts. `Container` trait + generic `Producer<C>`/`Consumer<C>`. Dual catalog (`catalog::hang`, `catalog::msf`).
 - `moq-audio` (lib): native PCM <-> Opus (`unsafe-libopus`). `AudioProducer`/`AudioConsumer`, `Encoder`/`Decoder`, `AudioFormat`. Optional `capture` feature (cpal microphone), `resample`.
 - `moq-video` (lib): native video capture, H.264/H.265 encode, and decode; no ffmpeg. Hardware backends (VideoToolbox / Media Foundation / NVENC / VAAPI / NVDEC) with openh264 as the software H.264 fallback; NVDEC frames stay in CUDA memory and feed NVENC zero-copy. `capture::Config`, `encode::{Encoder, Producer, publish_capture}`, `decode::{Consumer, Decoder}`.
-- `moq-transcode` (lib): just-in-time live transcoding of hang broadcasts. `run(source, output, config)` publishes a derivative catalog (ladder rungs + relative refs to the source) and encodes each rung only while subscribed/fetched, via `moq-video`. Output groups mirror source group sequences 1:1.
+- `moq-transcode` (lib): just-in-time live transcoding of hang broadcasts. `run(source, output, config)` publishes a derivative catalog (ladder rungs + relative refs to the source) and encodes each rung only while subscribed/fetched, via `moq-video`. Live rungs share one decode per source (the `feed` module); output groups mirror source group sequences 1:1. Also a moq-cli verb (`moq ... transcode`, feature-gated).
 
 **Apps / binaries**
 
 - `moq-relay` (lib+bin): clusterable, media-agnostic relay. axum HTTP API, JWT auth, WebSocket fallback, clustering. Config/TOML merge pattern lives here (see below).
-- `moq-cli` (bin, `moq`): the unified media router (`moq <MoQ side> <import|export> <endpoint>`); stdin/stdout media piping. The CLI surface for the gateway library crates below lives here.
+- `moq-cli` (bin, `moq`): the unified media router (`moq <MoQ side> <import|export> <endpoint>`, plus the feature-gated `transcode` verb); stdin/stdout media piping. The CLI surface for the gateway library crates below lives here.
 - `moq-rtc` (lib): WebRTC (WHIP/WHEP) gateway. Bridges browser WebRTC ingest/playback to MoQ broadcasts (str0m ICE/DTLS, A/V sync, NACK). Embeddable axum routers / `Client`; the CLI surface lives in `moq-cli`.
 - `moq-rtmp` (lib): RTMP / enhanced-RTMP gateway (ingest + egress, `rml_rtmp`, FLV via `moq-mux`). RTMPS (rustls + tokio-rustls) is the optional `tls` feature.
 - `moq-srt` (lib): bidirectional SRT gateway (MPEG-TS via `srt-tokio` + `moq-mux`).

rs/moq-cli/Cargo.toml

@@ -31,13 +31,17 @@ websocket = ["moq-native/websocket"]
 # extra install. H.264 is vendored static (openh264), so no system ffmpeg/libav.
 # Enable with `cargo build -p moq-cli --features capture`.
 capture = ["dep:moq-video", "dep:moq-audio", "moq-audio/capture"]
+# Just-in-time transcoding (`moq ... transcode`). Off by default for the same
+# reason as `capture`: it pulls in moq-video, whose default-on `vaapi` feature
+# links libva on Linux. Enable with `cargo build -p moq-cli --features transcode`.
+transcode = ["dep:moq-transcode", "dep:moq-video"]
 # NVENC / VAAPI / NVDEC hardware codecs for `capture` (Linux), on by default. Each
 # just turns on the matching moq-video feature, and only bites when `capture` pulls
 # in moq-video. For a `capture` build with no CUDA / libva system deps, drop them, e.g.
 #   cargo build -p moq-cli --no-default-features --features "iroh noq websocket capture"
-nvenc = ["moq-video?/nvenc"]
-vaapi = ["moq-video?/vaapi"]
-nvdec = ["moq-video?/nvdec"]
+nvenc = ["moq-video?/nvenc", "moq-transcode?/nvenc"]
+vaapi = ["moq-video?/vaapi", "moq-transcode?/vaapi"]
+nvdec = ["moq-video?/nvdec", "moq-transcode?/nvdec"]
 
 [dependencies]
 anyhow = { version = "1", features = ["backtrace"] }
@@ -55,6 +59,7 @@ moq-native = { workspace = true, default-features = false, features = ["aws-lc-r
 moq-rtc = { workspace = true }
 moq-rtmp = { workspace = true }
 moq-srt = { workspace = true }
+moq-transcode = { workspace = true, optional = true }
 moq-video = { workspace = true, optional = true }
 rustls = { version = "0.23", features = ["aws-lc-rs"], default-features = false }
 tokio = { workspace = true, features = ["full"] }

rs/moq-cli/src/args.rs

@@ -72,6 +72,10 @@ pub enum Direction {
 	/// Route media OUT OF MoQ to one sink.
 	#[command(alias = "subscribe")]
 	Export(Export),
+	/// Re-encode `--broadcast` into a lower ladder, published next to it and
+	/// only encoded while watched (just-in-time).
+	#[cfg(feature = "transcode")]
+	Transcode(crate::transcode::Args),
 }
 
 // ------------------------------------------------------------------ import

rs/moq-cli/src/main.rs

@@ -12,6 +12,8 @@ mod rtc;
 mod rtmp;
 mod srt;
 mod subscribe;
+#[cfg(feature = "transcode")]
+mod transcode;
 mod web;
 
 use args::{Cli, Direction, Export, ExportSink, Import, ImportSource, MoqSide};
@@ -66,6 +68,8 @@ async fn main() -> anyhow::Result<()> {
 	match cli.direction {
 		Direction::Import(import) => run_import(cli.moq, import, net).await,
 		Direction::Export(export) => run_export(cli.moq, export, net).await,
+		#[cfg(feature = "transcode")]
+		Direction::Transcode(args) => transcode::run(cli.moq, args, net).await,
 	}
 }
 

rs/moq-cli/src/transcode.rs

@@ -0,0 +1,125 @@
+//! The `transcode` verb: consume a source broadcast and publish a just-in-time
+//! transcoded ladder next to it.
+//!
+//! The derivative appears at `<broadcast>/transcode.hang` (or `--output`): its
+//! catalog references the source renditions directly and adds the lower rungs,
+//! which are only decoded and encoded while someone watches (or fetches) them.
+//! On an NVIDIA GPU the whole pipeline is GPU-resident (NVDEC -> CUDA resize ->
+//! NVENC); otherwise it falls back to software codecs.
+
+use anyhow::Context;
+
+use crate::Net;
+use crate::args::MoqSide;
+use hang::moq_net;
+
+/// Ladder and codec options for the `transcode` verb.
+#[derive(clap::Args, Clone)]
+pub struct Args {
+	/// The derivative broadcast path. Defaults to `<broadcast>/transcode.hang`.
+	#[arg(long)]
+	pub output: Option<String>,
+
+	/// A ladder rung as `height:bitrate` (pixels : bits per second), repeatable,
+	/// e.g. `--rung 720:2500000 --rung 360:600000`. Rungs at or above the source
+	/// are dropped at runtime. Defaults to a 1080p..240p ladder.
+	#[arg(long = "rung", value_parser = parse_rung)]
+	pub rungs: Vec<moq_transcode::Rung>,
+
+	/// The video encoder: `auto` (hardware first), `hardware`, `software`, or a
+	/// backend name like `nvenc`.
+	#[arg(long, default_value = "auto")]
+	pub encoder: String,
+
+	/// The video decoder: `auto` (hardware first), `hardware`, `software`, or a
+	/// backend name like `nvdec`.
+	#[arg(long, default_value = "auto")]
+	pub decoder: String,
+}
+
+/// Parse a `height:bitrate` rung, e.g. `720:2500000`.
+fn parse_rung(arg: &str) -> Result<moq_transcode::Rung, String> {
+	let (height, bitrate) = arg
+		.split_once(':')
+		.ok_or_else(|| format!("expected height:bitrate, got `{arg}`"))?;
+	let height: u32 = height.parse().map_err(|e| format!("invalid height `{height}`: {e}"))?;
+	let bitrate: u64 = bitrate
+		.parse()
+		.map_err(|e| format!("invalid bitrate `{bitrate}`: {e}"))?;
+	Ok(moq_transcode::Rung::new(height, bitrate))
+}
+
+/// Run the transcoder: subscribe to the source through the relay, publish the
+/// derivative back through the same session, and serve rungs until either ends.
+pub async fn run(moq: MoqSide, args: Args, net: Net) -> anyhow::Result<()> {
+	let source_path = moq
+		.broadcast
+		.clone()
+		.filter(|name| !name.is_empty())
+		.context("`transcode` requires the source broadcast: pass --broadcast <name>")?;
+	let output_path = args
+		.output
+		.clone()
+		.unwrap_or_else(|| format!("{source_path}/transcode.hang"));
+
+	// Publish the derivative through one origin and consume the source through
+	// another, over a single auto-reconnecting session.
+	let url = moq
+		.client
+		.connect
+		.clone()
+		.context("`transcode` requires a relay: pass --client-connect <url>")?;
+	let publish = moq_net::Origin::random().produce();
+	let remote = moq_net::Origin::random().produce();
+	let mut session = net
+		.client(moq.client.clone())?
+		.with_publisher(&publish)
+		.with_subscriber(remote.clone())
+		.reconnect(url);
+
+	// Wait for the first session: the origin can't route a broadcast request
+	// until a connected session registers its handler.
+	while !matches!(session.status().await?, moq_native::Status::Connected) {}
+
+	// Request the source broadcast; the session subscribes upstream on demand.
+	let source = remote
+		.consume()
+		.request_broadcast(&source_path)
+		.await
+		.context("source broadcast unavailable")?;
+
+	let mut config = moq_transcode::Config::default();
+	if !args.rungs.is_empty() {
+		config.rungs = args.rungs.clone();
+	}
+	config.encoder = match args.encoder.as_str() {
+		"auto" => moq_video::encode::Kind::Auto,
+		"hardware" => moq_video::encode::Kind::Hardware,
+		"software" => moq_video::encode::Kind::Software,
+		name => moq_video::encode::Kind::Named(name.to_string()),
+	};
+	config.decoder = match args.decoder.as_str() {
+		"auto" => moq_video::decode::Kind::Auto,
+		"hardware" => moq_video::decode::Kind::Hardware,
+		"software" => moq_video::decode::Kind::Software,
+		name => moq_video::decode::Kind::Named(name.to_string()),
+	};
+	// Reference the source renditions relatively when the output nests under
+	// the source (`a/b` -> `a/b/transcode.hang` is `..`, one `..` per level);
+	// otherwise the derivative catalog advertises only the rungs.
+	config.source = output_path.strip_prefix(&format!("{source_path}/")).map(|rest| {
+		let depth = rest.split('/').count();
+		moq_net::PathRelativeOwned::from(vec![".."; depth].join("/"))
+	});
+
+	let output = moq_net::broadcast::Info::default().produce();
+	let _announce = publish
+		.publish_broadcast(&output_path, &output)
+		.context("failed to publish the derivative broadcast")?;
+	tracing::info!(source = %source_path, output = %output_path, "transcoding");
+
+	tokio::select! {
+		res = moq_transcode::run(source, output, config) => Ok(res?),
+		res = session.closed() => Ok(res?),
+	}
+}

rs/moq-transcode/CHANGELOG.md

@@ -9,6 +9,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- Shared live decode: all rungs of a source with live demand now share one
+  subscription and one decoder (a broadcast feed of decoded frames), instead of
+  each rung decoding the source independently. NVDEC throughput and upstream
+  bandwidth now scale with source count, not ladder depth; each rung resizes
+  its copy on the GPU (`decode::Frame::resize`) and encodes it in place. Group
+  fetches keep their own one-shot pipeline.
+- `moq transcode`: the transcoder is now a moq-cli verb (behind the `transcode`
+  feature), publishing `<broadcast>/transcode.hang` with a configurable ladder
+  (`--rung height:bitrate`) and codec pins (`--encoder`, `--decoder`).
+
 - Initial release: just-in-time live transcoding of hang broadcasts.
   `run(source, output, config)` publishes a derivative catalog (ladder rungs
   strictly below the source, plus relative references to the source renditions)

rs/moq-transcode/Cargo.toml

@@ -30,9 +30,6 @@ nvdec = ["moq-video/nvdec"]
 
 [dependencies]
 bytes = "1"
-# CPU I420 scaling between decode and re-encode (SIMD per-plane resize), for
-# decoders without a built-in scaler; NVDEC scales in hardware instead.
-fast_image_resize = "5"
 hang = { workspace = true }
 moq-mux = { workspace = true }
 moq-net = { workspace = true }
@@ -46,5 +43,6 @@ anyhow = "1"
 clap = { version = "4", features = ["derive"] }
 # Default features on (unlike the workspace entry) so the example gets a QUIC backend.
 moq-native = { path = "../moq-native" }
-tokio = { workspace = true, features = ["full"] }
+# test-util unlocks tokio::time::pause for the deterministic live-feed tests.
+tokio = { workspace = true, features = ["full", "test-util"] }
 url = "2"

rs/moq-transcode/README.md

@@ -9,9 +9,10 @@ combined ladder and the transcoder never proxies what it doesn't re-encode.
 
 Nothing is encoded until someone asks, Cloudflare-style just-in-time per rung:
 
-- **Subscribe** to a rung and the transcoder subscribes to the source track,
-  decoding, scaling, and re-encoding group for group until the last subscriber
-  leaves.
+- **Subscribe** to a rung and the transcoder decodes the source live, resizing
+  and re-encoding group for group until the last subscriber leaves. Every
+  active rung of a source shares one subscription and one decoder, so decode
+  cost scales with sources, not ladder depth.
 - **Fetch** a specific group and the transcoder fetches that same group from
   the source and transcodes just that group. Output groups mirror source group
   sequence numbers 1:1, so group N of every rung is the same content as source

rs/moq-transcode/examples/transcode.rs

@@ -43,11 +43,15 @@ async fn main() -> anyhow::Result<()> {
 	let remote = moq_net::Origin::random().produce();
 
 	let client = moq_native::ClientConfig::default().init()?;
-	let session = client
+	let mut session = client
 		.with_publisher(&publish)
 		.with_subscriber(remote.clone())
 		.reconnect(args.url.clone());
 
+	// Wait for the first session: the origin can't route a broadcast request
+	// until a connected session registers its handler.
+	while !matches!(session.status().await?, moq_native::Status::Connected) {}
+
 	// Request the source broadcast; the session subscribes upstream on demand.
 	let source = remote
 		.consume()