diff --git a/src/media.rs b/src/media.rs index bee52c4..1cac996 100644 --- a/src/media.rs +++ b/src/media.rs @@ -35,7 +35,7 @@ pub(crate) enum TranscodeError { /// When `abort` is signalled, the pipeline exits early with /// [`TranscodeError::Aborted`] and the ffmpeg process is killed. pub(crate) async fn transcode_to_fmp4( - stream_url: &str, + stream_url: String, publishers: TrackPublishers, mut abort: tokio::sync::watch::Receiver, group_id: u64, @@ -51,7 +51,7 @@ pub(crate) async fn transcode_to_fmp4( .args([ "-re", "-i", - stream_url, + stream_url.as_str(), "-c:v", "libx264", "-profile:v", diff --git a/src/playback.rs b/src/playback.rs index c95eed1..98b7ec9 100644 --- a/src/playback.rs +++ b/src/playback.rs @@ -54,10 +54,25 @@ impl Player { /// timestamp already set by the caller. /// /// If a pipeline is already running, call [`abort`](Self::abort) first. - pub(crate) fn start(&mut self, info: &ActiveTrackInfo, publishers: TrackPublishers) { + /// + /// If `cached_stream_url` is `Some`, the pipeline skips the source + /// resolution step and uses it directly. + /// Returns the assigned MoQ `group_id` for this pipeline. + #[must_use] + pub(crate) fn start( + &mut self, + info: &ActiveTrackInfo, + publishers: TrackPublishers, + cached_stream_url: Option, + ) -> u64 { + // Clear the cached init segment from the previous pipeline so + // wait_for_init doesn't fire PipelineReady based on stale data. + publishers.clear_init_cache(); + let (abort_tx, abort_rx) = watch::channel(false); self.group_seq += 1; let group_id = self.group_seq; + let pipeline_group_id = group_id; // Copy for return before spawn moves it let track_id = info.id; let source_registry = self.source_registry.clone(); let url = info.url.clone(); @@ -65,29 +80,72 @@ impl Player { let pipeline_tx = self.pipeline_tx.clone(); tokio::spawn(async move { - let stream_url = match source_registry.resolve(&url, &cache_dir).await { - Ok(url) => url, - Err(e) => { - tracing::error!(%url, error = %e, "player: failed to resolve stream"); - send_track_ended(&pipeline_tx, track_id).await; - return; - } + // Resolve stream URL (use cached version if available). + let stream_url = match cached_stream_url { + Some(cached) => cached, + None => match source_registry.resolve(&url, &cache_dir).await { + Ok(url) => url, + Err(e) => { + tracing::error!(%url, error = %e, "player: failed to resolve stream"); + send_command(&pipeline_tx, RoomCommand::TrackEnded { item_id: track_id }) + .await; + return; + } + }, }; - let result = - media::transcode_to_fmp4(&stream_url, publishers, abort_rx, group_id).await; + let init_waiter = publishers.video_init_waiter(); + + // Spawn transcode; runs concurrently and monitors abort_rx internally. + // Pin the JoinHandle so we can race it against init without consuming it. + let transcode = tokio::spawn(media::transcode_to_fmp4( + stream_url, + publishers.clone(), + abort_rx, + group_id, + )); + tokio::pin!(transcode); + + // Race: wait for init (30s timeout) OR transcode to finish early + // (FFmpeg died/aborted before producing any init segment). + // Without this select!, a dead FFmpeg causes a 30s stall. + // Track whether select! consumed transcode to avoid polling + // a completed JoinHandle (which panics). + let (transcode_done, init_ok) = tokio::select! { + biased; + result = tokio::time::timeout( + Duration::from_secs(30), + wait_for_init(init_waiter, &publishers), + ) => (false, result.unwrap_or(false)), + _ = transcode.as_mut() => (true, false), + }; + + if init_ok { + send_command( + &pipeline_tx, + RoomCommand::PipelineReady { item_id: track_id }, + ) + .await; + } else { + tracing::warn!(%url, "player: init segment not published (timeout or transcode failure)"); + } - if let Err(e) = &result { - tracing::warn!(%url, error = %e, "player: transcoding finished with error"); + if transcode_done { + tracing::debug!(%url, "player: transcode already finished (select! consumed it)"); + } else { + // Only await the handle if select! didn't already poll it to completion. + let result = transcode.await.unwrap_or(Ok(())); + if let Err(e) = &result { + tracing::warn!(%url, error = %e, "player: transcoding finished with error"); + } } - // Small delay to give Skip time to clear state first. - tokio::time::sleep(Duration::from_millis(200)).await; - send_track_ended(&pipeline_tx, track_id).await; + send_command(&pipeline_tx, RoomCommand::TrackEnded { item_id: track_id }).await; }); self.track_id = Some(info.id); self.abort_tx = Some(abort_tx); + pipeline_group_id } /// Abort the current pipeline. Idempotent. @@ -114,9 +172,21 @@ impl Player { } } -async fn send_track_ended(tx: &mpsc::Sender, item_id: TrackId) { - if let Err(e) = tx.send(RoomCommand::TrackEnded { item_id }).await { - tracing::warn!(error = %e, "player: failed to send TrackEnded"); +/// Wait for the init segment to be published, either by watching the signal +/// or by polling the cached init. +async fn wait_for_init(mut init_rx: watch::Receiver, publishers: &TrackPublishers) -> bool { + // Fast path: already cached. + if publishers.get_init_segment().is_some() { + return true; + } + // Wait for the signal. + _ = init_rx.changed().await; + publishers.get_init_segment().is_some() +} + +async fn send_command(tx: &mpsc::Sender, cmd: RoomCommand) { + if let Err(e) = tx.send(cmd).await { + tracing::warn!(error = %e, "player: failed to send command"); } } @@ -169,6 +239,7 @@ mod tests { duration: "3:45".into(), thumbnail: None, started_at_wall: 1_700_000_000_000, + group_id: 0, } } @@ -178,11 +249,7 @@ mod tests { let (tx, _rx) = mpsc::channel(256); let mut player = Player::new(test_registry(), PathBuf::from("/tmp"), tx); - player.start(&test_info(item_id), TrackPublishers::new()); - - player.abort(); - - assert!(!player.on_track_ended(TrackId(uuid::Uuid::from_u128(99999)))); + let _ = player.start(&test_info(item_id), TrackPublishers::new(), None); } #[tokio::test] @@ -191,7 +258,7 @@ mod tests { let (tx, _rx) = mpsc::channel(256); let mut player = Player::new(test_registry(), PathBuf::from("/tmp"), tx); - player.start(&test_info(item_id), TrackPublishers::new()); + let _ = player.start(&test_info(item_id), TrackPublishers::new(), None); player.abort(); player.abort(); @@ -201,6 +268,7 @@ mod tests { #[tokio::test] async fn test_player_abort_when_idle() { let (tx, _rx) = mpsc::channel(256); + let mut player = Player::new(test_registry(), PathBuf::from("/tmp"), tx); player.abort(); assert!(!player.on_track_ended(TrackId(uuid::Uuid::from_u128(99999)))); @@ -212,7 +280,7 @@ mod tests { let (tx, _rx) = mpsc::channel(256); let mut player = Player::new(test_registry(), PathBuf::from("/tmp"), tx); - player.start(&test_info(item_id), TrackPublishers::new()); + let _ = player.start(&test_info(item_id), TrackPublishers::new(), None); assert!(player.on_track_ended(item_id)); } @@ -223,7 +291,7 @@ mod tests { let (tx, _rx) = mpsc::channel(256); let mut player = Player::new(test_registry(), PathBuf::from("/tmp"), tx); - player.start(&test_info(item_id), TrackPublishers::new()); + let _ = player.start(&test_info(item_id), TrackPublishers::new(), None); assert!(!player.on_track_ended(TrackId(uuid::Uuid::from_u128(99999)))); } diff --git a/src/room.rs b/src/room.rs index 8e5a4bd..cef3cdb 100644 --- a/src/room.rs +++ b/src/room.rs @@ -59,6 +59,11 @@ struct RoomActor { playlist: Vec, /// Index into the playlist for round-robin auto-fill. playlist_index: usize, + + /// Cache of resolved stream URLs (original URL → direct stream URL). + /// Populated during metadata extraction; avoids re-running yt-dlp -g + /// when the pipeline starts. + stream_cache: HashMap, } /// Thread-safe registry of all active rooms with persistent store. @@ -118,6 +123,7 @@ impl Registry { last_active_tx, playlist: Vec::new(), playlist_index: 0, + stream_cache: HashMap::new(), }; tokio::spawn(actor.run()); @@ -167,6 +173,7 @@ impl Registry { last_active_tx, playlist: snapshot.playlist, playlist_index: 0, + stream_cache: HashMap::new(), }; tokio::spawn(actor.run()); @@ -217,6 +224,7 @@ impl Registry { thumbnail: h.thumbnail, url: h.url, source: "direct".parse().unwrap(), + stream_url: None, }) .collect(); Some((name, history)) @@ -237,6 +245,7 @@ impl Registry { thumbnail: t.thumbnail, url: t.url, source: "direct".parse().unwrap(), // best-effort, source is not tracked in QueuedTrack + stream_url: None, }) .collect() } @@ -471,10 +480,14 @@ impl RoomActor { let sources = self.source_registry.clone(); let cache_dir = self.cache_dir.clone(); tokio::spawn(async move { - match sources.extract(&url, &cache_dir).await { - Ok(meta) => { + match sources.extract_and_resolve(&url, &cache_dir).await { + Ok(result) => { let _ = cmd_tx - .send(RoomCommand::MetadataReady { item_id, meta }) + .send(RoomCommand::MetadataReady { + item_id, + meta: result.meta, + stream_url: result.stream_url, + }) .await; } Err(e) => { @@ -489,8 +502,16 @@ impl RoomActor { }); } - RoomCommand::MetadataReady { item_id, meta } => { + RoomCommand::MetadataReady { + item_id, + meta, + stream_url, + } => { tracing::debug!(room = %self.room_id, %item_id, title = %meta.title, "metadata ready"); + // Cache the resolved stream URL for pipeline startup. + if let Some(url) = stream_url { + self.stream_cache.insert(meta.url.clone(), url); + } let now = Utc::now().timestamp_millis(); @@ -561,6 +582,25 @@ impl RoomActor { self.maybe_auto_fill().await; } + RoomCommand::PipelineReady { item_id } => { + tracing::debug!(room = %self.room_id, %item_id, "pipeline ready"); + // Only react if this matches the current active track. + if !self + .state + .active + .as_ref() + .map(|t| t.id == item_id) + .unwrap_or(false) + { + return false; + } + let now = Utc::now().timestamp_millis(); + self.state.resolve_started_at(now); + // Update the active track info in the stored effect? No — + // just re-publish the snapshot with the new started_at. + self.publish_state_snapshot().await; + } + RoomCommand::TrackEnded { item_id } => { tracing::debug!(room = %self.room_id, %item_id, "track ended"); @@ -596,6 +636,7 @@ impl RoomActor { added_by, added_at: crate::util::now_iso(), pending: true, + stream_url: None, }; let effects = vec![Effect::AddPlaylistEntry { id, @@ -620,10 +661,14 @@ impl RoomActor { let sources = self.source_registry.clone(); let cache_dir = self.cache_dir.clone(); tokio::spawn(async move { - match sources.extract(&url, &cache_dir).await { - Ok(meta) => { + match sources.extract_and_resolve(&url, &cache_dir).await { + Ok(result) => { let _ = cmd_tx - .send(RoomCommand::PlaylistMetadataReady { id, meta }) + .send(RoomCommand::PlaylistMetadataReady { + id, + meta: result.meta, + stream_url: result.stream_url, + }) .await; } Err(e) => { @@ -652,7 +697,15 @@ impl RoomActor { self.publish_state_snapshot().await; } - RoomCommand::PlaylistMetadataReady { id, meta } => { + RoomCommand::PlaylistMetadataReady { + id, + meta, + stream_url, + } => { + // Cache the resolved stream URL for pipeline startup. + if let Some(url) = stream_url { + self.stream_cache.insert(meta.url.clone(), url); + } if let Some(entry) = self.playlist.iter_mut().find(|e| e.id == id) { entry.title = meta.title.clone(); entry.duration = meta.duration.clone(); @@ -756,19 +809,30 @@ impl RoomActor { match effect { Effect::AbortPipeline => { self.player.abort(); + // Clear the cached init segment so the next pipeline + // doesn't fire PipelineReady based on stale data. + self.publishers.clear_init_cache(); } Effect::StartPipeline(track) => { - let now = Utc::now().timestamp_millis(); - self.state.resolve_started_at(now); - let info = ActiveTrackInfo { - id: track.id, - title: track.title, - url: track.url, - duration: track.duration, - thumbnail: track.thumbnail, - started_at_wall: now, + let cached = self.stream_cache.remove(&track.url); + let group_id = { + let info = ActiveTrackInfo { + id: track.id, + title: track.title, + url: track.url, + duration: track.duration, + thumbnail: track.thumbnail, + started_at_wall: 0, + group_id: 0, // set below from player + }; + self.player.start(&info, self.publishers.clone(), cached) }; - self.player.start(&info, self.publishers.clone()); + // Store the real group_id on the active track so the state + // snapshot carries it — clients use it to synchronise video + // data with state updates. + if let Some(ref mut active) = self.state.active { + active.group_id = group_id; + } } Effect::PublishSnapshot => { self.publish_state_snapshot().await; @@ -821,6 +885,7 @@ impl RoomActor { duration: t.duration.clone(), thumbnail: t.thumbnail.clone(), started_at: t.started_at_wall, + group_id: t.group_id, }); let queue: Vec = self @@ -919,6 +984,7 @@ mod tests { thumbnail: None, url: url.into(), source: crate::types::SourceKind::Direct, + stream_url: Some(format!("{url}/stream")), }) } @@ -955,6 +1021,7 @@ mod tests { duration: "3:00".into(), thumbnail: None, started_at_wall: 0, + group_id: 0, }); let playlist = vec![PlaylistEntry { id: pid(1), @@ -966,6 +1033,7 @@ mod tests { added_by: None, added_at: "now".into(), pending: false, + stream_url: None, }]; assert!(next_playlist_track(&state, &playlist, 0).is_none()); } @@ -991,6 +1059,7 @@ mod tests { added_by: None, added_at: "now".into(), pending: false, + stream_url: None, }]; assert!(next_playlist_track(&state, &playlist, 0).is_none()); } @@ -1015,6 +1084,7 @@ mod tests { added_by: None, added_at: "now".into(), pending: false, + stream_url: None, }]; let (entry, index) = next_playlist_track(&state, &playlist, 0).unwrap(); assert_eq!(entry.id, pid(1)); @@ -1035,6 +1105,7 @@ mod tests { added_by: None, added_at: "now".into(), pending: false, + stream_url: None, }, PlaylistEntry { id: pid(2), @@ -1046,6 +1117,7 @@ mod tests { added_by: None, added_at: "now".into(), pending: false, + stream_url: None, }, ]; let (entry, index) = next_playlist_track(&state, &playlist, 1).unwrap(); diff --git a/src/source/direct.rs b/src/source/direct.rs index 842bf2c..325b8b2 100644 --- a/src/source/direct.rs +++ b/src/source/direct.rs @@ -59,6 +59,7 @@ impl MediaSource for DirectSource { thumbnail: None, url: url.to_string(), source: crate::types::SourceKind::Direct, + stream_url: Some(url.to_string()), }) } diff --git a/src/source/mod.rs b/src/source/mod.rs index 20599b9..8fd6502 100644 --- a/src/source/mod.rs +++ b/src/source/mod.rs @@ -55,6 +55,12 @@ pub(crate) trait MediaSource: Send + Sync { async fn resolve(&self, url: &str, cache_dir: &Path) -> Result; } +/// Combined extraction result: metadata plus a pre-resolved stream URL. +pub(crate) struct ExtractAndResolve { + pub meta: TrackMeta, + pub stream_url: Option, +} + /// A registry of media sources tried in registration order. /// /// When a URL arrives, each source is asked via @@ -131,6 +137,20 @@ impl SourceRegistry { "no registered source could resolve URL: {unsupported_hint}" ))) } + + /// Extract metadata and pre-resolve the stream URL in one call. + /// + /// Falls back gracefully: if resolve fails after a successful extract, + /// returns the metadata with `stream_url: None`. + pub(crate) async fn extract_and_resolve( + &self, + url: &str, + cache_dir: &Path, + ) -> Result { + let meta = self.extract(url, cache_dir).await?; + let stream_url = self.resolve(url, cache_dir).await.ok(); + Ok(ExtractAndResolve { meta, stream_url }) + } } #[cfg(test)] @@ -150,6 +170,7 @@ mod tests { thumbnail: None, url: url.into(), source: crate::types::SourceKind::Ytdlp, + stream_url: Some(format!("{url}/stream")), }) } else { Err(SourceError::Unsupported(url.into())) @@ -178,6 +199,7 @@ mod tests { thumbnail: None, url: url.into(), source: crate::types::SourceKind::Direct, + stream_url: None, }) } else { Err(SourceError::Unsupported(url.into())) diff --git a/src/source/ytdlp.rs b/src/source/ytdlp.rs index 5cb76e0..5550939 100644 --- a/src/source/ytdlp.rs +++ b/src/source/ytdlp.rs @@ -89,6 +89,7 @@ impl MediaSource for YtdlpSource { thumbnail: raw.thumbnail, url: raw.webpage_url, source: crate::types::SourceKind::Ytdlp, + stream_url: None, }) } diff --git a/src/state.rs b/src/state.rs index 8729c7e..6863f03 100644 --- a/src/state.rs +++ b/src/state.rs @@ -275,6 +275,7 @@ impl PlaybackState { duration: next.duration.clone(), thumbnail: next.thumbnail.clone(), started_at_wall: 0, + group_id: 0, // placeholders; RoomActor sets real values in execute_effects }; self.active = Some(info); diff --git a/src/store.rs b/src/store.rs index 504c768..c45892d 100644 --- a/src/store.rs +++ b/src/store.rs @@ -30,6 +30,8 @@ pub(crate) struct PlaylistEntry { pub added_by: Option, pub added_at: String, pub pending: bool, + /// Pre-resolved direct stream URL, cached at extraction time. + pub stream_url: Option, } /// Full snapshot of a room's state loaded from the store. @@ -106,6 +108,7 @@ struct InMemoryPlaylistEntry { added_by: Option, added_at: String, pending: bool, + stream_url: Option, } impl From for PlaylistEntry { @@ -120,6 +123,7 @@ impl From for PlaylistEntry { added_by: e.added_by, added_at: e.added_at, pending: e.pending, + stream_url: e.stream_url, } } } @@ -279,6 +283,7 @@ impl RoomStore for InMemoryStore { added_by: None, added_at: crate::util::now_iso(), pending: true, + stream_url: None, }); } Effect::UpdatePlaylistEntry { @@ -452,7 +457,8 @@ impl SqliteStore { source TEXT NOT NULL DEFAULT '', added_by TEXT, added_at TEXT NOT NULL, - pending INTEGER NOT NULL DEFAULT 1 + pending INTEGER NOT NULL DEFAULT 1, + stream_url TEXT )", ) .execute(&pool) @@ -494,6 +500,7 @@ struct PlaylistRow { added_by: Option, added_at: String, pending: i64, + stream_url: Option, } /// Helper row type for deserialising `chat_messages` rows. @@ -751,7 +758,7 @@ impl RoomStore for SqliteStore { // 6. Playlist entries, ordered by id. let playlist_rows: Vec = sqlx::query_as( - "SELECT id, url, title, duration, thumbnail, source, added_by, added_at, pending + "SELECT id, url, title, duration, thumbnail, source, added_by, added_at, pending, stream_url FROM playlist_entries WHERE room_id = ? ORDER BY id", @@ -816,6 +823,7 @@ impl RoomStore for SqliteStore { added_by: r.added_by, added_at: r.added_at, pending: r.pending != 0, + stream_url: r.stream_url, }) .collect(); diff --git a/src/transport.rs b/src/transport.rs index 59fa3d6..db81458 100644 --- a/src/transport.rs +++ b/src/transport.rs @@ -338,6 +338,18 @@ impl TrackPublishers { result } + /// Clear the cached init segment and reset the watch signal. + /// + /// Called when aborting a pipeline so the next pipeline doesn't see a + /// stale init segment from the previous track and fire PipelineReady + /// prematurely. + pub(crate) fn clear_init_cache(&self) { + if let Ok(mut guard) = self.video_init.lock() { + *guard = None; + } + self.video_init_watch.send_replace(false); + } + pub(crate) fn video_init_waiter(&self) -> watch::Receiver { self.video_init_watch.subscribe() } @@ -434,74 +446,6 @@ pub(crate) async fn handle_ws_session( } } -/// WebSocket session for video-only delivery. -/// -/// No MoQ protocol negotiation: the server auto-subscribes the client to -/// the video broadcast. Messages are raw fMP4 bytes (no MoQ framing). -/// A separate WS-control connection handles chat, state, and MoQ handshake. -pub(crate) async fn handle_video_session(ws: WebSocket, publishers: TrackPublishers) { - let (mut ws_tx, mut ws_rx) = ws.split(); - - let init = match publishers.get_init_segment() { - Some(cached) => { - tracing::debug!("video ws: using cached init segment"); - Some(cached) - } - None => { - tracing::debug!("video ws: no cached init, waiting for watch signal"); - let mut init_rx = publishers.video_init_waiter(); - let _ = init_rx.changed().await; - let result = publishers.get_init_segment(); - if result.is_some() { - tracing::debug!("video ws: received init segment via watch"); - } else { - tracing::warn!("video ws: watch fired but no init available"); - } - result - } - }; - - let mut video_rx = publishers.video.subscribe(); - loop { - match video_rx.try_recv() { - Ok(_) => continue, - Err(broadcast::error::TryRecvError::Empty) => break, - Err(broadcast::error::TryRecvError::Closed) => return, - Err(broadcast::error::TryRecvError::Lagged(_)) => break, - } - } - - if let Some((_group_id, payload)) = init { - if ws_tx.send(Message::Binary(payload)).await.is_err() { - return; - } - } - - loop { - tokio::select! { - msg = ws_rx.next() => { - match msg { - Some(Ok(Message::Close(_))) | None => break, - _ => continue, - } - } - result = video_rx.recv() => { - match result { - Ok(obj) => { - if ws_tx.send(Message::Binary(obj.payload)).await.is_err() { - break; - } - } - Err(broadcast::error::RecvError::Lagged(n)) => { - tracing::debug!("video ws: broadcast lagged by {n}"); - } - Err(broadcast::error::RecvError::Closed) => break, - } - } - } - } -} - #[allow(clippy::too_many_arguments)] // justified: all params are distinct concerns async fn handle_moq_message( @@ -585,6 +529,52 @@ async fn handle_moq_message( } })); } + MoqTrackKind::Video => { + let tx = object_tx.clone(); + // Send cached init first (late joiners need it to initialise MSE). + if let Some((group_id, payload)) = publishers.get_init_segment() { + let init_msg = encode_message(&MoqMessage::Object { + track_id: MoqTrackKind::Video as u64, + group_id, + object_id: 0, + payload, + }); + if tx.send(init_msg).is_err() { + return; + } + } + // Drain stale objects from the broadcast buffer so we only + // forward objects published after this subscription. + let mut video_rx = publishers.video.subscribe(); + loop { + match video_rx.try_recv() { + Ok(_) => continue, + Err(broadcast::error::TryRecvError::Empty) => break, + Err(broadcast::error::TryRecvError::Closed) => return, + Err(broadcast::error::TryRecvError::Lagged(_)) => break, + } + } + task_handles.push(tokio::spawn(async move { + loop { + match video_rx.recv().await { + Ok(obj) => { + let msg = encode_message(&MoqMessage::Object { + track_id: obj.track_id as u64, + group_id: obj.group_id, + object_id: obj.object_id, + payload: obj.payload, + }); + if tx.send(msg).is_err() { + break; + } + } + Err(broadcast::error::RecvError::Lagged(_)) => continue, + Err(broadcast::error::RecvError::Closed) => break, + } + } + })); + } + MoqTrackKind::State => { let initial = publishers.state.borrow().clone(); let obj = encode_message(&MoqMessage::Object { @@ -939,4 +929,99 @@ mod tests { assert_eq!(group_id, 2); assert_eq!(data, Bytes::from("init-2")); } + + #[test] + fn test_clear_init_cache() { + let publishers = TrackPublishers::new(); + + // Cache an init segment. + publishers.cache_init_segment(Bytes::from("init-data"), 1); + assert!(publishers.get_init_segment().is_some()); + + // Clear it — should be gone. + publishers.clear_init_cache(); + assert!(publishers.get_init_segment().is_none()); + } + + #[test] + fn test_clear_init_cache_resets_watch() { + let publishers = TrackPublishers::new(); + let waiter = publishers.video_init_waiter(); + + // Initially false. + assert!(!*waiter.borrow()); + + // Cache → watch becomes true. + publishers.cache_init_segment(Bytes::from("init"), 1); + assert!(*waiter.borrow()); + + // Clear → watch resets to false. + publishers.clear_init_cache(); + assert!(!*waiter.borrow()); + + // New waiter sees false (would need to wait for next signal). + let new_waiter = publishers.video_init_waiter(); + assert!(!*new_waiter.borrow()); + } + + #[tokio::test] + async fn test_clear_init_cache_then_cache_again() { + let publishers = TrackPublishers::new(); + + // Cache, then clear. + publishers.cache_init_segment(Bytes::from("old"), 1); + publishers.clear_init_cache(); + assert!(publishers.get_init_segment().is_none()); + + // Now cache again (simulates a new pipeline starting). + let mut waiter = publishers.video_init_waiter(); + assert!(!*waiter.borrow()); + + publishers.cache_init_segment(Bytes::from("new"), 2); + + // Waiter should fire. + let _ = tokio::time::timeout(std::time::Duration::from_secs(2), waiter.changed()) + .await + .expect("watch should fire after re-caching"); + assert!(*waiter.borrow()); + + let retrieved = publishers.get_init_segment().unwrap(); + assert_eq!(retrieved.0, 2); + assert_eq!(retrieved.1, Bytes::from("new")); + } + + #[tokio::test] + async fn test_wait_for_init_waits_after_clear() { + // Simulate the pipeline flow: old init cached, then cleared, then + // new pipeline starts and wait_for_init should NOT immediately return. + let publishers = TrackPublishers::new(); + + // Simulate previous track's init. + publishers.cache_init_segment(Bytes::from("old-init"), 1); + + // Simulate pipeline abort/start: clear cache. + publishers.clear_init_cache(); + + // Now simulate wait_for_init: it should block until new init arrives. + let mut waiter = publishers.video_init_waiter(); + assert!(!*waiter.borrow()); + assert!(publishers.get_init_segment().is_none()); + + // Start a task that caches new init after a delay. + let pubs = publishers.clone(); + tokio::spawn(async move { + tokio::time::sleep(std::time::Duration::from_millis(50)).await; + pubs.cache_init_segment(Bytes::from("new-init"), 2); + }); + + // This is what wait_for_init does: wait for the watch signal. + let _ = tokio::time::timeout(std::time::Duration::from_secs(2), waiter.changed()) + .await + .expect("watch should fire within timeout"); + assert!(*waiter.borrow()); + + let retrieved = publishers.get_init_segment().unwrap(); + assert_eq!(retrieved.0, 2); + assert_eq!(retrieved.1, Bytes::from("new-init")); + } } diff --git a/src/types.rs b/src/types.rs index 917544a..0387c93 100644 --- a/src/types.rs +++ b/src/types.rs @@ -142,6 +142,9 @@ pub(crate) struct TrackMeta { pub url: String, /// How this track was resolved (yt-dlp, direct URL, etc.). pub source: SourceKind, + /// Pre-resolved direct stream URL, if available. + /// Avoids a second yt-dlp invocation when the pipeline starts. + pub stream_url: Option, } /// Snapshot of the currently playing track for state publishing. @@ -153,6 +156,8 @@ pub(crate) struct TrackState { pub(crate) duration: String, pub(crate) thumbnail: Option, pub(crate) started_at: i64, + /// MoQ group_id of the pipeline producing this track. + pub(crate) group_id: u64, } /// Summary of a queued track for state publishing. @@ -237,6 +242,8 @@ pub(crate) enum RoomCommand { MetadataReady { item_id: TrackId, meta: TrackMeta, + /// Resolved stream URL for caching (avoids re-running yt-dlp -g). + stream_url: Option, }, /// Async metadata extraction failed. MetadataFailed { @@ -248,6 +255,11 @@ pub(crate) enum RoomCommand { TrackEnded { item_id: TrackId, }, + /// Sent by the pipeline task when the init segment is published — actual + /// playback has started and the elapsed timer should begin. + PipelineReady { + item_id: TrackId, + }, Chat { user_name: String, content: String, @@ -270,6 +282,8 @@ pub(crate) enum RoomCommand { PlaylistMetadataReady { id: PlaylistEntryId, meta: TrackMeta, + /// Resolved stream URL for caching (avoids re-running yt-dlp -g). + stream_url: Option, }, /// Async metadata extraction failed for a playlist entry. PlaylistMetadataFailed { @@ -290,6 +304,9 @@ pub(crate) struct ActiveTrackInfo { pub(crate) thumbnail: Option, /// Epoch ms when the track started, for client-side elapsed computation. pub(crate) started_at_wall: i64, + /// MoQ group_id of the current pipeline — used by the client to + /// synchronise video data with state updates. + pub(crate) group_id: u64, } /// Public handle to a room: allows sending commands and reading publishers. diff --git a/src/web.rs b/src/web.rs index 9711fab..86e55a9 100644 --- a/src/web.rs +++ b/src/web.rs @@ -82,7 +82,6 @@ pub(crate) fn router(rooms: room::Registry) -> Router { .route("/api/playlist/add", post(add_playlist_entry)) .route("/api/playlist/remove", post(remove_playlist_entry)) .route("/ws/{room_id}", get(ws_handler)) - .route("/ws/{room_id}/video", get(video_ws_handler)) .with_state(state) } @@ -303,28 +302,6 @@ async fn ws_handler( })) } -async fn video_ws_handler( - ws: WebSocketUpgrade, - Path(room_id): Path, - State(state): State, -) -> Result { - let rid = RoomId::parse(&room_id).ok_or((StatusCode::BAD_REQUEST, "invalid room ID"))?; - - if !state.rooms.exists(&rid).await { - return Err((StatusCode::NOT_FOUND, "room not found")); - } - - let handle = state - .rooms - .handle(&rid) - .await - .ok_or((StatusCode::NOT_FOUND, "room not found"))?; - - Ok(ws.on_upgrade(move |socket| async move { - crate::transport::handle_video_session(socket, handle.publishers).await; - })) -} - #[cfg(test)] mod tests { use super::*; diff --git a/templates/partials/room_scripts.html b/templates/partials/room_scripts.html index 2c11a52..934ce2e 100644 --- a/templates/partials/room_scripts.html +++ b/templates/partials/room_scripts.html @@ -72,9 +72,12 @@ let mediaSource = null; let sourceBuffer = null; - const pendingBoxes = []; // buffered while waiting for sourceopen + const pendingBoxes = []; // buffered while awaitingInit let currentTrackId = null; // tracks the currently playing track (from state) - let awaitingInit = true; // true until an init segment is processed + let currentGroupId = -1; // group_id from the latest state snapshot + let processedGroupId = -1; // group_id of the last processed init segment + let awaitingInit = true; // true until an init segment is received + let currentStartedAt = 0; // latest started_at from state, for timer closure let seekInitDone = false; // true after first seek on join or track change function initVideo() { @@ -170,7 +173,51 @@ } } + function handleVideoObject(groupId, data) { + const isInit = + data.length >= 8 && String.fromCharCode(data[4], data[5], data[6], data[7]) === "ftyp"; + + // Discard data from groups before our last processed init — stale old-track. + if (groupId < processedGroupId) { + debug("stale video data, discarding (group", groupId, "<", processedGroupId, ")"); + return; + } + + if (isInit) { + debug("init segment received (group", groupId, ")"); + processedGroupId = groupId; + awaitingInit = false; + pendingBoxes.length = 0; + seekInitDone = false; + + // Reset sourceBuffer for the new track. + if (sourceBuffer && !sourceBuffer.updating) { + try { + sourceBuffer.abort(); + if (sourceBuffer.buffered.length > 0) { + const end = sourceBuffer.buffered.end(sourceBuffer.buffered.length - 1); + sourceBuffer.remove(0, end); + } + } catch (_) {} + } + appendVideo(data); + return; + } + + // Non-init data: buffer until we have an init for this track. + if (awaitingInit) { + pendingBoxes.push(data); + return; + } + + appendVideo(data); + } + function subscribeTracks() { + // Subscribe to video track (MoQ objects carry group_id for sync). + ws.send( + concat([encVarint(0x01), encString("moqbox/room/" + roomId + "/video"), encString("video")]), + ); // Subscribe to chat track. ws.send( concat([encVarint(0x01), encString("moqbox/room/" + roomId + "/chat"), encString("chat")]), @@ -203,7 +250,13 @@ updateConnStatus(true); reconnectDelay = 1000; if (elapsedInterval) { clearInterval(elapsedInterval); elapsedInterval = null; } + // Reset sync state — control WS is the single source of truth. currentTrackId = null; + currentGroupId = -1; + processedGroupId = -1; + awaitingInit = true; + pendingBoxes.length = 0; + seekInitDone = false; subscribeTracks(); }; @@ -230,6 +283,12 @@ updateNowPlaying(state); } catch (_) {} off = buf.length; + } else if (trackId === 1) { + // Video object — use group_id to synchronise with state. + const payload = new Uint8Array(buf.slice(o4)); + // group_id is available from the MoQ framing. + handleVideoObject(groupId, payload); + off = buf.length; } else { off = buf.length; } @@ -259,66 +318,7 @@ connect(); - let videoWs = null; - let videoReconnectDelay = 1000; - function connectVideoWs() { - videoWs = new WebSocket(proto + "//" + location.host + "/ws/" + roomId + "/video"); - videoWs.binaryType = "arraybuffer"; - - videoWs.onopen = () => { - debug("video WS connected"); - videoReconnectDelay = 1000; - awaitingInit = true; - pendingBoxes.length = 0; - seekInitDone = false; - }; - - videoWs.onmessage = (event) => { - const data = new Uint8Array(event.data); - - // Always detect init segments, even when not awaitingInit. - // The video WS and control WS are independent connections with - // no ordering guarantee — init can arrive before the state update. - const isInit = - data.length >= 8 && String.fromCharCode(data[4], data[5], data[6], data[7]) === "ftyp"; - - if (isInit) { - debug("init segment received (video WS)"); - awaitingInit = false; - pendingBoxes.length = 0; - if (sourceBuffer && !sourceBuffer.updating) { - try { - sourceBuffer.abort(); - if (sourceBuffer.buffered.length > 0) { - const end = sourceBuffer.buffered.end(sourceBuffer.buffered.length - 1); - sourceBuffer.remove(0, end); - } - } catch (_) {} - } - appendVideo(data); - return; - } - - if (awaitingInit) { - pendingBoxes.push(data); - return; - } - - appendVideo(data); - }; - - videoWs.onclose = () => { - setTimeout(() => { - videoReconnectDelay = Math.min(videoReconnectDelay * 2, 30000); - connectVideoWs(); - }, videoReconnectDelay); - }; - } - - // Open video connection after a short delay to let the control WS - // establish the room subscription first. - setTimeout(connectVideoWs, 100); function copyRoomLink() { const url = window.location.href; @@ -402,13 +402,17 @@ if (state.current_track) { const trackId = state.current_track.id; + const newGroupId = state.current_track.group_id ?? -1; const trackChanged = currentTrackId !== trackId; - // Detect track change — reset video state. - if (currentTrackId !== null && trackChanged) { + // Detect track change — reset video state if we don't already have + // the init for this track's pipeline (group_id comparison). + if (trackChanged) { debug("track change", currentTrackId, "->", trackId); - awaitingInit = true; - pendingBoxes.length = 0; + if (processedGroupId < newGroupId) { + awaitingInit = true; // need to wait for new init + pendingBoxes.length = 0; + } seekInitDone = false; if (elapsedInterval) { clearInterval(elapsedInterval); @@ -416,29 +420,13 @@ } } currentTrackId = trackId; + currentGroupId = newGroupId; - // Skip overlay DOM work when the track hasn't changed. - // Queue display is still updated below. - if (!trackChanged) { - if (state.current_track) { - const toThumb = document.getElementById("to-thumb"); - const toTitle = document.getElementById("to-title"); - if (toThumb) { - if (state.current_track.thumbnail) { - toThumb.src = state.current_track.thumbnail; - toThumb.style.display = "block"; - } else { - toThumb.style.display = "none"; - } - } - if (toTitle) toTitle.textContent = state.current_track.title; - } - updateQueueDisplay(state); - updatePlaylistDisplay(state); - return; - } - - const startedAt = state.current_track.started_at; + // Always keep the module-level currentStartedAt up to date so the + // elapsed timer (which reads it in its closure) gets the correct value + // even when the second state update arrives (PipelineReady) and + // replaces the placeholder started_at=0 with the real timestamp. + currentStartedAt = state.current_track.started_at || 0; const sourceUrl = state.current_track.url || ""; // Show track overlay, hide idle message @@ -488,33 +476,41 @@ ? " · " + state.clients + " viewer" + (state.clients !== 1 ? "s" : "") : ""; - // Elapsed (initial) - toElapsed.textContent = formatTime((Date.now() - startedAt) / 1000); - - // Ensure overlay is visible on first play - toOverlay.classList.add("show"); - setTimeout(() => toOverlay.classList.remove("show"), 3000); + // Update elapsed display from module-level currentStartedAt (updated + // on every state change — survives the closure lifetime). + toElapsed.textContent = currentStartedAt ? formatTime((Date.now() - currentStartedAt) / 1000) : "0:00"; - // Update elapsed every second from local clock. - elapsedInterval = setInterval(() => { - const elapsed = (Date.now() - startedAt) / 1000; - const elSpan = document.getElementById("to-elapsed"); - if (elSpan) elSpan.textContent = formatTime(elapsed); + // Show overlay briefly on track change + if (trackChanged) { + toOverlay.classList.add("show"); + setTimeout(() => toOverlay.classList.remove("show"), 3000); + } - const fill = document.getElementById("to-progress-fill"); - if (fill && state.current_track) { - const total = parseDuration(state.current_track.duration); - if (total > 0) { - fill.style.width = Math.min(100, (elapsed / total) * 100) + "%"; + // Start the elapsed timer. On track change the timer was cleared + // above. When the second state update (PipelineReady) arrives with a + // real started_at, this condition fires and creates the interval with + // the correct currentStartedAt. + if (!elapsedInterval && currentStartedAt) { + elapsedInterval = setInterval(() => { + if (!currentStartedAt) return; + const elapsed = (Date.now() - currentStartedAt) / 1000; + const elSpan = document.getElementById("to-elapsed"); + if (elSpan) elSpan.textContent = formatTime(elapsed); + const fill = document.getElementById("to-progress-fill"); + if (fill) { + const total = parseDuration(state.current_track.duration); + if (total > 0) { + fill.style.width = Math.min(100, (elapsed / total) * 100) + "%"; + } } - } - }, 1000); + }, 1000); + } // Forward-only sync: if video is more than 3 seconds behind // the live edge, seek forward to catch up. Never seek backward. const video = document.getElementById("video-player"); - if (video && video.buffered.length > 0 && startedAt) { - const target = (Date.now() - startedAt) / 1000; + if (video && video.buffered.length > 0 && currentStartedAt) { + const target = (Date.now() - currentStartedAt) / 1000; if (target > 0) { const behind = target - video.currentTime; if (behind > 3) {