1use std::time::Duration;23use moq_net::{BroadcastConsumer, Error as MoqError, Track, TrackConsumer};4use sigil_protocol::{GoqCatalogDocument, MAX_MOQ_CATALOG_BYTES};567pub(crate) enum MoqCatalogMode {8 GoqV1,9 AbsentStaticTrackCompat,10}1112impl MoqCatalogMode {13 pub(crate) const fn label(self) -> &'static str {14 match self {15 Self::GoqV1 => "goq-v1",16 Self::AbsentStaticTrackCompat => "absent-static-track-compat",17 }18 }19}2021pub(crate) struct MoqCatalogSelection {22 pub(crate) track: TrackConsumer,23 pub(crate) mode: MoqCatalogMode,24}2526fn is_track_not_found(error: &MoqError) -> bool {27 matches!(error, MoqError::NotFound)28 || matches!(error, MoqError::Remote(code) if *code == MoqError::NotFound.to_code())29}3031fn subscribe_static_video_track(32 broadcast: &BroadcastConsumer,33) -> Result<MoqCatalogSelection, String> {34 let track = broadcast35 .subscribe_track(&Track::new(sigil_protocol::MOQ_VIDEO_H264_TRACK))36 .map_err(|error| {37 format!("Failed to subscribe to static MoQ H.264 compatibility track: {error}")38 })?;39 Ok(MoqCatalogSelection {40 track,41 mode: MoqCatalogMode::AbsentStaticTrackCompat,42 })43}4445pub(crate) async fn subscribe_goq_video_track(46 broadcast: &BroadcastConsumer,47 timeout: Duration,48) -> Result<MoqCatalogSelection, String> {49 let catalog_track = match broadcast.subscribe_track(&hang::Catalog::default_track()) {50 Ok(track) => track,51 Err(MoqError::NotFound) => {52 return subscribe_static_video_track(broadcast);53 }54 Err(error) => {55 return Err(format!(56 "Failed to subscribe to catalog.json track: {error}"57 ));58 }59 };6061 let mut catalog_track = catalog_track;62 let deadline = tokio::time::Instant::now() + timeout;63 let mut group = match tokio::time::timeout_at(deadline, catalog_track.next_group()).await {64 Err(_) => return Err("Timed out waiting for Goq catalog snapshot".to_string()),65 Ok(Err(error)) if is_track_not_found(&error) => {66 return subscribe_static_video_track(broadcast);67 }68 Ok(Err(error)) => return Err(format!("Failed to read Goq catalog snapshot: {error}")),69 Ok(Ok(None)) => return Err("Goq catalog track ended before its snapshot".to_string()),70 Ok(Ok(Some(group))) => group,71 };72 if group.sequence != 0 {73 return Err("Goq catalog snapshot must be immutable group 0".to_string());74 }75 let frame_count = tokio::time::timeout_at(deadline, group.finished())76 .await77 .map_err(|_| "Timed out waiting for Goq catalog group completion".to_string())?78 .map_err(|error| format!("Failed to finish Goq catalog group: {error}"))?;79 if frame_count != 1 {80 return Err("Goq catalog snapshot group must contain exactly one frame".to_string());81 }82 let mut frame = tokio::time::timeout_at(deadline, group.get_frame(0))83 .await84 .map_err(|_| "Timed out waiting for Goq catalog frame".to_string())?85 .map_err(|error| format!("Failed to open Goq catalog frame: {error}"))?86 .ok_or_else(|| "Goq catalog snapshot group has no frame 0".to_string())?;87 if frame.size > MAX_MOQ_CATALOG_BYTES as u64 {88 return Err(format!(89 "Goq catalog snapshot exceeds {MAX_MOQ_CATALOG_BYTES} bytes"90 ));91 }92 let snapshot = tokio::time::timeout_at(deadline, frame.read_all())93 .await94 .map_err(|_| "Timed out reading Goq catalog frame".to_string())?95 .map_err(|error| format!("Failed to read Goq catalog frame: {error}"))?;96 let document: GoqCatalogDocument = serde_json::from_slice(&snapshot)97 .map_err(|error| format!("Failed to decode Goq catalog snapshot: {error}"))?;98 document99 .validate()100 .map_err(|error| format!("Invalid Goq catalog: {error}"))?;101 let track = broadcast102 .subscribe_track(&Track::new(document.goq.video.track.name))103 .map_err(|error| {104 format!("Failed to subscribe to catalog-selected Goq video track: {error}")105 })?;106 Ok(MoqCatalogSelection {107 track,108 mode: MoqCatalogMode::GoqV1,109 })110}111112113mod tests {114 use super::*;115 use moq_net::Broadcast;116117 fn publish_catalog(118 broadcast: &mut moq_net::BroadcastProducer,119 document: &GoqCatalogDocument,120 ) -> moq_net::TrackProducer {121 let mut track = broadcast122 .create_track(hang::Catalog::default_track())123 .unwrap();124 track125 .write_frame(serde_json::to_vec(document).unwrap())126 .unwrap();127 track128 }129130 131 async fn catalog_selects_the_exact_goq_video_track() {132 let mut broadcast = Broadcast::new().produce();133 let _video = broadcast134 .create_track(Track::new(sigil_protocol::MOQ_VIDEO_H264_TRACK))135 .unwrap();136 let _catalog = publish_catalog(&mut broadcast, &GoqCatalogDocument::video_h264());137 let selection = subscribe_goq_video_track(&broadcast.consume(), Duration::from_millis(100))138 .await139 .unwrap();140 assert_eq!(selection.mode, MoqCatalogMode::GoqV1);141 }142143 144 async fn only_an_absent_catalog_uses_the_legacy_static_track() {145 let mut broadcast = Broadcast::new().produce();146 let _video = broadcast147 .create_track(Track::new(sigil_protocol::MOQ_VIDEO_H264_TRACK))148 .unwrap();149 let selection = subscribe_goq_video_track(&broadcast.consume(), Duration::from_millis(100))150 .await151 .unwrap();152 assert_eq!(selection.mode, MoqCatalogMode::AbsentStaticTrackCompat);153 }154155 156 async fn asynchronous_remote_not_found_uses_static_track_compatibility() {157 let mut broadcast = Broadcast::new().produce();158 let _video = broadcast159 .create_track(Track::new(sigil_protocol::MOQ_VIDEO_H264_TRACK))160 .unwrap();161 let mut dynamic = broadcast.dynamic();162 let consumer = broadcast.consume();163164 let select = subscribe_goq_video_track(&consumer, Duration::from_millis(100));165 let reject = async {166 let mut requested = dynamic.requested_track().await.unwrap();167 assert_eq!(requested.name, hang::Catalog::default_track().name);168 requested169 .abort(MoqError::Remote(MoqError::NotFound.to_code()))170 .unwrap();171 };172 let (selection, ()) = tokio::join!(select, reject);173 assert_eq!(174 selection.unwrap().mode,175 MoqCatalogMode::AbsentStaticTrackCompat176 );177 }178179 180 async fn asynchronous_non_not_found_error_fails_closed() {181 let mut broadcast = Broadcast::new().produce();182 let _video = broadcast183 .create_track(Track::new(sigil_protocol::MOQ_VIDEO_H264_TRACK))184 .unwrap();185 let mut dynamic = broadcast.dynamic();186 let consumer = broadcast.consume();187188 let select = subscribe_goq_video_track(&consumer, Duration::from_millis(100));189 let reject = async {190 let mut requested = dynamic.requested_track().await.unwrap();191 requested192 .abort(MoqError::Remote(MoqError::WrongSize.to_code()))193 .unwrap();194 };195 let (selection, ()) = tokio::join!(select, reject);196 assert!(197 selection198 .err()199 .is_some_and(|error| error.contains("remote error: code=14"))200 );201 }202203 204 async fn present_invalid_or_stalled_catalogs_fail_closed() {205 let mut invalid_broadcast = Broadcast::new().produce();206 let _video = invalid_broadcast207 .create_track(Track::new(sigil_protocol::MOQ_VIDEO_H264_TRACK))208 .unwrap();209 let mut invalid = GoqCatalogDocument::video_h264();210 invalid.goq.video.object_format = "hang/legacy".into();211 let _catalog = publish_catalog(&mut invalid_broadcast, &invalid);212 let result =213 subscribe_goq_video_track(&invalid_broadcast.consume(), Duration::from_millis(100))214 .await;215 assert!(216 result217 .err()218 .is_some_and(|error| error.contains("Invalid Goq catalog"))219 );220221 let mut stalled_broadcast = Broadcast::new().produce();222 let _catalog = stalled_broadcast223 .create_track(hang::Catalog::default_track())224 .unwrap();225 let result =226 subscribe_goq_video_track(&stalled_broadcast.consume(), Duration::from_millis(10))227 .await;228 assert!(229 result230 .err()231 .is_some_and(|error| error.contains("Timed out"))232 );233234 let mut oversized_broadcast = Broadcast::new().produce();235 let _video = oversized_broadcast236 .create_track(Track::new(sigil_protocol::MOQ_VIDEO_H264_TRACK))237 .unwrap();238 let mut catalog = oversized_broadcast239 .create_track(hang::Catalog::default_track())240 .unwrap();241 catalog242 .write_frame(vec![b' '; MAX_MOQ_CATALOG_BYTES + 1])243 .unwrap();244 let result =245 subscribe_goq_video_track(&oversized_broadcast.consume(), Duration::from_millis(100))246 .await;247 assert!(248 result249 .err()250 .is_some_and(|error| error.contains("exceeds 4096 bytes"))251 );252253 let mut multi_frame_broadcast = Broadcast::new().produce();254 let _video = multi_frame_broadcast255 .create_track(Track::new(sigil_protocol::MOQ_VIDEO_H264_TRACK))256 .unwrap();257 let mut catalog = multi_frame_broadcast258 .create_track(hang::Catalog::default_track())259 .unwrap();260 let mut group = catalog.append_group().unwrap();261 group262 .write_frame(serde_json::to_vec(&GoqCatalogDocument::video_h264()).unwrap())263 .unwrap();264 group.write_frame(b"{}".as_slice()).unwrap();265 group.finish().unwrap();266 let result =267 subscribe_goq_video_track(&multi_frame_broadcast.consume(), Duration::from_millis(100))268 .await;269 assert!(270 result271 .err()272 .is_some_and(|error| error.contains("exactly one frame"))273 );274 }275}