1use std::time::Duration;23use anyhow::{Context, Result};4use moq_net::{BroadcastConsumer, Error as MoqError, Track, TrackConsumer};5use sigil_protocol::{GoqCatalogDocument, MAX_MOQ_CATALOG_BYTES};678pub(crate) enum MoqCatalogMode {9 GoqV1,10 AbsentStaticTrackCompat,11}1213impl MoqCatalogMode {14 pub(crate) const fn label(self) -> &'static str {15 match self {16 Self::GoqV1 => "goq-v1",17 Self::AbsentStaticTrackCompat => "absent-static-track-compat",18 }19 }20}2122pub(crate) struct MoqCatalogSelection {23 pub(crate) track: TrackConsumer,24 pub(crate) mode: MoqCatalogMode,25}2627fn is_track_not_found(error: &MoqError) -> bool {28 matches!(error, MoqError::NotFound)29 || matches!(error, MoqError::Remote(code) if *code == MoqError::NotFound.to_code())30}3132fn subscribe_static_video_track(broadcast: &BroadcastConsumer) -> Result<MoqCatalogSelection> {33 let track = broadcast34 .subscribe_track(&Track::new(sigil_protocol::MOQ_VIDEO_H264_TRACK))35 .context("subscribing to static MoQ H.264 compatibility track")?;36 Ok(MoqCatalogSelection {37 track,38 mode: MoqCatalogMode::AbsentStaticTrackCompat,39 })40}4142pub(crate) async fn subscribe_goq_video_track(43 broadcast: &BroadcastConsumer,44 timeout: Duration,45) -> Result<MoqCatalogSelection> {46 let catalog_track = match broadcast.subscribe_track(&hang::Catalog::default_track()) {47 Ok(track) => track,48 Err(MoqError::NotFound) => {49 return subscribe_static_video_track(broadcast);50 }51 Err(error) => return Err(error).context("subscribing to catalog.json track"),52 };5354 let mut catalog_track = catalog_track;55 let deadline = tokio::time::Instant::now() + timeout;56 let mut group = match tokio::time::timeout_at(deadline, catalog_track.next_group()).await {57 Err(_) => anyhow::bail!("timed out waiting for Goq catalog snapshot"),58 Ok(Err(error)) if is_track_not_found(&error) => {59 return subscribe_static_video_track(broadcast);60 }61 Ok(Err(error)) => return Err(error).context("reading Goq catalog snapshot"),62 Ok(Ok(None)) => anyhow::bail!("Goq catalog track ended before its snapshot"),63 Ok(Ok(Some(group))) => group,64 };65 anyhow::ensure!(66 group.sequence == 0,67 "Goq catalog snapshot must be immutable group 0"68 );69 let frame_count = tokio::time::timeout_at(deadline, group.finished())70 .await71 .context("timed out waiting for Goq catalog group completion")?72 .context("finishing Goq catalog group")?;73 anyhow::ensure!(74 frame_count == 1,75 "Goq catalog snapshot group must contain exactly one frame"76 );77 let mut frame = tokio::time::timeout_at(deadline, group.get_frame(0))78 .await79 .context("timed out waiting for Goq catalog frame")?80 .context("opening Goq catalog frame")?81 .context("Goq catalog snapshot group has no frame 0")?;82 anyhow::ensure!(83 frame.size <= MAX_MOQ_CATALOG_BYTES as u64,84 "Goq catalog snapshot exceeds {MAX_MOQ_CATALOG_BYTES} bytes"85 );86 let snapshot = tokio::time::timeout_at(deadline, frame.read_all())87 .await88 .context("timed out reading Goq catalog frame")?89 .context("reading Goq catalog frame")?;90 let document: GoqCatalogDocument =91 serde_json::from_slice(&snapshot).context("decoding Goq catalog snapshot")?;92 document93 .validate()94 .context("validating Goq catalog document")?;95 let track = broadcast96 .subscribe_track(&Track::new(document.goq.video.track.name))97 .context("subscribing to catalog-selected Goq video track")?;98 Ok(MoqCatalogSelection {99 track,100 mode: MoqCatalogMode::GoqV1,101 })102}103104105mod tests {106 use super::*;107 use moq_net::Broadcast;108109 fn publish_catalog(110 broadcast: &mut moq_net::BroadcastProducer,111 document: &GoqCatalogDocument,112 ) -> moq_net::TrackProducer {113 let mut track = broadcast114 .create_track(hang::Catalog::default_track())115 .unwrap();116 track117 .write_frame(serde_json::to_vec(document).unwrap())118 .unwrap();119 track120 }121122 123 async fn catalog_selects_the_exact_goq_video_track() {124 let mut broadcast = Broadcast::new().produce();125 let _video = broadcast126 .create_track(Track::new(sigil_protocol::MOQ_VIDEO_H264_TRACK))127 .unwrap();128 let _catalog = publish_catalog(&mut broadcast, &GoqCatalogDocument::video_h264());129 let selection = subscribe_goq_video_track(&broadcast.consume(), Duration::from_millis(100))130 .await131 .unwrap();132 assert_eq!(selection.mode, MoqCatalogMode::GoqV1);133 }134135 136 async fn only_an_absent_catalog_uses_the_legacy_static_track() {137 let mut broadcast = Broadcast::new().produce();138 let _video = broadcast139 .create_track(Track::new(sigil_protocol::MOQ_VIDEO_H264_TRACK))140 .unwrap();141 let selection = subscribe_goq_video_track(&broadcast.consume(), Duration::from_millis(100))142 .await143 .unwrap();144 assert_eq!(selection.mode, MoqCatalogMode::AbsentStaticTrackCompat);145 }146147 148 async fn asynchronous_remote_not_found_uses_static_track_compatibility() {149 let mut broadcast = Broadcast::new().produce();150 let _video = broadcast151 .create_track(Track::new(sigil_protocol::MOQ_VIDEO_H264_TRACK))152 .unwrap();153 let mut dynamic = broadcast.dynamic();154 let consumer = broadcast.consume();155156 let select = subscribe_goq_video_track(&consumer, Duration::from_millis(100));157 let reject = async {158 let mut requested = dynamic.requested_track().await.unwrap();159 assert_eq!(requested.name, hang::Catalog::default_track().name);160 requested161 .abort(MoqError::Remote(MoqError::NotFound.to_code()))162 .unwrap();163 };164 let (selection, ()) = tokio::join!(select, reject);165 assert_eq!(166 selection.unwrap().mode,167 MoqCatalogMode::AbsentStaticTrackCompat168 );169 }170171 172 async fn asynchronous_non_not_found_error_fails_closed() {173 let mut broadcast = Broadcast::new().produce();174 let _video = broadcast175 .create_track(Track::new(sigil_protocol::MOQ_VIDEO_H264_TRACK))176 .unwrap();177 let mut dynamic = broadcast.dynamic();178 let consumer = broadcast.consume();179180 let select = subscribe_goq_video_track(&consumer, Duration::from_millis(100));181 let reject = async {182 let mut requested = dynamic.requested_track().await.unwrap();183 requested184 .abort(MoqError::Remote(MoqError::WrongSize.to_code()))185 .unwrap();186 };187 let (selection, ()) = tokio::join!(select, reject);188 let error = selection.err().expect("non-NotFound must fail closed");189 assert!(190 format!("{error:#}").contains("remote error: code=14"),191 "{error:#}"192 );193 }194195 196 async fn present_invalid_or_stalled_catalogs_fail_closed() {197 let mut invalid_broadcast = Broadcast::new().produce();198 let _video = invalid_broadcast199 .create_track(Track::new(sigil_protocol::MOQ_VIDEO_H264_TRACK))200 .unwrap();201 let mut invalid = GoqCatalogDocument::video_h264();202 invalid.goq.video.object_format = "hang/legacy".into();203 let _catalog = publish_catalog(&mut invalid_broadcast, &invalid);204 let result =205 subscribe_goq_video_track(&invalid_broadcast.consume(), Duration::from_millis(100))206 .await;207 assert!(result.err().is_some_and(|error| {208 format!("{error:#}").contains("validating Goq catalog document")209 }));210211 let mut stalled_broadcast = Broadcast::new().produce();212 let _catalog = stalled_broadcast213 .create_track(hang::Catalog::default_track())214 .unwrap();215 let result =216 subscribe_goq_video_track(&stalled_broadcast.consume(), Duration::from_millis(10))217 .await;218 assert!(219 result220 .err()221 .is_some_and(|error| error.to_string().contains("timed out"))222 );223224 let mut oversized_broadcast = Broadcast::new().produce();225 let _video = oversized_broadcast226 .create_track(Track::new(sigil_protocol::MOQ_VIDEO_H264_TRACK))227 .unwrap();228 let mut catalog = oversized_broadcast229 .create_track(hang::Catalog::default_track())230 .unwrap();231 catalog232 .write_frame(vec![b' '; MAX_MOQ_CATALOG_BYTES + 1])233 .unwrap();234 let result =235 subscribe_goq_video_track(&oversized_broadcast.consume(), Duration::from_millis(100))236 .await;237 assert!(238 result239 .err()240 .is_some_and(|error| error.to_string().contains("exceeds 4096 bytes"))241 );242243 let mut multi_frame_broadcast = Broadcast::new().produce();244 let _video = multi_frame_broadcast245 .create_track(Track::new(sigil_protocol::MOQ_VIDEO_H264_TRACK))246 .unwrap();247 let mut catalog = multi_frame_broadcast248 .create_track(hang::Catalog::default_track())249 .unwrap();250 let mut group = catalog.append_group().unwrap();251 group252 .write_frame(serde_json::to_vec(&GoqCatalogDocument::video_h264()).unwrap())253 .unwrap();254 group.write_frame(b"{}".as_slice()).unwrap();255 group.finish().unwrap();256 let result =257 subscribe_goq_video_track(&multi_frame_broadcast.consume(), Duration::from_millis(100))258 .await;259 assert!(260 result261 .err()262 .is_some_and(|error| error.to_string().contains("exactly one frame"))263 );264 }265}