- commit
- d72ae8f3978963dccf24dc4e8ee8ef8a95f32739
- parent
- 4e16190dd5f1233173ac2fc7bb7f791a615b7722
- author
- FelineStateMachine <dami.etoile@gmail.com>
| M | Cargo.lock | 67 | ++++++- |
| M | Cargo.toml | 2 | + |
| M | README.md | 7 | +- |
| M | crates/ | 2 | + |
| A | crates/ | 294 | +++++++++++++++++++++++++++++ |
| M | crates/ | 33 | ++- |
| M | crates/ | 1 | + |
| A | crates/ | 106 | ++++++++++ |
| M | crates/ | 15 | +- |
| M | crates/ | 6 | + |
| A | crates/ | 139 | ++++++++++++++ |
| M | scripts/ | 3 | +- |
| M | scripts/ | 10 | + |
| M | src-tauri/ | 1 | + |
| M | src-tauri/ | 1 | + |
| A | src-tauri/ | 304 | ++++++++++++++++++++++++++++++ |
| M | src-tauri/ | 30 | ++- |
@@ -2482,6 +2482,25 @@ dependencies = [
24822482 "zerocopy",24832483 ]24842484 2485+[[package]]2486+name = "hang"2487+version = "0.19.1"2488+source = "registry+https://github.com/rust-lang/crates.io-index"2489+checksum = "32320866e0e43ac542badd472790d8267dcca0284a344bf9a55465278ff44f3d"2490+dependencies = [2491+ "bytes",2492+ "derive_more",2493+ "hex",2494+ "lazy_static",2495+ "moq-net",2496+ "regex",2497+ "serde",2498+ "serde_json",2499+ "serde_with",2500+ "thiserror 2.0.19",2501+ "url",2502+]2503+24852504 [[package]]24862505 name = "hashbrown"24872506 version = "0.12.3"@@ -3346,12 +3365,24 @@ version = "3.0.1"
33463365 source = "registry+https://github.com/rust-lang/crates.io-index"33473366 checksum = "863726d7afb6bc2590eeff7135d923545e5e964f004c2ccf8716c25e70a86f08"33483367 dependencies = [3349- "jsonptr",3368+ "jsonptr 0.6.3",33503369 "serde",33513370 "serde_json",33523371 "thiserror 1.0.69",33533372 ]33543373 3374+[[package]]3375+name = "json-patch"3376+version = "4.2.0"3377+source = "registry+https://github.com/rust-lang/crates.io-index"3378+checksum = "7421438de105a0827e44fadd05377727847d717c80ce29a229f85fd04c427b72"3379+dependencies = [3380+ "jsonptr 0.7.1",3381+ "serde",3382+ "serde_json",3383+ "thiserror 2.0.19",3384+]3385+33553386 [[package]]33563387 name = "jsonptr"33573388 version = "0.6.3"@@ -3362,6 +3393,16 @@ dependencies = [
33623393 "serde_json",33633394 ]33643395 3396+[[package]]3397+name = "jsonptr"3398+version = "0.7.1"3399+source = "registry+https://github.com/rust-lang/crates.io-index"3400+checksum = "a5a3cc660ba5d72bce0b3bb295bf20847ccbb40fd423f3f05b61273672e561fe"3401+dependencies = [3402+ "serde",3403+ "serde_json",3404+]3405+33653406 [[package]]33663407 name = "keyboard-types"33673408 version = "0.7.0"@@ -3601,6 +3642,21 @@ dependencies = [
36013642 "uuid",36023643 ]36033644 3645+[[package]]3646+name = "moq-json"3647+version = "0.0.3"3648+source = "registry+https://github.com/rust-lang/crates.io-index"3649+checksum = "1de30545123e2391b890c01f82650a868f8b4e87bc57579627c32d1a0ece51b1"3650+dependencies = [3651+ "bytes",3652+ "json-patch 4.2.0",3653+ "kio",3654+ "moq-net",3655+ "serde",3656+ "serde_json",3657+ "thiserror 2.0.19",3658+]3659+36043660 [[package]]36053661 name = "moq-net"36063662 version = "0.1.11"@@ -4717,6 +4773,7 @@ dependencies = [
47174773 "ctap-hid-fido2",47184774 "ed25519-dalek",47194775 "getrandom 0.4.3",4776+ "hang",47204777 "image",47214778 "iroh",47224779 "iroh-moq",@@ -5642,10 +5699,12 @@ dependencies = [
56425699 "gstreamer",56435700 "gstreamer-app",56445701 "gstreamer-video",5702+ "hang",56455703 "input-linux",56465704 "iroh",56475705 "iroh-moq",56485706 "libc",5707+ "moq-json",56495708 "moq-net",56505709 "serde",56515710 "serde_json",@@ -6140,7 +6199,7 @@ dependencies = [
61406199 "dirs",61416200 "glob",61426201 "heck 0.5.0",6143- "json-patch",6202+ "json-patch 3.0.1",61446203 "schemars 0.8.22",61456204 "semver",61466205 "serde",@@ -6159,7 +6218,7 @@ dependencies = [
61596218 "base64 0.22.1",61606219 "brotli",61616220 "ico",6162- "json-patch",6221+ "json-patch 3.0.1",61636222 "plist",61646223 "png 0.17.16",61656224 "proc-macro2",@@ -6374,7 +6433,7 @@ dependencies = [
63746433 "glob",63756434 "http",63766435 "infer",6377- "json-patch",6436+ "json-patch 3.0.1",63786437 "log",63796438 "memchr",63806439 "phf",@@ -6,6 +6,8 @@ resolver = "2"
66 bytes = "1.11"77 iroh = { version = "=1.0.3", default-features = false, features = ["metrics", "portmapper", "fast-apple-datapath", "tls-aws-lc-rs"] }88 iroh-moq = { version = "=0.1.0", git = "https://github.com/n0-computer/iroh-live", rev = "5f95758fcd1450e443a9134c9d9342bcc3957b85" }9+hang = "=0.19.1"10+moq-json = "=0.0.3"911 moq-net = "=0.1.11"1012 web-transport-iroh = "=0.6.0"1113 @@ -87,7 +87,12 @@ Steam and games.
8787 The software JPEG client decode fallback and legacy wire compatibility remain.8888 The preferred video path authenticates the client over a dedicated control8989 connection, then admits that exact peer to a session-scoped upstream90-`iroh-moq` broadcast and H.264 track. Configured GOPs become bounded native MoQ90+`iroh-moq` broadcast. Sigil publishes an immutable `catalog.json` snapshot;91+Portal discovers and strictly validates the versioned Goq extension before it92+subscribes to the H.264 track. The extension identifies Goq's `SGV1` media93+envelope and GOP layout, so it does not falsely advertise the track as a94+standard Hang rendition. Older hosts without a catalog retain an explicit95+static-track compatibility path. Configured GOPs become bounded native MoQ9196 groups, and a new independently decodable group cancels the stale predecessor.9297 The custom media v3, v2, and v1 transports remain explicit compatibility modes.9398 A dedicated persistent PipeWire sink, bounded Opus datagrams, WebCodecs decode,@@ -26,9 +26,11 @@ anyhow = "1"
2626 bytes.workspace = true2727 clap = { version = "4", features = ["derive"] }2828 getrandom = "0.4"29+hang.workspace = true2930 iroh.workspace = true3031 iroh-moq.workspace = true3132 libc = "0.2"33+moq-json.workspace = true3234 moq-net.workspace = true3335 serde = { version = "1", features = ["derive"] }3436 serde_json = "1"@@ -0,0 +1,294 @@
1+use std::time::Duration;2+3+use anyhow::{Context, Result};4+use moq_net::{BroadcastConsumer, Error as MoqError, Track, TrackConsumer};5+use serde::{Deserialize, Serialize};6+use sigil_protocol::{MAX_MOQ_CATALOG_BYTES, MoqCatalogExtensionV1};7+8+9+10+struct GoqCatalogDocument {11+ 12+ media: hang::Catalog,13+ goq: MoqCatalogExtensionV1,14+}15+16+impl GoqCatalogDocument {17+ 18+ fn video_h264() -> Self {19+ Self {20+ media: hang::Catalog::default(),21+ goq: MoqCatalogExtensionV1::video_h264(),22+ }23+ }24+25+ fn validate(&self) -> Result<()> {26+ anyhow::ensure!(27+ self.media == hang::Catalog::default(),28+ "Goq catalog must not advertise a standard Hang rendition for enveloped media"29+ );30+ self.goq31+ .validate()32+ .context("validating Goq catalog extension")33+ }34+}35+36+37+pub(crate) enum MoqCatalogMode {38+ GoqV1,39+ AbsentStaticTrackCompat,40+}41+42+impl MoqCatalogMode {43+ pub(crate) const fn label(self) -> &'static str {44+ match self {45+ Self::GoqV1 => "goq-v1",46+ Self::AbsentStaticTrackCompat => "absent-static-track-compat",47+ }48+ }49+}50+51+pub(crate) struct MoqCatalogSelection {52+ pub(crate) track: TrackConsumer,53+ pub(crate) mode: MoqCatalogMode,54+}55+56+fn is_track_not_found(error: &MoqError) -> bool {57+ matches!(error, MoqError::NotFound)58+ || matches!(error, MoqError::Remote(code) if *code == MoqError::NotFound.to_code())59+}60+61+fn subscribe_static_video_track(broadcast: &BroadcastConsumer) -> Result<MoqCatalogSelection> {62+ let track = broadcast63+ .subscribe_track(&Track::new(sigil_protocol::MOQ_VIDEO_H264_TRACK))64+ .context("subscribing to static MoQ H.264 compatibility track")?;65+ Ok(MoqCatalogSelection {66+ track,67+ mode: MoqCatalogMode::AbsentStaticTrackCompat,68+ })69+}70+71+pub(crate) async fn subscribe_goq_video_track(72+ broadcast: &BroadcastConsumer,73+ timeout: Duration,74+) -> Result<MoqCatalogSelection> {75+ let catalog_track = match broadcast.subscribe_track(&hang::Catalog::default_track()) {76+ Ok(track) => track,77+ Err(MoqError::NotFound) => {78+ return subscribe_static_video_track(broadcast);79+ }80+ Err(error) => return Err(error).context("subscribing to catalog.json track"),81+ };82+83+ let mut catalog_track = catalog_track;84+ let deadline = tokio::time::Instant::now() + timeout;85+ let mut group = match tokio::time::timeout_at(deadline, catalog_track.next_group()).await {86+ Err(_) => anyhow::bail!("timed out waiting for Goq catalog snapshot"),87+ Ok(Err(error)) if is_track_not_found(&error) => {88+ return subscribe_static_video_track(broadcast);89+ }90+ Ok(Err(error)) => return Err(error).context("reading Goq catalog snapshot"),91+ Ok(Ok(None)) => anyhow::bail!("Goq catalog track ended before its snapshot"),92+ Ok(Ok(Some(group))) => group,93+ };94+ anyhow::ensure!(95+ group.sequence == 0,96+ "Goq catalog snapshot must be immutable group 0"97+ );98+ let frame_count = tokio::time::timeout_at(deadline, group.finished())99+ .await100+ .context("timed out waiting for Goq catalog group completion")?101+ .context("finishing Goq catalog group")?;102+ anyhow::ensure!(103+ frame_count == 1,104+ "Goq catalog snapshot group must contain exactly one frame"105+ );106+ let mut frame = tokio::time::timeout_at(deadline, group.get_frame(0))107+ .await108+ .context("timed out waiting for Goq catalog frame")?109+ .context("opening Goq catalog frame")?110+ .context("Goq catalog snapshot group has no frame 0")?;111+ anyhow::ensure!(112+ frame.size <= MAX_MOQ_CATALOG_BYTES as u64,113+ "Goq catalog snapshot exceeds {MAX_MOQ_CATALOG_BYTES} bytes"114+ );115+ let snapshot = tokio::time::timeout_at(deadline, frame.read_all())116+ .await117+ .context("timed out reading Goq catalog frame")?118+ .context("reading Goq catalog frame")?;119+ let document: GoqCatalogDocument =120+ serde_json::from_slice(&snapshot).context("decoding Goq catalog snapshot")?;121+ document.validate()?;122+ let track = broadcast123+ .subscribe_track(&Track::new(document.goq.video.track.name))124+ .context("subscribing to catalog-selected Goq video track")?;125+ Ok(MoqCatalogSelection {126+ track,127+ mode: MoqCatalogMode::GoqV1,128+ })129+}130+131+132+mod tests {133+ use super::*;134+ use moq_net::Broadcast;135+136+ fn publish_catalog(137+ broadcast: &mut moq_net::BroadcastProducer,138+ document: &GoqCatalogDocument,139+ ) -> moq_net::TrackProducer {140+ let mut track = broadcast141+ .create_track(hang::Catalog::default_track())142+ .unwrap();143+ track144+ .write_frame(serde_json::to_vec(document).unwrap())145+ .unwrap();146+ track147+ }148+149+ 150+ async fn catalog_selects_the_exact_goq_video_track() {151+ let mut broadcast = Broadcast::new().produce();152+ let _video = broadcast153+ .create_track(Track::new(sigil_protocol::MOQ_VIDEO_H264_TRACK))154+ .unwrap();155+ let _catalog = publish_catalog(&mut broadcast, &GoqCatalogDocument::video_h264());156+ let selection = subscribe_goq_video_track(&broadcast.consume(), Duration::from_millis(100))157+ .await158+ .unwrap();159+ assert_eq!(selection.mode, MoqCatalogMode::GoqV1);160+ }161+162+ 163+ async fn only_an_absent_catalog_uses_the_legacy_static_track() {164+ let mut broadcast = Broadcast::new().produce();165+ let _video = broadcast166+ .create_track(Track::new(sigil_protocol::MOQ_VIDEO_H264_TRACK))167+ .unwrap();168+ let selection = subscribe_goq_video_track(&broadcast.consume(), Duration::from_millis(100))169+ .await170+ .unwrap();171+ assert_eq!(selection.mode, MoqCatalogMode::AbsentStaticTrackCompat);172+ }173+174+ 175+ async fn asynchronous_remote_not_found_uses_static_track_compatibility() {176+ let mut broadcast = Broadcast::new().produce();177+ let _video = broadcast178+ .create_track(Track::new(sigil_protocol::MOQ_VIDEO_H264_TRACK))179+ .unwrap();180+ let mut dynamic = broadcast.dynamic();181+ let consumer = broadcast.consume();182+183+ let select = subscribe_goq_video_track(&consumer, Duration::from_millis(100));184+ let reject = async {185+ let mut requested = dynamic.requested_track().await.unwrap();186+ assert_eq!(requested.name, hang::Catalog::default_track().name);187+ requested188+ .abort(MoqError::Remote(MoqError::NotFound.to_code()))189+ .unwrap();190+ };191+ let (selection, ()) = tokio::join!(select, reject);192+ assert_eq!(193+ selection.unwrap().mode,194+ MoqCatalogMode::AbsentStaticTrackCompat195+ );196+ }197+198+ 199+ async fn asynchronous_non_not_found_error_fails_closed() {200+ let mut broadcast = Broadcast::new().produce();201+ let _video = broadcast202+ .create_track(Track::new(sigil_protocol::MOQ_VIDEO_H264_TRACK))203+ .unwrap();204+ let mut dynamic = broadcast.dynamic();205+ let consumer = broadcast.consume();206+207+ let select = subscribe_goq_video_track(&consumer, Duration::from_millis(100));208+ let reject = async {209+ let mut requested = dynamic.requested_track().await.unwrap();210+ requested211+ .abort(MoqError::Remote(MoqError::WrongSize.to_code()))212+ .unwrap();213+ };214+ let (selection, ()) = tokio::join!(select, reject);215+ let error = selection.err().expect("non-NotFound must fail closed");216+ assert!(217+ format!("{error:#}").contains("remote error: code=14"),218+ "{error:#}"219+ );220+ }221+222+ 223+ async fn present_invalid_or_stalled_catalogs_fail_closed() {224+ let mut invalid_broadcast = Broadcast::new().produce();225+ let _video = invalid_broadcast226+ .create_track(Track::new(sigil_protocol::MOQ_VIDEO_H264_TRACK))227+ .unwrap();228+ let mut invalid = GoqCatalogDocument::video_h264();229+ invalid.goq.video.object_format = "hang/legacy".into();230+ let _catalog = publish_catalog(&mut invalid_broadcast, &invalid);231+ let result =232+ subscribe_goq_video_track(&invalid_broadcast.consume(), Duration::from_millis(100))233+ .await;234+ assert!(result.err().is_some_and(|error| {235+ error236+ .to_string()237+ .contains("validating Goq catalog extension")238+ }));239+240+ let mut stalled_broadcast = Broadcast::new().produce();241+ let _catalog = stalled_broadcast242+ .create_track(hang::Catalog::default_track())243+ .unwrap();244+ let result =245+ subscribe_goq_video_track(&stalled_broadcast.consume(), Duration::from_millis(10))246+ .await;247+ assert!(248+ result249+ .err()250+ .is_some_and(|error| error.to_string().contains("timed out"))251+ );252+253+ let mut oversized_broadcast = Broadcast::new().produce();254+ let _video = oversized_broadcast255+ .create_track(Track::new(sigil_protocol::MOQ_VIDEO_H264_TRACK))256+ .unwrap();257+ let mut catalog = oversized_broadcast258+ .create_track(hang::Catalog::default_track())259+ .unwrap();260+ catalog261+ .write_frame(vec![b' '; MAX_MOQ_CATALOG_BYTES + 1])262+ .unwrap();263+ let result =264+ subscribe_goq_video_track(&oversized_broadcast.consume(), Duration::from_millis(100))265+ .await;266+ assert!(267+ result268+ .err()269+ .is_some_and(|error| error.to_string().contains("exceeds 4096 bytes"))270+ );271+272+ let mut multi_frame_broadcast = Broadcast::new().produce();273+ let _video = multi_frame_broadcast274+ .create_track(Track::new(sigil_protocol::MOQ_VIDEO_H264_TRACK))275+ .unwrap();276+ let mut catalog = multi_frame_broadcast277+ .create_track(hang::Catalog::default_track())278+ .unwrap();279+ let mut group = catalog.append_group().unwrap();280+ group281+ .write_frame(serde_json::to_vec(&GoqCatalogDocument::video_h264()).unwrap())282+ .unwrap();283+ group.write_frame(b"{}".as_slice()).unwrap();284+ group.finish().unwrap();285+ let result =286+ subscribe_goq_video_track(&multi_frame_broadcast.consume(), Duration::from_millis(100))287+ .await;288+ assert!(289+ result290+ .err()291+ .is_some_and(|error| error.to_string().contains("exactly one frame"))292+ );293+ }294+}@@ -7,17 +7,21 @@ use anyhow::{Context, Result, bail, ensure};
77 use clap::Parser;88 use iroh::{Endpoint, EndpointAddr, EndpointId, SecretKey, endpoint::presets};99 use iroh_moq::{Moq, MoqSession};10-use moq_net::{BroadcastConsumer, GroupConsumer, Track, TrackConsumer};10+use moq_net::{BroadcastConsumer, GroupConsumer, TrackConsumer};1111 use sigil_protocol::{1212 CONTROL_ALPN_V1, Capability, ClientHello, FrameFlags, GAMEPAD_AXIS_MAX, GAMEPAD_AXIS_MIN,1313 GAMEPAD_TRIGGER_MAX, GamepadState, INPUT_ALPN_V1, InputEvent, KeyframeRequestReasonV3,1414 MAX_MEDIA_GROUP_BYTES_V3, MAX_MEDIA_OBJECT_ID_V3, MEDIA_ALPN_V1, MEDIA_ALPN_V2, MEDIA_ALPN_V3,15- MOQ_VIDEO_H264_TRACK, MediaCodec, MediaControlRequestV3, MediaFrame, MediaObjectV3,16- PointerSurfaceDimensions, ProtocolError, decode_media_frame_object, media_moq_broadcast_name,17- read_host_hello, read_input_ack, read_media_frame, read_media_object, read_media_object_v3,18- write_client_hello, write_input_event, write_media_control_request_v3,15+ MediaCodec, MediaControlRequestV3, MediaFrame, MediaObjectV3, PointerSurfaceDimensions,16+ ProtocolError, decode_media_frame_object, media_moq_broadcast_name, read_host_hello,17+ read_input_ack, read_media_frame, read_media_object, read_media_object_v3, write_client_hello,18+ write_input_event, write_media_control_request_v3,1919 };2020 21+mod moq_catalog;22+23+use moq_catalog::{MoqCatalogMode, subscribe_goq_video_track};24+2125 const MEDIA_OBJECT_CAPACITY: usize = 4;2226 2327 @@ -641,6 +645,7 @@ struct MoqProbeLifetime {
641645 struct MoqProbeReceiver {642646 lifetime: MoqProbeLifetime,643647 track: TrackConsumer,648+ catalog_mode: MoqCatalogMode,644649 current_group: Option<MoqProbeGroupCursor>,645650 last_group_sequence: Option<u64>,646651 last_frame_sequence: Option<u64>,@@ -679,16 +684,17 @@ impl MoqProbeReceiver {
679684 format!("timed out subscribing to upstream MoQ broadcast {broadcast_name}")680685 })?681686 .with_context(|| format!("subscribing to upstream MoQ broadcast {broadcast_name}"))?;682- let track = broadcast683- .subscribe_track(&Track::new(MOQ_VIDEO_H264_TRACK))684- .with_context(|| format!("subscribing to MoQ track {MOQ_VIDEO_H264_TRACK}"))?;687+ let catalog = subscribe_goq_video_track(&broadcast, timeout)688+ .await689+ .context("resolving Goq MoQ catalog")?;685690 Ok(Self {686691 lifetime: MoqProbeLifetime {687692 _moq: moq,688693 session,689694 _broadcast: broadcast,690695 },691- track,696+ track: catalog.track,697+ catalog_mode: catalog.mode,692698 current_group: None,693699 last_group_sequence: None,694700 last_frame_sequence: None,@@ -1508,6 +1514,14 @@ async fn main() -> Result<()> {
15081514 }15091515 );15101516 if media_transport == MediaTransport::UpstreamMoq {1517+ println!(1518+ "moq_catalog={}",1519+ moq_receiver1520+ .as_ref()1521+ .expect("upstream MoQ receiver is present")1522+ .catalog_mode1523+ .label()1524+ );15111525 println!("moq_group_capacity=1");15121526 println!("moq_cancelled_groups={moq_cancelled_groups}");15131527 println!("moq_group_gaps={moq_group_gaps}");@@ -1516,6 +1530,7 @@ async fn main() -> Result<()> {
15161530 println!("moq_maximum_group_bytes={moq_maximum_group_bytes}");15171531 println!("moq_historical_suffix_frames={moq_historical_suffix_frames}");15181532 } else {1533+ println!("moq_catalog=not-applicable");15191534 for field in [15201535 "moq_group_capacity",15211536 "moq_cancelled_groups",@@ -5,6 +5,7 @@ mod config;
55 mod cursor;66 mod identity;77 mod input;8+mod moq_catalog;89 mod server;910 mod source;1011 @@ -0,0 +1,106 @@
1+use anyhow::{Context, Result, ensure};2+use moq_net::BroadcastProducer;3+use serde::{Deserialize, Serialize};4+use sigil_protocol::{MAX_MOQ_CATALOG_BYTES, MoqCatalogExtensionV1};5+6+7+8+struct GoqCatalogDocument {9+ 10+ media: hang::Catalog,11+ goq: MoqCatalogExtensionV1,12+}13+14+impl GoqCatalogDocument {15+ fn video_h264() -> Self {16+ Self {17+ media: hang::Catalog::default(),18+ goq: MoqCatalogExtensionV1::video_h264(),19+ }20+ }21+22+ fn validate(&self) -> Result<()> {23+ ensure!(24+ self.media == hang::Catalog::default(),25+ "Goq catalog must not advertise a standard Hang rendition for enveloped media"26+ );27+ self.goq28+ .validate()29+ .context("validating Goq catalog extension")30+ }31+}32+33+pub(crate) struct GoqCatalogProducer {34+ producer: moq_json::Producer<GoqCatalogDocument>,35+}36+37+impl GoqCatalogProducer {38+ pub(crate) fn finish(mut self) -> Result<()> {39+ self.producer40+ .finish()41+ .context("finishing Goq catalog track")42+ }43+}44+45+pub(crate) fn publish_goq_catalog(broadcast: &mut BroadcastProducer) -> Result<GoqCatalogProducer> {46+ let track = broadcast47+ .create_track(hang::Catalog::default_track())48+ .context("creating catalog.json track")?;49+ let mut producer = moq_json::Producer::new(track, moq_json::Config::default());50+ let document = GoqCatalogDocument::video_h264();51+ document.validate()?;52+ let snapshot = serde_json::to_vec(&document).context("serializing Goq catalog snapshot")?;53+ ensure!(54+ snapshot.len() <= MAX_MOQ_CATALOG_BYTES,55+ "Goq catalog snapshot exceeds {MAX_MOQ_CATALOG_BYTES} bytes"56+ );57+ producer58+ .update(&document)59+ .context("publishing immutable Goq catalog snapshot")?;60+ Ok(GoqCatalogProducer { producer })61+}62+63+64+mod tests {65+ use super::*;66+ use moq_net::{Broadcast, Track};67+ use std::time::Duration;68+69+ 70+ async fn catalog_snapshot_is_late_subscribable_and_hang_compatible() {71+ let mut broadcast = Broadcast::new().produce();72+ let _video = broadcast73+ .create_track(Track {74+ name: sigil_protocol::MOQ_VIDEO_H264_TRACK.to_owned(),75+ priority: sigil_protocol::MOQ_VIDEO_TRACK_PRIORITY,76+ })77+ .unwrap();78+ let catalog = publish_goq_catalog(&mut broadcast).unwrap();79+80+ let base_track = broadcast81+ .consume()82+ .subscribe_track(&hang::Catalog::default_track())83+ .unwrap();84+ let mut base_consumer = moq_json::Consumer::<hang::Catalog>::new(base_track);85+ let base = tokio::time::timeout(Duration::from_millis(100), base_consumer.next())86+ .await87+ .unwrap()88+ .unwrap()89+ .unwrap();90+ assert_eq!(base, hang::Catalog::default());91+92+ let document_track = broadcast93+ .consume()94+ .subscribe_track(&hang::Catalog::default_track())95+ .unwrap();96+ let mut document_consumer = moq_json::Consumer::<GoqCatalogDocument>::new(document_track);97+ let document = tokio::time::timeout(Duration::from_millis(100), document_consumer.next())98+ .await99+ .unwrap()100+ .unwrap()101+ .unwrap();102+ document.validate().unwrap();103+ assert_eq!(document, GoqCatalogDocument::video_h264());104+ catalog.finish().unwrap();105+ }106+}@@ -18,9 +18,9 @@ use sigil_protocol::{
1818 FrameFlags, HostHello, InputAck, InvitationGrants, KeyframeRequestReasonV3,1919 MAX_AUDIO_PAYLOAD_LEN, MAX_MEDIA_GROUP_BYTES_V3, MAX_MEDIA_OBJECT_DELIVERY_TIMEOUT_MS,2020 MAX_MEDIA_OBJECT_ID_V3, MIN_MEDIA_OBJECT_DELIVERY_TIMEOUT_MS, MOQ_VIDEO_H264_TRACK,21- MediaControlRequestV3, MediaFeedbackFlags, MediaFeedbackReportV1, MediaFrame, MediaFrameHeader,22- MediaObjectHeaderV3, MediaObjectV3, encode_media_frame_object, media_moq_broadcast_name,23- read_client_hello, read_input_event, read_media_control_request_v3,21+ MOQ_VIDEO_TRACK_PRIORITY, MediaControlRequestV3, MediaFeedbackFlags, MediaFeedbackReportV1,22+ MediaFrame, MediaFrameHeader, MediaObjectHeaderV3, MediaObjectV3, encode_media_frame_object,23+ media_moq_broadcast_name, read_client_hello, read_input_event, read_media_control_request_v3,2424 read_media_feedback_report_v1, write_adaptive_bitrate_decision_v1, write_host_hello,2525 write_input_ack, write_media_frame, write_media_object_v3,2626 };@@ -32,6 +32,7 @@ use crate::clock::SessionClock;
3232 use crate::config::{GamescopeEncoderBackend, HostConfig, VaapiRateControl, VideoSource};3333 use crate::cursor::{PointerPositionTracker, PointerState};3434 use crate::input::{InputBackend, InputDisposition};35+use crate::moq_catalog::publish_goq_catalog;3536 use crate::source::{3637 EncodedFrame, EncodedGop, EncodedSource, EncoderControl,3738 spawn_gamescope_pipewire_after_static_preflight, spawn_test_pattern,@@ -69,7 +70,6 @@ const GAMESCOPE_STARTUP_TARGET_MIN_FRAMES: u64 = 8;
6970 const GAMESCOPE_STARTUP_TARGET_MIN_FPS: f64 = 45.0;7071 const SOURCE_REAP_GRACE_TIMEOUT: Duration = Duration::from_secs(1);7172 const MOQ_ATTACHMENT_TIMEOUT: Duration = Duration::from_secs(10);72-const MOQ_VIDEO_TRACK_PRIORITY: u8 = u8::MAX;7373 const MOQ_REJECT_CODE: u32 = 0x534d;7474 const ADAPTIVE_BITRATE_FLOOR_KBPS: u32 = 1_000;7575 const TEST_PATTERN_BITRATE_CEILING_KBPS: u32 = 12_000;@@ -2689,6 +2689,7 @@ async fn serve_control_moq(
26892689 priority: MOQ_VIDEO_TRACK_PRIORITY,26902690 })26912691 .context("creating static MoQ H.264 track")?;2692+ let catalog = publish_goq_catalog(&mut broadcast)?;26922693 let broadcast_name = media_moq_broadcast_name(lease.session_id)?;26932694 let attachment = sessions.expect_moq(26942695 remote,@@ -2743,13 +2744,17 @@ async fn serve_control_moq(
27432744 Arc::clone(&lease.media_v3_telemetry),27442745 )27452746 .await;2747+ let catalog_result = catalog.finish();27462748 27472749 drop(current_gop_receiver);27482750 drop(frame_receiver);27492751 source_task.wait_or_abort(SOURCE_REAP_GRACE_TIMEOUT).await;27502752 drop(lease);27512753 info!(%remote, "MoQ control client released");2752- session_result2754+ match session_result {2755+ Err(error) => Err(error),2756+ Ok(()) => catalog_result,2757+ }27532758 }27542759 27552760 @@ -14,6 +14,7 @@ mod input;
1414 mod invitation;1515 mod media;1616 mod media_v3;17+mod moq_catalog;1718 1819 pub use audio::{AUDIO_HEADER_LEN, AudioCodec, AudioFlags, AudioPacket, AudioPacketHeader};1920 pub use error::{ProtocolError, Result};@@ -50,6 +51,11 @@ pub use media_v3::{
5051 MediaObjectHeaderV3, MediaObjectV3, read_media_control_request_v3, read_media_object_v3,5152 write_media_control_request_v3, write_media_object_v3,5253 };54+pub use moq_catalog::{55+ MAX_MOQ_CATALOG_BYTES, MOQ_CATALOG_EXTENSION_VERSION_V1, MOQ_GOP_GROUP_FORMAT_V1,56+ MOQ_MEDIA_OBJECT_FORMAT_V1, MOQ_VIDEO_TRACK_PRIORITY, MoqCatalogExtensionV1,57+ MoqTrackDescriptorV1, MoqVideoCatalogV1,58+};5359 5460 /// Protocol version encoded in v1 messages.5561 pub const PROTOCOL_VERSION: u16 = 1;@@ -0,0 +1,139 @@
1+use serde::{Deserialize, Serialize};2+3+use crate::{MOQ_VIDEO_H264_TRACK, ProtocolError, Result};4+5+pub const MOQ_CATALOG_EXTENSION_VERSION_V1: u16 = 1;6+/// Maximum immutable catalog snapshot accepted from a peer before JSON decoding.7+pub const MAX_MOQ_CATALOG_BYTES: usize = 4 * 1024;8+pub const MOQ_MEDIA_OBJECT_FORMAT_V1: &str = "sigil/media-frame/1";9+pub const MOQ_GOP_GROUP_FORMAT_V1: &str = "sigil/moq-gop/1";10+pub const MOQ_VIDEO_TRACK_PRIORITY: u8 = u8::MAX;11+12+13+14+pub struct MoqTrackDescriptorV1 {15+ pub name: String,16+ pub priority: u8,17+}18+19+20+21+pub struct MoqVideoCatalogV1 {22+ pub track: MoqTrackDescriptorV1,23+ pub codec: String,24+ pub object_format: String,25+ pub group_format: String,26+}27+28+29+30+pub struct MoqCatalogExtensionV1 {31+ pub version: u16,32+ pub video: MoqVideoCatalogV1,33+}34+35+impl MoqCatalogExtensionV1 {36+ pub fn video_h264() -> Self {37+ Self {38+ version: MOQ_CATALOG_EXTENSION_VERSION_V1,39+ video: MoqVideoCatalogV1 {40+ track: MoqTrackDescriptorV1 {41+ name: MOQ_VIDEO_H264_TRACK.to_owned(),42+ priority: MOQ_VIDEO_TRACK_PRIORITY,43+ },44+ codec: "h264".to_owned(),45+ object_format: MOQ_MEDIA_OBJECT_FORMAT_V1.to_owned(),46+ group_format: MOQ_GOP_GROUP_FORMAT_V1.to_owned(),47+ },48+ }49+ }50+51+ pub fn validate(&self) -> Result<()> {52+ if self.version != MOQ_CATALOG_EXTENSION_VERSION_V1 {53+ return Err(ProtocolError::InvalidMessage {54+ message_type: "Goq MoQ catalog extension",55+ reason: "unsupported extension version",56+ });57+ }58+ if self.video.track.name != MOQ_VIDEO_H264_TRACK {59+ return Err(ProtocolError::InvalidMessage {60+ message_type: "Goq MoQ catalog extension",61+ reason: "unexpected video track name",62+ });63+ }64+ if self.video.track.priority != MOQ_VIDEO_TRACK_PRIORITY {65+ return Err(ProtocolError::InvalidMessage {66+ message_type: "Goq MoQ catalog extension",67+ reason: "unexpected video track priority",68+ });69+ }70+ if self.video.codec != "h264" {71+ return Err(ProtocolError::InvalidMessage {72+ message_type: "Goq MoQ catalog extension",73+ reason: "unsupported video codec",74+ });75+ }76+ if self.video.object_format != MOQ_MEDIA_OBJECT_FORMAT_V1 {77+ return Err(ProtocolError::InvalidMessage {78+ message_type: "Goq MoQ catalog extension",79+ reason: "unsupported media object format",80+ });81+ }82+ if self.video.group_format != MOQ_GOP_GROUP_FORMAT_V1 {83+ return Err(ProtocolError::InvalidMessage {84+ message_type: "Goq MoQ catalog extension",85+ reason: "unsupported GOP group format",86+ });87+ }88+ Ok(())89+ }90+}91+92+93+mod tests {94+ use super::*;95+96+ 97+ fn goq_catalog_extension_has_a_stable_golden_document() {98+ let extension = MoqCatalogExtensionV1::video_h264();99+ extension.validate().unwrap();100+ assert_eq!(101+ serde_json::to_string(&extension).unwrap(),102+ r#"{"version":1,"video":{"track":{"name":"video/h264","priority":255},"codec":"h264","objectFormat":"sigil/media-frame/1","groupFormat":"sigil/moq-gop/1"}}"#103+ );104+ }105+106+ 107+ fn goq_catalog_validation_rejects_every_mismatched_contract_field() {108+ let mut cases = Vec::new();109+110+ let mut value = MoqCatalogExtensionV1::video_h264();111+ value.version = 2;112+ cases.push(value);113+ let mut value = MoqCatalogExtensionV1::video_h264();114+ value.video.track.name = "video/other".into();115+ cases.push(value);116+ let mut value = MoqCatalogExtensionV1::video_h264();117+ value.video.track.priority = 0;118+ cases.push(value);119+ let mut value = MoqCatalogExtensionV1::video_h264();120+ value.video.codec = "av1".into();121+ cases.push(value);122+ let mut value = MoqCatalogExtensionV1::video_h264();123+ value.video.object_format = "hang/legacy".into();124+ cases.push(value);125+ let mut value = MoqCatalogExtensionV1::video_h264();126+ value.video.group_format = "hang/gop".into();127+ cases.push(value);128+129+ for invalid in cases {130+ assert!(invalid.validate().is_err(), "accepted {invalid:?}");131+ }132+ }133+134+ 135+ fn goq_catalog_extension_rejects_unknown_fields() {136+ let json = r#"{"version":1,"video":{"track":{"name":"video/h264","priority":255,"extra":true},"codec":"h264","objectFormat":"sigil/media-frame/1","groupFormat":"sigil/moq-gop/1"}}"#;137+ assert!(serde_json::from_str::<MoqCatalogExtensionV1>(json).is_err());138+ }139+}@@ -429,6 +429,7 @@ validate_probe_log() {
429429 grep -Fxq 'control_alpn=sigil/control/1' "$log_path" || return 1430430 grep -Fxq 'group_sequence=monotonic' "$log_path" || return 1431431 grep -Fxq 'moq_group_capacity=1' "$log_path" || return 1432+ grep -Fxq 'moq_catalog=goq-v1' "$log_path" || return 1432433 grep -Fxq 'moq_unrecovered_group_gaps=0' "$log_path" || return 1433434 grep -Fxq 'moq_historical_suffix_frames=0' "$log_path" || return 1434435 grep -Fxq 'recovery_barrier=configured-idr' "$log_path" || return 1@@ -622,7 +623,7 @@ printf 'probe_binary=%s\n' "$probe_bin"
622623 printf 'probe_sha256=%s\n' "$(sha256_file "$probe_bin")"623624 printf 'node_id=%s\n' "$node_id"624625 printf 'primary_frames=%s\n' "$primary_frames"625-grep -E '^(transport|control_alpn|transport_alpn|first_configured_idr|frame_sequence|group_sequence|keyframe_recovery|keyframe_request_id|keyframes|sequence_gaps|media_objects_dropped|media_objects_late|moq_group_capacity|moq_cancelled_groups|moq_group_gaps|moq_unrecovered_group_gaps|moq_maximum_group_objects|moq_maximum_group_bytes|moq_historical_suffix_frames|recovery_barrier|input_ack_micros|path_mode|path_rtt_ms)=' "$primary_log"626+grep -E '^(transport|control_alpn|transport_alpn|first_configured_idr|frame_sequence|group_sequence|keyframe_recovery|keyframe_request_id|keyframes|sequence_gaps|media_objects_dropped|media_objects_late|moq_catalog|moq_group_capacity|moq_cancelled_groups|moq_group_gaps|moq_unrecovered_group_gaps|moq_maximum_group_objects|moq_maximum_group_bytes|moq_historical_suffix_frames|recovery_barrier|input_ack_micros|path_mode|path_rtt_ms)=' "$primary_log"626627 printf 'keyframe_request_correlation=%s\n' \627628 "$([[ "$media_v1" == false && "$media_v2" == false ]] && printf unique || printf not-requested)"628629 printf 'active_client_rejection=ok\n'@@ -101,6 +101,16 @@ if grep -Eiq '(^|[[:space:]├└│─])(tauri|wry|webkit)([[:space:]-]|$)' <<<
101101 exit 1102102 fi103103 104+catalog_dependencies="$({105+ cargo tree --locked -p sigil-host --edges normal106+ cargo tree --locked -p portal --edges normal107+})"108+if grep -Eiq '(^|[[:space:]├└│─])(moq-media|moq-mux)([[:space:]-]|$)' <<<"$catalog_dependencies"; then109+ echo 'MoQ catalog boundary failed: standard-media dependency detected' >&2110+ grep -Ei 'moq-media|moq-mux' <<<"$catalog_dependencies" >&2111+ exit 1112+fi113+104114 ./scripts/loopback-proof.sh105115 106116 cargo test --locked -p portal --release \@@ -0,0 +1,304 @@
1+use std::time::Duration;2+3+use moq_net::{BroadcastConsumer, Error as MoqError, Track, TrackConsumer};4+use serde::{Deserialize, Serialize};5+use sigil_protocol::{MAX_MOQ_CATALOG_BYTES, MoqCatalogExtensionV1};6+7+8+9+struct GoqCatalogDocument {10+ 11+ media: hang::Catalog,12+ goq: MoqCatalogExtensionV1,13+}14+15+impl GoqCatalogDocument {16+ 17+ fn video_h264() -> Self {18+ Self {19+ media: hang::Catalog::default(),20+ goq: MoqCatalogExtensionV1::video_h264(),21+ }22+ }23+24+ fn validate(&self) -> Result<(), String> {25+ if self.media != hang::Catalog::default() {26+ return Err(27+ "Goq catalog falsely advertises a standard Hang rendition for enveloped media"28+ .to_string(),29+ );30+ }31+ self.goq32+ .validate()33+ .map_err(|error| format!("Invalid Goq catalog extension: {error}"))34+ }35+}36+37+38+pub(super) enum MoqCatalogMode {39+ GoqV1,40+ AbsentStaticTrackCompat,41+}42+43+impl MoqCatalogMode {44+ pub(super) const fn label(self) -> &'static str {45+ match self {46+ Self::GoqV1 => "goq-v1",47+ Self::AbsentStaticTrackCompat => "absent-static-track-compat",48+ }49+ }50+}51+52+pub(super) struct MoqCatalogSelection {53+ pub(super) track: TrackConsumer,54+ pub(super) mode: MoqCatalogMode,55+}56+57+fn is_track_not_found(error: &MoqError) -> bool {58+ matches!(error, MoqError::NotFound)59+ || matches!(error, MoqError::Remote(code) if *code == MoqError::NotFound.to_code())60+}61+62+fn subscribe_static_video_track(63+ broadcast: &BroadcastConsumer,64+) -> Result<MoqCatalogSelection, String> {65+ let track = broadcast66+ .subscribe_track(&Track::new(sigil_protocol::MOQ_VIDEO_H264_TRACK))67+ .map_err(|error| {68+ format!("Failed to subscribe to static MoQ H.264 compatibility track: {error}")69+ })?;70+ Ok(MoqCatalogSelection {71+ track,72+ mode: MoqCatalogMode::AbsentStaticTrackCompat,73+ })74+}75+76+pub(super) async fn subscribe_goq_video_track(77+ broadcast: &BroadcastConsumer,78+ timeout: Duration,79+) -> Result<MoqCatalogSelection, String> {80+ let catalog_track = match broadcast.subscribe_track(&hang::Catalog::default_track()) {81+ Ok(track) => track,82+ Err(MoqError::NotFound) => {83+ return subscribe_static_video_track(broadcast);84+ }85+ Err(error) => {86+ return Err(format!(87+ "Failed to subscribe to catalog.json track: {error}"88+ ));89+ }90+ };91+92+ let mut catalog_track = catalog_track;93+ let deadline = tokio::time::Instant::now() + timeout;94+ let mut group = match tokio::time::timeout_at(deadline, catalog_track.next_group()).await {95+ Err(_) => return Err("Timed out waiting for Goq catalog snapshot".to_string()),96+ Ok(Err(error)) if is_track_not_found(&error) => {97+ return subscribe_static_video_track(broadcast);98+ }99+ Ok(Err(error)) => return Err(format!("Failed to read Goq catalog snapshot: {error}")),100+ Ok(Ok(None)) => return Err("Goq catalog track ended before its snapshot".to_string()),101+ Ok(Ok(Some(group))) => group,102+ };103+ if group.sequence != 0 {104+ return Err("Goq catalog snapshot must be immutable group 0".to_string());105+ }106+ let frame_count = tokio::time::timeout_at(deadline, group.finished())107+ .await108+ .map_err(|_| "Timed out waiting for Goq catalog group completion".to_string())?109+ .map_err(|error| format!("Failed to finish Goq catalog group: {error}"))?;110+ if frame_count != 1 {111+ return Err("Goq catalog snapshot group must contain exactly one frame".to_string());112+ }113+ let mut frame = tokio::time::timeout_at(deadline, group.get_frame(0))114+ .await115+ .map_err(|_| "Timed out waiting for Goq catalog frame".to_string())?116+ .map_err(|error| format!("Failed to open Goq catalog frame: {error}"))?117+ .ok_or_else(|| "Goq catalog snapshot group has no frame 0".to_string())?;118+ if frame.size > MAX_MOQ_CATALOG_BYTES as u64 {119+ return Err(format!(120+ "Goq catalog snapshot exceeds {MAX_MOQ_CATALOG_BYTES} bytes"121+ ));122+ }123+ let snapshot = tokio::time::timeout_at(deadline, frame.read_all())124+ .await125+ .map_err(|_| "Timed out reading Goq catalog frame".to_string())?126+ .map_err(|error| format!("Failed to read Goq catalog frame: {error}"))?;127+ let document: GoqCatalogDocument = serde_json::from_slice(&snapshot)128+ .map_err(|error| format!("Failed to decode Goq catalog snapshot: {error}"))?;129+ document.validate()?;130+ let track = broadcast131+ .subscribe_track(&Track::new(document.goq.video.track.name))132+ .map_err(|error| {133+ format!("Failed to subscribe to catalog-selected Goq video track: {error}")134+ })?;135+ Ok(MoqCatalogSelection {136+ track,137+ mode: MoqCatalogMode::GoqV1,138+ })139+}140+141+142+mod tests {143+ use super::*;144+ use moq_net::Broadcast;145+146+ fn publish_catalog(147+ broadcast: &mut moq_net::BroadcastProducer,148+ document: &GoqCatalogDocument,149+ ) -> moq_net::TrackProducer {150+ let mut track = broadcast151+ .create_track(hang::Catalog::default_track())152+ .unwrap();153+ track154+ .write_frame(serde_json::to_vec(document).unwrap())155+ .unwrap();156+ track157+ }158+159+ 160+ async fn catalog_selects_the_exact_goq_video_track() {161+ let mut broadcast = Broadcast::new().produce();162+ let _video = broadcast163+ .create_track(Track::new(sigil_protocol::MOQ_VIDEO_H264_TRACK))164+ .unwrap();165+ let _catalog = publish_catalog(&mut broadcast, &GoqCatalogDocument::video_h264());166+ let selection = subscribe_goq_video_track(&broadcast.consume(), Duration::from_millis(100))167+ .await168+ .unwrap();169+ assert_eq!(selection.mode, MoqCatalogMode::GoqV1);170+ }171+172+ 173+ async fn only_an_absent_catalog_uses_the_legacy_static_track() {174+ let mut broadcast = Broadcast::new().produce();175+ let _video = broadcast176+ .create_track(Track::new(sigil_protocol::MOQ_VIDEO_H264_TRACK))177+ .unwrap();178+ let selection = subscribe_goq_video_track(&broadcast.consume(), Duration::from_millis(100))179+ .await180+ .unwrap();181+ assert_eq!(selection.mode, MoqCatalogMode::AbsentStaticTrackCompat);182+ }183+184+ 185+ async fn asynchronous_remote_not_found_uses_static_track_compatibility() {186+ let mut broadcast = Broadcast::new().produce();187+ let _video = broadcast188+ .create_track(Track::new(sigil_protocol::MOQ_VIDEO_H264_TRACK))189+ .unwrap();190+ let mut dynamic = broadcast.dynamic();191+ let consumer = broadcast.consume();192+193+ let select = subscribe_goq_video_track(&consumer, Duration::from_millis(100));194+ let reject = async {195+ let mut requested = dynamic.requested_track().await.unwrap();196+ assert_eq!(requested.name, hang::Catalog::default_track().name);197+ requested198+ .abort(MoqError::Remote(MoqError::NotFound.to_code()))199+ .unwrap();200+ };201+ let (selection, ()) = tokio::join!(select, reject);202+ assert_eq!(203+ selection.unwrap().mode,204+ MoqCatalogMode::AbsentStaticTrackCompat205+ );206+ }207+208+ 209+ async fn asynchronous_non_not_found_error_fails_closed() {210+ let mut broadcast = Broadcast::new().produce();211+ let _video = broadcast212+ .create_track(Track::new(sigil_protocol::MOQ_VIDEO_H264_TRACK))213+ .unwrap();214+ let mut dynamic = broadcast.dynamic();215+ let consumer = broadcast.consume();216+217+ let select = subscribe_goq_video_track(&consumer, Duration::from_millis(100));218+ let reject = async {219+ let mut requested = dynamic.requested_track().await.unwrap();220+ requested221+ .abort(MoqError::Remote(MoqError::WrongSize.to_code()))222+ .unwrap();223+ };224+ let (selection, ()) = tokio::join!(select, reject);225+ assert!(226+ selection227+ .err()228+ .is_some_and(|error| error.contains("remote error: code=14"))229+ );230+ }231+232+ 233+ async fn present_invalid_or_stalled_catalogs_fail_closed() {234+ let mut invalid_broadcast = Broadcast::new().produce();235+ let _video = invalid_broadcast236+ .create_track(Track::new(sigil_protocol::MOQ_VIDEO_H264_TRACK))237+ .unwrap();238+ let mut invalid = GoqCatalogDocument::video_h264();239+ invalid.goq.video.object_format = "hang/legacy".into();240+ let _catalog = publish_catalog(&mut invalid_broadcast, &invalid);241+ let result =242+ subscribe_goq_video_track(&invalid_broadcast.consume(), Duration::from_millis(100))243+ .await;244+ assert!(245+ result246+ .err()247+ .is_some_and(|error| error.contains("Invalid Goq catalog extension"))248+ );249+250+ let mut stalled_broadcast = Broadcast::new().produce();251+ let _catalog = stalled_broadcast252+ .create_track(hang::Catalog::default_track())253+ .unwrap();254+ let result =255+ subscribe_goq_video_track(&stalled_broadcast.consume(), Duration::from_millis(10))256+ .await;257+ assert!(258+ result259+ .err()260+ .is_some_and(|error| error.contains("Timed out"))261+ );262+263+ let mut oversized_broadcast = Broadcast::new().produce();264+ let _video = oversized_broadcast265+ .create_track(Track::new(sigil_protocol::MOQ_VIDEO_H264_TRACK))266+ .unwrap();267+ let mut catalog = oversized_broadcast268+ .create_track(hang::Catalog::default_track())269+ .unwrap();270+ catalog271+ .write_frame(vec![b' '; MAX_MOQ_CATALOG_BYTES + 1])272+ .unwrap();273+ let result =274+ subscribe_goq_video_track(&oversized_broadcast.consume(), Duration::from_millis(100))275+ .await;276+ assert!(277+ result278+ .err()279+ .is_some_and(|error| error.contains("exceeds 4096 bytes"))280+ );281+282+ let mut multi_frame_broadcast = Broadcast::new().produce();283+ let _video = multi_frame_broadcast284+ .create_track(Track::new(sigil_protocol::MOQ_VIDEO_H264_TRACK))285+ .unwrap();286+ let mut catalog = multi_frame_broadcast287+ .create_track(hang::Catalog::default_track())288+ .unwrap();289+ let mut group = catalog.append_group().unwrap();290+ group291+ .write_frame(serde_json::to_vec(&GoqCatalogDocument::video_h264()).unwrap())292+ .unwrap();293+ group.write_frame(b"{}".as_slice()).unwrap();294+ group.finish().unwrap();295+ let result =296+ subscribe_goq_video_track(&multi_frame_broadcast.consume(), Duration::from_millis(100))297+ .await;298+ assert!(299+ result300+ .err()301+ .is_some_and(|error| error.contains("exactly one frame"))302+ );303+ }304+}@@ -1,5 +1,6 @@
11 use super::auth::derive_iroh_secret_from_key;22 use super::enrollment::{connection_enrollment, mark_invitation_redeemed};3+use super::moq_catalog::subscribe_goq_video_track;34 use super::state::{45 AUDIO_DELIVERY_CAPACITY, AccumulatedMediaFeedback, AppState, AudioDeliveryState,56 CLIENT_INPUT_QUEUE_CAPACITY, FRAME_ALPN, INPUT_ALPN, MediaFeedbackSender,@@ -8,22 +9,26 @@ use super::state::{
89 use base64::Engine;910 use iroh::{Endpoint, SecretKey, endpoint::presets};1011 use iroh_moq::{Moq, MoqSession};11-use moq_net::{BroadcastConsumer, GroupConsumer, Track, TrackConsumer};12+13+use moq_net::Track;14+use moq_net::{BroadcastConsumer, GroupConsumer, TrackConsumer};1215 use openh264::{formats::YUVSource, nal_units};1316 use serde::{Deserialize, Serialize};17+18+use sigil_protocol::MOQ_VIDEO_H264_TRACK;1419 use sigil_protocol::{1520 AUDIO_ALPN_V1, AdaptiveBitrateDecisionV1, AdaptiveBitrateReasonFlagsV1, AdaptiveBitrateStateV1,1621 AudioFlags, AudioPacket, AudioPacketHeader, CONTROL_ALPN_V1, Capability, ClientHello,1722 FrameFlags, INPUT_ALPN_V1, InputEvent, InvitationGrants, KeyframeRequestReasonV3,1823 MAX_MEDIA_GROUP_BYTES_V3, MAX_MEDIA_OBJECT_DELIVERY_TIMEOUT_MS, MAX_MEDIA_OBJECT_ID_V3,1924 MAX_MEDIA_PAYLOAD_LEN, MAX_VIDEO_DIMENSION, MAX_VIDEO_PIXELS, MEDIA_ALPN_V1, MEDIA_ALPN_V2,20- MEDIA_ALPN_V3, MEDIA_FEEDBACK_ALPN_V1, MOQ_VIDEO_H264_TRACK, MediaCodec, MediaControlRequestV3,21- MediaFeedbackFlags, MediaFeedbackReportV1, MediaFrame, MediaObjectV3, PointerPosition,22- PointerSurfaceDimensions, ProtocolError, RELATIVE_POINTER_DELTA_MAX,23- RELATIVE_POINTER_DELTA_MIN, decode_media_frame_object, media_moq_broadcast_name,24- read_adaptive_bitrate_decision_v1, read_host_hello, read_input_ack, read_media_frame,25- read_media_object, read_media_object_v3, write_client_hello, write_input_event,26- write_media_control_request_v3, write_media_feedback_report_v1,25+ MEDIA_ALPN_V3, MEDIA_FEEDBACK_ALPN_V1, MediaCodec, MediaControlRequestV3, MediaFeedbackFlags,26+ MediaFeedbackReportV1, MediaFrame, MediaObjectV3, PointerPosition, PointerSurfaceDimensions,27+ ProtocolError, RELATIVE_POINTER_DELTA_MAX, RELATIVE_POINTER_DELTA_MIN,28+ decode_media_frame_object, media_moq_broadcast_name, read_adaptive_bitrate_decision_v1,29+ read_host_hello, read_input_ack, read_media_frame, read_media_object, read_media_object_v3,30+ write_client_hello, write_input_event, write_media_control_request_v3,31+ write_media_feedback_report_v1,2732 };2833 use std::collections::{BTreeMap, VecDeque};2934 use std::io::Cursor;@@ -2356,13 +2361,10 @@ async fn open_upstream_moq_media(
23562361 .map_err(|error| {23572362 format!("Failed to subscribe to upstream MoQ broadcast {broadcast_name}: {error}")23582363 })?;2359- let track = broadcast2360- .subscribe_track(&Track::new(MOQ_VIDEO_H264_TRACK))2361- .map_err(|error| {2362- format!("Failed to subscribe to upstream MoQ track {MOQ_VIDEO_H264_TRACK}: {error}")2363- })?;2364+ let catalog = subscribe_goq_video_track(&broadcast, CLIENT_MOQ_SUBSCRIBE_TIMEOUT).await?;2365+ eprintln!("[client] moq catalog: {}", catalog.mode.label());23642366 Ok((2365- MoqMediaReceiver::new(moq, session, broadcast, track),2367+ MoqMediaReceiver::new(moq, session, broadcast, catalog.track),23662368 diagnostics_connection,23672369 ))23682370 }