diff --git a/rs/hang/examples/subscribe.rs b/rs/hang/examples/subscribe.rs index 628aaa90a..977d1ca2a 100644 --- a/rs/hang/examples/subscribe.rs +++ b/rs/hang/examples/subscribe.rs @@ -2,6 +2,8 @@ use std::time::Duration; +use anyhow::Context; + #[tokio::main] async fn main() -> anyhow::Result<()> { // Optional: Use moq_native to configure a logger. @@ -40,13 +42,11 @@ async fn run_session(origin: moq_net::OriginProducer) -> anyhow::Result<()> { // Subscribe to a broadcast and read media frames. async fn run_subscribe(consumer: moq_net::OriginConsumer) -> anyhow::Result<()> { // Wait for a broadcast to be announced. - let (path, broadcast) = consumer - .announced() - .next() - .await - .ok_or_else(|| anyhow::anyhow!("origin closed"))?; + let (path, broadcast) = consumer.announced().next().await.context("origin closed")?; - let broadcast = broadcast.ok_or_else(|| anyhow::anyhow!("broadcast unannounced: {path}"))?; + let broadcast = broadcast + .broadcast() + .with_context(|| format!("broadcast unannounced: {path}"))?; tracing::info!(%path, "broadcast announced"); diff --git a/rs/libmoq/src/origin.rs b/rs/libmoq/src/origin.rs index 68f076d08..350f5fb25 100644 --- a/rs/libmoq/src/origin.rs +++ b/rs/libmoq/src/origin.rs @@ -76,7 +76,7 @@ impl Origin { ) -> Result<(), Error> { loop { // `biased` so a pending close always wins over a ready announcement. - let (path, broadcast) = tokio::select! { + let (path, event) = tokio::select! { biased; _ = &mut close => return Ok(()), next = consumer.next() => match next { @@ -86,10 +86,11 @@ impl Origin { }; // Hold the lock only to buffer the announcement; release it before the callback. + // Active and Restart both carry a broadcast; Ended does not. let announced_id = State::lock() .origin .announced - .insert((path.to_string(), broadcast.is_some()))?; + .insert((path.to_string(), event.broadcast().is_some()))?; callback.call(announced_id); } } diff --git a/rs/moq-boy/src/input.rs b/rs/moq-boy/src/input.rs index 08b6816c5..d4635fdda 100644 --- a/rs/moq-boy/src/input.rs +++ b/rs/moq-boy/src/input.rs @@ -71,7 +71,7 @@ pub async fn handle_viewers( let viewer_id = path.to_string(); - if let Some(broadcast) = broadcast { + if let Some(broadcast) = broadcast.broadcast() { tracing::info!(%viewer_id, "viewer connected"); let cmd_tx = cmd_tx.clone(); let vid = viewer_id.clone(); diff --git a/rs/moq-ffi/src/origin.rs b/rs/moq-ffi/src/origin.rs index 32785f694..8a51309d1 100644 --- a/rs/moq-ffi/src/origin.rs +++ b/rs/moq-ffi/src/origin.rs @@ -28,14 +28,16 @@ impl Announced { async fn next(&mut self) -> Result>, MoqError> { loop { match self.inner.next().await { - Some((path, Some(broadcast))) => { + // Active and Restart both carry a broadcast; skip unannounce events. + Some((path, event)) => { + let Some(broadcast) = event.broadcast() else { + continue; + }; return Ok(Some(Arc::new(MoqAnnouncement { path: path.to_string(), broadcast: Arc::new(MoqBroadcastConsumer::new(broadcast)), }))); } - // TODO moq-lite will change to not emit None (unannounce) events here. - Some((_path, None)) => continue, None => return Ok(None), } } @@ -44,11 +46,11 @@ impl Announced { async fn available(&mut self) -> Result, MoqError> { loop { match self.inner.next().await { - Some((_path, Some(broadcast))) => { - return Ok(Arc::new(MoqBroadcastConsumer::new(broadcast))); - } - // TODO moq-lite will change to not emit None (unannounce) events here. - Some((_path, None)) => continue, + // Active and Restart both carry a broadcast; skip unannounce events. + Some((_path, event)) => match event.broadcast() { + Some(broadcast) => return Ok(Arc::new(MoqBroadcastConsumer::new(broadcast))), + None => continue, + }, None => return Err(MoqError::Closed), } } diff --git a/rs/moq-native/examples/clock.rs b/rs/moq-native/examples/clock.rs index d8ff48736..4ea556f1b 100644 --- a/rs/moq-native/examples/clock.rs +++ b/rs/moq-native/examples/clock.rs @@ -101,13 +101,13 @@ async fn main() -> anyhow::Result<()> { loop { tokio::select! { - Some(announce) = origin.next() => match announce { - (path, Some(broadcast)) => { + Some((path, event)) = origin.next() => match event.broadcast() { + Some(broadcast) => { tracing::info!(broadcast = %path, "broadcast is online, subscribing to track"); let track = broadcast.subscribe_track(&track)?; clock = Some(Subscriber::new(track)); } - (path, None) => { + None => { tracing::warn!(broadcast = %path, "broadcast is offline, waiting..."); } }, diff --git a/rs/moq-native/tests/backend.rs b/rs/moq-native/tests/backend.rs index 2b2e3796a..7f2ffd1ab 100644 --- a/rs/moq-native/tests/backend.rs +++ b/rs/moq-native/tests/backend.rs @@ -67,7 +67,7 @@ async fn backend_test(scheme: &str, backend: moq_native::QuicBackend) { .expect("origin closed"); assert_eq!(path.as_str(), "test"); - let bc = bc.expect("expected announce, got unannounce"); + let bc = bc.broadcast().expect("expected announce, got unannounce"); let mut track_sub = bc .subscribe_track(&Track::new("video")) @@ -219,7 +219,7 @@ async fn iroh_connect() { .expect("origin closed"); assert_eq!(path.as_str(), "test"); - let bc = bc.expect("expected announce, got unannounce"); + let bc = bc.broadcast().expect("expected announce, got unannounce"); let mut track_sub = bc .subscribe_track(&Track::new("video")) diff --git a/rs/moq-native/tests/broadcast.rs b/rs/moq-native/tests/broadcast.rs index 09bf5da14..f3a800023 100644 --- a/rs/moq-native/tests/broadcast.rs +++ b/rs/moq-native/tests/broadcast.rs @@ -84,7 +84,7 @@ async fn broadcast_test(scheme: &str, client_version: Option<&str>, server_versi .expect("origin closed"); assert_eq!(path.as_str(), "test"); - let bc = bc.expect("expected announce, got unannounce"); + let bc = bc.broadcast().expect("expected announce, got unannounce"); // Subscribe to the track. let mut track_sub = bc @@ -496,7 +496,7 @@ async fn broadcast_websocket() { .expect("origin closed"); assert_eq!(path.as_str(), "test"); - let bc = bc.expect("expected announce, got unannounce"); + let bc = bc.broadcast().expect("expected announce, got unannounce"); // Subscribe to the track. let mut track_sub = bc @@ -603,7 +603,7 @@ async fn broadcast_websocket_fallback() { .expect("origin closed"); assert_eq!(path.as_str(), "test"); - let bc = bc.expect("expected announce, got unannounce"); + let bc = bc.broadcast().expect("expected announce, got unannounce"); // Subscribe to the track. let mut track_sub = bc @@ -844,7 +844,7 @@ async fn linger_resubscribe_keeps_flowing_moq_lite_03() { .expect("announce timeout") .expect("origin closed"); assert_eq!(path.as_str(), "test"); - let bc = bc.expect("expected announce"); + let bc = bc.broadcast().expect("expected announce"); // First subscription: receive group 0. let mut sub1 = bc.subscribe_track(&Track::new("video")).expect("subscribe1"); diff --git a/rs/moq-net/src/ietf/publisher.rs b/rs/moq-net/src/ietf/publisher.rs index 2f390bf7e..e59fc1e21 100644 --- a/rs/moq-net/src/ietf/publisher.rs +++ b/rs/moq-net/src/ietf/publisher.rs @@ -448,79 +448,94 @@ impl Publisher { next = announced.next() => next, }; - let Some((path, active)) = next else { + let Some((path, event)) = next else { break; }; let suffix = path.to_owned(); - if active.is_some() { - tracing::debug!(broadcast = %self.origin.absolute(&path), "announce"); - - let request_id = self.control.next_request_id().await?; - let mut stream = Stream::open(&self.session, self.version).await?; - - // Write the PublishNamespace message - stream.writer.encode(&ietf::PublishNamespace::ID).await?; - stream - .writer - .encode(&ietf::PublishNamespace { - request_id, - track_namespace: suffix.as_path(), - }) - .await?; - - // Read response from stream.reader - let type_id: u64 = stream.reader.decode().await?; - let size: u16 = stream.reader.decode().await?; - let mut data = stream.reader.read_exact(size as usize).await?; - - match (self.version, type_id) { - // Draft14 uses PublishNamespaceOk (0x07) / PublishNamespaceError (0x08) - (Version::Draft14, ietf::PublishNamespaceOk::ID) => { - let msg = ietf::PublishNamespaceOk::decode_msg(&mut data, self.version)?; - tracing::debug!(message = ?msg, "publish namespace ok"); - namespace_streams.insert(suffix, (request_id, stream)); - } - (Version::Draft14, ietf::PublishNamespaceError::ID) => { - let msg = ietf::PublishNamespaceError::decode_msg(&mut data, self.version)?; - tracing::warn!(message = ?msg, "publish namespace error"); - } - // Draft15+ uses RequestOk (0x07) / RequestError (0x05) - (_, ietf::RequestOk::ID) => { - let msg = ietf::RequestOk::decode_msg(&mut data, self.version)?; - tracing::debug!(message = ?msg, "publish namespace ok"); - namespace_streams.insert(suffix, (request_id, stream)); - } - (_, ietf::RequestError::ID) => { - let msg = ietf::RequestError::decode_msg(&mut data, self.version)?; - tracing::warn!(message = ?msg, "publish namespace error"); - } - _ => return Err(Error::UnexpectedMessage), + match event { + crate::Announced::Active(_) => { + self.announce_namespace(suffix, &mut namespace_streams).await?; } - } else { - tracing::debug!(broadcast = %self.origin.absolute(&path), "unannounce"); - if let Some((request_id, mut stream)) = namespace_streams.remove(&suffix) { - // v14-16 sends PublishNamespaceDone; v17+ just closes the stream. - match self.version { - Version::Draft14 | Version::Draft15 | Version::Draft16 => { - let _ = stream - .writer - .encode_message(&ietf::PublishNamespaceDone { - track_namespace: suffix.as_path(), - request_id, - }) - .await; - } - _ => {} - } - stream.writer.finish().ok(); + crate::Announced::Restart(_) => { + // moq-transport has no RESTART; fall back to done + re-announce. + self.unannounce_namespace(&suffix, &mut namespace_streams).await; + self.announce_namespace(suffix, &mut namespace_streams).await?; + } + crate::Announced::Ended => { + self.unannounce_namespace(&suffix, &mut namespace_streams).await; } } } // Clean up remaining streams - for (suffix, (request_id, mut stream)) in namespace_streams { + let suffixes: Vec = namespace_streams.keys().cloned().collect(); + for suffix in suffixes { + self.unannounce_namespace(&suffix, &mut namespace_streams).await; + } + + Ok(()) + } + + /// Open a bidi stream and send a PublishNamespace, recording the stream for later teardown. + async fn announce_namespace( + &self, + suffix: crate::PathOwned, + namespace_streams: &mut HashMap)>, + ) -> Result<(), Error> { + tracing::debug!(broadcast = %self.origin.absolute(&suffix), "announce"); + + let request_id = self.control.next_request_id().await?; + let mut stream = Stream::open(&self.session, self.version).await?; + + stream.writer.encode(&ietf::PublishNamespace::ID).await?; + stream + .writer + .encode(&ietf::PublishNamespace { + request_id, + track_namespace: suffix.as_path(), + }) + .await?; + + let type_id: u64 = stream.reader.decode().await?; + let size: u16 = stream.reader.decode().await?; + let mut data = stream.reader.read_exact(size as usize).await?; + + match (self.version, type_id) { + (Version::Draft14, ietf::PublishNamespaceOk::ID) => { + let msg = ietf::PublishNamespaceOk::decode_msg(&mut data, self.version)?; + tracing::debug!(message = ?msg, "publish namespace ok"); + namespace_streams.insert(suffix, (request_id, stream)); + } + (Version::Draft14, ietf::PublishNamespaceError::ID) => { + let msg = ietf::PublishNamespaceError::decode_msg(&mut data, self.version)?; + tracing::warn!(message = ?msg, "publish namespace error"); + } + (_, ietf::RequestOk::ID) => { + let msg = ietf::RequestOk::decode_msg(&mut data, self.version)?; + tracing::debug!(message = ?msg, "publish namespace ok"); + namespace_streams.insert(suffix, (request_id, stream)); + } + (_, ietf::RequestError::ID) => { + let msg = ietf::RequestError::decode_msg(&mut data, self.version)?; + tracing::warn!(message = ?msg, "publish namespace error"); + } + _ => return Err(Error::UnexpectedMessage), + } + + Ok(()) + } + + /// Tear down the namespace stream for a suffix, sending PublishNamespaceDone where required. + async fn unannounce_namespace( + &self, + suffix: &crate::PathOwned, + namespace_streams: &mut HashMap)>, + ) { + tracing::debug!(broadcast = %self.origin.absolute(suffix), "unannounce"); + if let Some((request_id, mut stream)) = namespace_streams.remove(suffix) { + // v14-16 sends PublishNamespaceDone; v17+ just closes the stream. match self.version { Version::Draft14 | Version::Draft15 | Version::Draft16 => { let _ = stream @@ -535,8 +550,6 @@ impl Publisher { } stream.writer.finish().ok(); } - - Ok(()) } /// Handle a SUBSCRIBE_NAMESPACE on its bidi stream. @@ -589,17 +602,17 @@ impl Publisher { let mut announced = origin.announced(); // Send initial NAMESPACE messages for currently active namespaces - while let Some((path, active)) = announced.try_next() { - let suffix = path.strip_prefix(&prefix).expect("origin returned invalid path"); - if active.is_some() { + // Send initial NAMESPACE messages for currently active namespaces. + // A restart is indistinguishable from an active here (the namespace is present). + while let Some((path, event)) = announced.try_next() { + if event.broadcast().is_some() { + let suffix = path + .strip_prefix(&prefix) + .expect("origin returned invalid path") + .to_owned(); tracing::debug!(broadcast = %origin.absolute(&path), "namespace"); stream.writer.encode(&ietf::Namespace::ID).await?; - stream - .writer - .encode(&ietf::Namespace { - suffix: suffix.to_owned(), - }) - .await?; + stream.writer.encode(&ietf::Namespace { suffix }).await?; } } @@ -608,27 +621,38 @@ impl Publisher { tokio::select! { biased; res = stream.reader.closed() => return res, - next = announced.next() => { - match next { - Some((path, active)) => { - let suffix = path.strip_prefix(&prefix).expect("origin returned invalid path").to_owned(); - if active.is_some() { - tracing::debug!(broadcast = %origin.absolute(&path), "namespace"); - stream.writer.encode(&ietf::Namespace::ID).await?; - stream.writer.encode(&ietf::Namespace { suffix }).await?; - } else { - tracing::debug!(broadcast = %origin.absolute(&path), "namespace_done"); - stream.writer.encode(&ietf::NamespaceDone::ID).await?; - stream.writer.encode(&ietf::NamespaceDone { suffix }).await?; - } - } - None => { - stream.writer.finish()?; - return stream.writer.closed().await; - } + next = announced.next() => { + let Some((path, event)) = next else { + stream.writer.finish()?; + return stream.writer.closed().await; + }; + + let suffix = path.strip_prefix(&prefix).expect("origin returned invalid path").to_owned(); + let absolute = origin.absolute(&path).to_owned(); + + match event { + crate::Announced::Active(_) => { + tracing::debug!(broadcast = %absolute, "namespace"); + stream.writer.encode(&ietf::Namespace::ID).await?; + stream.writer.encode(&ietf::Namespace { suffix }).await?; + } + crate::Announced::Restart(_) => { + // moq-transport has no RESTART; fall back to namespace_done + namespace. + tracing::debug!(broadcast = %absolute, "namespace_done"); + stream.writer.encode(&ietf::NamespaceDone::ID).await?; + stream.writer.encode(&ietf::NamespaceDone { suffix: suffix.clone() }).await?; + tracing::debug!(broadcast = %absolute, "namespace"); + stream.writer.encode(&ietf::Namespace::ID).await?; + stream.writer.encode(&ietf::Namespace { suffix }).await?; + } + crate::Announced::Ended => { + tracing::debug!(broadcast = %absolute, "namespace_done"); + stream.writer.encode(&ietf::NamespaceDone::ID).await?; + stream.writer.encode(&ietf::NamespaceDone { suffix }).await?; } } } + } } } } diff --git a/rs/moq-net/src/lite/announce.rs b/rs/moq-net/src/lite/announce.rs index f0738d269..e3e2fb568 100644 --- a/rs/moq-net/src/lite/announce.rs +++ b/rs/moq-net/src/lite/announce.rs @@ -4,6 +4,18 @@ use crate::{Origin, OriginList, Path, coding::*}; use super::{Message, Version}; +/// Whether the negotiated version carries restart (REANNOUNCE) semantics: a duplicate ANNOUNCE +/// (and the draft's explicit `restart` status) for an already-announced path. Older versions never +/// defined this, so we neither send nor interpret it there; a restart is sent as an unannounce +/// followed by a fresh announce instead. +pub fn restart_supported(version: Version) -> bool { + // Explicitly list older versions so future versions default to supported. + !matches!( + version, + Version::Lite01 | Version::Lite02 | Version::Lite03 | Version::Lite04 + ) +} + /// Sent by the publisher to announce the availability of a track. /// The payload contains the contents of the wildcard. #[derive(Clone, Debug, PartialEq, Eq)] @@ -43,6 +55,13 @@ impl Message for Announce<'_> { Ok(match status { AnnounceStatus::Active => Self::Active { suffix, hops }, AnnounceStatus::Ended => Self::Ended { suffix, hops }, + // We encode a restart as a duplicate ANNOUNCE (a second `Active`), but on versions that + // support restart we also accept the draft's explicit `restart` status and treat it the + // same. For an already-announced path the subscriber turns it into a restart; for an + // unknown path it's a fresh announce. Older versions never defined this status, so it's + // an invalid value there. + AnnounceStatus::Restart if restart_supported(version) => Self::Active { suffix, hops }, + AnnounceStatus::Restart => return Err(DecodeError::InvalidValue), }) } @@ -110,6 +129,9 @@ impl Message for AnnounceInterest<'_> { enum AnnounceStatus { Ended = 0, Active = 1, + /// The draft's explicit restart status. We never encode it (a restart goes out as a duplicate + /// `Active`), but we accept it on decode for forward/cross-compatibility. + Restart = 2, } impl Decode for AnnounceStatus { @@ -173,3 +195,56 @@ impl Message for AnnounceInit<'_> { Ok(()) } } + +#[cfg(test)] +mod tests { + use super::*; + use bytes::Buf; + + // Forge an ANNOUNCE with the draft's explicit `restart` status (2) for the given version. + fn encode_forged_restart(version: Version) -> bytes::Bytes { + // Encode a normal Active, then flip its status byte (1 -> 2). + let mut buf = bytes::BytesMut::new(); + Announce::Active { + suffix: Path::new("foo/bar"), + hops: OriginList::new(), + } + .encode(&mut buf, version) + .expect("encode"); + + // Layout: <...>. The message is small, so the size is one byte and + // the status byte sits at index 1. + assert_eq!( + buf[1], + u8::from(AnnounceStatus::Active), + "expected an Active status byte" + ); + buf[1] = u8::from(AnnounceStatus::Restart); + buf.freeze() + } + + // On lite-05+ the explicit `restart` status is accepted and surfaced as an `Active` (the + // subscriber turns it into a restart for an already-announced path). + #[test] + fn decodes_explicit_restart_status_as_active_on_lite05() { + let version = Version::Lite05Wip; + let mut slice = encode_forged_restart(version); + let decoded = Announce::decode(&mut slice, version).expect("explicit restart must decode"); + assert!(!slice.has_remaining(), "trailing bytes after decode"); + assert!( + matches!(decoded, Announce::Active { .. }), + "restart should decode as Active" + ); + } + + // Older versions never defined the restart status, so it's an invalid value there. + #[test] + fn rejects_explicit_restart_status_before_lite05() { + let version = Version::Lite04; + let mut slice = encode_forged_restart(version); + assert!( + matches!(Announce::decode(&mut slice, version), Err(DecodeError::InvalidValue)), + "restart status must be rejected before lite-05" + ); + } +} diff --git a/rs/moq-net/src/lite/publisher.rs b/rs/moq-net/src/lite/publisher.rs index b01126b54..2c8648737 100644 --- a/rs/moq-net/src/lite/publisher.rs +++ b/rs/moq-net/src/lite/publisher.rs @@ -202,21 +202,27 @@ impl Publisher { // Send ANNOUNCE_INIT as the first message with all currently active paths // We use `try_next()` to synchronously get the initial updates. - while let Some((path, active)) = announced.try_next() { - let suffix = path.strip_prefix(&prefix).expect("origin returned invalid path"); - - if active.is_some() { - tracing::debug!(broadcast = %origin.absolute(&path), "announce"); - let absolute = origin.absolute(&path).to_owned(); + while let Some((path, event)) = announced.try_next() { + let suffix = path + .strip_prefix(&prefix) + .expect("origin returned invalid path") + .to_owned(); + let absolute = origin.absolute(&path).to_owned(); + + // Lite01/02 only carries the set of active paths, so a restart is + // indistinguishable from an active here. + if event.broadcast().is_some() { + tracing::debug!(broadcast = %absolute, "announce"); let guard = stats.broadcast(&absolute).publisher(); - let prev = stats_guards.insert(absolute, guard); - debug_assert!(prev.is_none(), "origin announced a path that was already active"); - init.push(suffix.to_owned()); + stats_guards.entry(absolute).or_insert(guard); + if !init.contains(&suffix) { + init.push(suffix); + } } else { // A potential race. - tracing::debug!(broadcast = %origin.absolute(&path), "unannounce"); - stats_guards.remove(&origin.absolute(&path).to_owned()); - init.retain(|path| path != &suffix); + tracing::debug!(broadcast = %absolute, "unannounce"); + stats_guards.remove(&absolute); + init.retain(|p| p != &suffix); } } @@ -234,70 +240,92 @@ impl Publisher { biased; res = stream.reader.closed() => return res, next = announced.next() => { - match next { - Some((path, active)) => { - let suffix = path.strip_prefix(&prefix).expect("origin returned invalid path").to_owned(); - - if let Some(active) = active { - // Skip if the peer asked us to exclude announces whose hop chain - // contains their id — they already saw this broadcast upstream. - if exclude_hop != 0 && active.hops.iter().any(|h| h.id == exclude_hop) { - tracing::debug!( - broadcast = %origin.absolute(&path), - %exclude_hop, - "skipping announce per peer's exclude_hop", - ); - continue; - } - // Defense in depth: never echo an announce that already passed - // through us. The subscriber should drop these before they reach - // our origin, but if one slips through, don't propagate the loop. - if active.hops.contains(&self_origin) { - tracing::debug!( - broadcast = %origin.absolute(&path), - "skipping reflected announce", - ); - continue; - } - tracing::debug!(broadcast = %origin.absolute(&path), "announce"); - // Append our origin id to the hops so the next relay can detect loops. - // If the chain is already at MAX_HOPS, skip the announce — this link is - // effectively unreachable and the peer will eventually prune the loop. - let mut hops = active.hops.clone(); - if hops.push(self_origin).is_err() { - tracing::warn!( - broadcast = %origin.absolute(&path), - "dropping announce; hop chain at MAX_HOPS (possible loop)", - ); + let Some((path, event)) = next else { + stream.writer.finish()?; + return stream.writer.closed().await; + }; + + let suffix = path.strip_prefix(&prefix).expect("origin returned invalid path").to_owned(); + let absolute = origin.absolute(&path).to_owned(); + + match event { + crate::Announced::Active(active) => { + let Some(hops) = Self::prepare_active_hops(&active.hops, self_origin, exclude_hop, &absolute) else { continue; - } - let absolute = origin.absolute(&path).to_owned(); + }; + tracing::debug!(broadcast = %absolute, "announce"); let guard = stats.broadcast(&absolute).publisher(); let prev = stats_guards.insert(absolute, guard); debug_assert!(prev.is_none(), "origin announced a path that was already active"); - let msg = lite::Announce::Active { suffix, hops }; - stream.writer.encode(&msg).await?; - } else { - tracing::debug!(broadcast = %origin.absolute(&path), "unannounce"); - stats_guards.remove(&origin.absolute(&path).to_owned()); - // An ended announce doesn't need hops — the receiver matches on path only. - let msg = lite::Announce::Ended { - suffix, - hops: OriginList::new(), - }; - stream.writer.encode(&msg).await?; + stream.writer.encode(&lite::Announce::Active { suffix, hops }).await?; + } + crate::Announced::Restart(active) => { + // On lite-05+ a restart travels as a duplicate ANNOUNCE (a second + // `Active` for an already-announced path). Older versions never defined + // that, so split it into an unannounce followed by a fresh announce. + match Self::prepare_active_hops(&active.hops, self_origin, exclude_hop, &absolute) { + Some(hops) => { + tracing::debug!(broadcast = %absolute, "restart"); + // Continuity: keep the existing stats guard (no close + reopen). + if lite::restart_supported(version) { + stream.writer.encode(&lite::Announce::Active { suffix, hops }).await?; + } else { + stream + .writer + .encode(&lite::Announce::Ended { + suffix: suffix.clone(), + hops: OriginList::new(), + }) + .await?; + stream.writer.encode(&lite::Announce::Active { suffix, hops }).await?; + } + } + None => { + // The replacement loops back to us; from this peer's view the broadcast is gone. + tracing::debug!(broadcast = %absolute, "restart replacement looped; unannouncing"); + stats_guards.remove(&absolute); + stream.writer.encode(&lite::Announce::Ended { suffix, hops: OriginList::new() }).await?; + } + } + } + crate::Announced::Ended => { + tracing::debug!(broadcast = %absolute, "unannounce"); + stats_guards.remove(&absolute); + // An ended announce doesn't need hops; the receiver matches on path only. + stream.writer.encode(&lite::Announce::Ended { suffix, hops: OriginList::new() }).await?; } - }, - None => { - stream.writer.finish()?; - return stream.writer.closed().await; } } - } } } } + /// Decide whether to forward an active announcement and compute the outgoing hop chain. + /// + /// Returns `None` when the announce should be skipped: the peer asked us to exclude it + /// (`exclude_hop`), it already passed through us (reflected loop), or the hop chain is full. + fn prepare_active_hops( + hops: &OriginList, + self_origin: Origin, + exclude_hop: u64, + absolute: &crate::Path, + ) -> Option { + if exclude_hop != 0 && hops.iter().any(|h| h.id == exclude_hop) { + tracing::debug!(broadcast = %absolute, %exclude_hop, "skipping announce per peer's exclude_hop"); + return None; + } + if hops.contains(&self_origin) { + tracing::debug!(broadcast = %absolute, "skipping reflected announce"); + return None; + } + let mut hops = hops.clone(); + if hops.push(self_origin).is_err() { + tracing::warn!(broadcast = %absolute, "dropping announce; hop chain at MAX_HOPS (possible loop)"); + return None; + } + Some(hops) + } + pub async fn recv_subscribe(&mut self, mut stream: Stream) -> Result<(), Error> { let subscribe = stream.reader.decode::().await?; diff --git a/rs/moq-net/src/lite/subscriber.rs b/rs/moq-net/src/lite/subscriber.rs index 94aa9fd9e..d92f766fa 100644 --- a/rs/moq-net/src/lite/subscriber.rs +++ b/rs/moq-net/src/lite/subscriber.rs @@ -194,7 +194,19 @@ impl Subscriber { lite::Announce::Active { suffix, hops } => { let path = prefix.join(&suffix); let abs = self.origin.absolute(&path).to_owned(); - if self.start_announce(path.clone(), hops, &mut producers)? { + if lite::restart_supported(self.version) && producers.contains_key(&path) { + // lite-05+ only: a duplicate ANNOUNCE for an already-announced path is a RESTART; + // atomically replace the broadcast. Older versions fall through to start_announce, + // which rejects the duplicate (Error::Duplicate). + if self.restart_announce(path.clone(), hops, &mut producers)? { + // Continuity: keep the existing stats guard if present. + stats_guards + .entry(abs.clone()) + .or_insert_with(|| self.stats.broadcast(&abs).subscriber()); + } else { + stats_guards.remove(&abs); + } + } else if self.start_announce(path.clone(), hops, &mut producers)? { stats_guards.insert(abs.clone(), self.stats.broadcast(&abs).subscriber()); } } @@ -308,6 +320,46 @@ impl Subscriber { Ok(true) } + /// Handle a RESTART (a duplicate ANNOUNCE): atomically replace the broadcast at `path`. + /// + /// Publishing the replacement before retiring the old producer lets the origin demote the old + /// broadcast to a backup and emit a single restart downstream, rather than an + /// unannounce/announce pair with a visible gap. Returns `Ok(false)` if the replacement was a + /// reflected loop (the broadcast is now gone), `Ok(true)` otherwise. + fn restart_announce( + &mut self, + path: PathOwned, + hops: crate::OriginList, + producers: &mut HashMap, + ) -> Result { + // Reflected loop: the replacement passed through us. Retire the broadcast. + if hops.contains(&self.self_origin) { + tracing::debug!(broadcast = %self.log_path(&path), "dropping reflected restart"); + if let Some(mut old) = producers.remove(&path) { + old.abort(Error::Cancel).ok(); + } + return Ok(false); + } + + tracing::debug!(broadcast = %self.log_path(&path), hops = hops.len(), "restart"); + + let broadcast = Broadcast { hops }.produce(); + let dynamic = broadcast.dynamic(); + + // Publish the replacement first so the origin restarts atomically; the old broadcast is + // demoted to a backup and dropped silently when we abort it below. + self.origin.publish_broadcast(path.clone(), broadcast.consume()); + + let old = producers.insert(path.clone(), broadcast.clone()); + web_async::spawn(self.clone().run_broadcast(path.clone(), dynamic)); + + if let Some(mut old) = old { + old.abort(Error::Cancel).ok(); + } + + Ok(true) + } + async fn run_broadcast(self, path: PathOwned, mut broadcast: BroadcastDynamic) { // Actually start serving subscriptions. loop { diff --git a/rs/moq-net/src/model/origin.rs b/rs/moq-net/src/model/origin.rs index 02db50141..060439f96 100644 --- a/rs/moq-net/src/model/origin.rs +++ b/rs/moq-net/src/model/origin.rs @@ -231,18 +231,21 @@ struct OriginBroadcast { backup: VecDeque, } -/// One coalesced update queued for an `OriginConsumer`. +/// One coalesced update queued for an `AnnounceConsumer`. /// /// At most one entry exists per path, so a slow consumer's pending set is bounded -/// by the number of distinct paths. `UnannounceAnnounce` preserves the -/// signal that the broadcast at a path was replaced (the consumer must see -/// `(path, None)` before `(path, Some(new))`), while a stale -/// `Announce` cancels with a subsequent `unannounce` because the consumer -/// has not yet observed it. +/// by the number of distinct paths. `UnannounceAnnounce` preserves the signal +/// that a broadcast genuinely went away and a different one took its place (the +/// consumer must see [`Announced::Ended`] before [`Announced::Active`]), while a +/// stale `Announce` cancels with a subsequent `unannounce` because the consumer +/// has not yet observed it. `Restart` is the atomic replacement: the broadcast +/// at the path never became unavailable, so it collapses to a single +/// [`Announced::Restart`] delivery. enum PendingUpdate { Announce(BroadcastConsumer), Unannounce, UnannounceAnnounce(BroadcastConsumer), + Restart(BroadcastConsumer), } /// Pending updates keyed by path. `BTreeMap` keeps memory strictly bounded by @@ -263,6 +266,22 @@ impl OriginConsumerState { Some(PendingUpdate::Unannounce | PendingUpdate::UnannounceAnnounce(_)) => { PendingUpdate::UnannounceAnnounce(broadcast) } + // A restart the consumer hasn't drained yet; just swap in the newer broadcast. + Some(PendingUpdate::Restart(_)) => PendingUpdate::Restart(broadcast), + }; + self.pending.insert(path, new); + } + + fn apply_restart(&mut self, path: PathOwned, broadcast: BroadcastConsumer) { + let new = match self.pending.remove(&path) { + // Consumer has already drained the prior active; replace it atomically. + None => PendingUpdate::Restart(broadcast), + // Consumer hasn't seen the original announce yet; keep it a fresh announce. + Some(PendingUpdate::Announce(_)) => PendingUpdate::Announce(broadcast), + // Consumer saw the original active; collapse everything into one restart. + Some(PendingUpdate::Unannounce | PendingUpdate::UnannounceAnnounce(_) | PendingUpdate::Restart(_)) => { + PendingUpdate::Restart(broadcast) + } }; self.pending.insert(path, new); } @@ -274,9 +293,9 @@ impl OriginConsumerState { None | Some(PendingUpdate::Unannounce) => { self.pending.insert(path, PendingUpdate::Unannounce); } - // The embedded announce cancels with this unannounce; the consumer - // still needs the leading unannounce. - Some(PendingUpdate::UnannounceAnnounce(_)) => { + // The embedded/replacement announce cancels with this unannounce; the + // consumer still needs the leading unannounce. + Some(PendingUpdate::UnannounceAnnounce(_) | PendingUpdate::Restart(_)) => { self.pending.insert(path, PendingUpdate::Unannounce); } } @@ -285,16 +304,17 @@ impl OriginConsumerState { /// Take one update to deliver to the consumer, if any. fn take(&mut self) -> Option { let path = self.pending.keys().next()?.clone(); - match self.pending.remove(&path).unwrap() { - PendingUpdate::Announce(broadcast) => Some((path, Some(broadcast))), - PendingUpdate::Unannounce => Some((path, None)), + Some(match self.pending.remove(&path).unwrap() { + PendingUpdate::Announce(broadcast) => (path, Announced::Active(broadcast)), + PendingUpdate::Unannounce => (path, Announced::Ended), PendingUpdate::UnannounceAnnounce(broadcast) => { // Deliver the unannounce now; leave the trailing announce pending so // the next take returns it for the same path. self.pending.insert(path.clone(), PendingUpdate::Announce(broadcast)); - Some((path, None)) + (path, Announced::Ended) } - } + PendingUpdate::Restart(broadcast) => (path, Announced::Restart(broadcast)), + }) } } @@ -314,11 +334,13 @@ impl AnnounceConsumerNotify { .apply_announce(path, broadcast); } - fn reannounce(&self, path: impl AsPath, broadcast: BroadcastConsumer) { + fn restart(&self, path: impl AsPath, broadcast: BroadcastConsumer) { let path = path.as_path().strip_prefix(&self.root).unwrap().to_owned(); - let mut state = self.state.write().ok().expect("consumer closed"); - state.apply_unannounce(path.clone()); - state.apply_announce(path, broadcast); + self.state + .write() + .ok() + .expect("consumer closed") + .apply_restart(path, broadcast); } fn unannounce(&self, path: impl AsPath) { @@ -353,13 +375,13 @@ impl NotifyNode { } } - fn reannounce(&mut self, path: impl AsPath, broadcast: &BroadcastConsumer) { + fn restart(&mut self, path: impl AsPath, broadcast: &BroadcastConsumer) { for consumer in self.consumers.values() { - consumer.reannounce(path.as_path(), broadcast.clone()); + consumer.restart(path.as_path(), broadcast.clone()); } if let Some(parent) = &self.parent { - parent.lock().reannounce(path, broadcast); + parent.lock().restart(path, broadcast); } } @@ -426,7 +448,7 @@ impl OriginNode { // // Drop duplicates (same underlying broadcast delivered via multiple links) so the // backup queue can't accumulate clones of the active entry and trigger redundant - // reannouncements when a peer churns. + // restartments when a peer churns. if existing.active.is_clone(broadcast) || existing.backup.iter().any(|b| b.is_clone(broadcast)) { return; } @@ -436,7 +458,7 @@ impl OriginNode { existing.active = broadcast.clone(); existing.backup.push_back(old); - self.notify.lock().reannounce(full, broadcast); + self.notify.lock().restart(full, broadcast); } else { // Longer path: keep as a backup in case the active one drops. existing.backup.push_back(broadcast.clone()); @@ -529,7 +551,7 @@ impl OriginNode { if let Some(idx) = best { let active = entry.backup.remove(idx).expect("index in range"); entry.active = active; - self.notify.lock().reannounce(full, &entry.active); + self.notify.lock().restart(full, &entry.active); } else { // No more backups, so remove the entry. self.broadcast = None; @@ -626,8 +648,39 @@ impl Default for OriginNodes { } } -/// A broadcast path and its associated consumer, or None if closed. -pub type OriginAnnounce = (PathOwned, Option); +/// A path and what happened to the broadcast there, delivered by [`AnnounceConsumer`]. +pub type OriginAnnounce = (PathOwned, Announced); + +/// What happened to a broadcast at a path. +#[derive(Clone)] +pub enum Announced { + /// A broadcast became available. + Active(BroadcastConsumer), + /// The broadcast was replaced without an interruption in availability (e.g. a + /// relay failover or a shorter hop path arriving). + /// + /// Carries the replacement broadcast. On the wire this is a duplicate ANNOUNCE + /// (an active announcement for a path that is already announced, with no + /// intervening unannounce); there is no distinct status byte. + Restart(BroadcastConsumer), + /// The broadcast is no longer available. + Ended, +} + +impl Announced { + /// The broadcast consumer, or `None` if the broadcast ended. + /// + /// Both [`Active`](Self::Active) and [`Restart`](Self::Restart) carry a + /// broadcast; [`Ended`](Self::Ended) does not. This is the legacy + /// `Option` view for callers that don't distinguish a fresh + /// announce from a restart. + pub fn broadcast(self) -> Option { + match self { + Self::Active(broadcast) | Self::Restart(broadcast) => Some(broadcast), + Self::Ended => None, + } + } +} /// Announces broadcasts to consumers over the network. #[derive(Clone)] @@ -679,7 +732,7 @@ impl OriginProducer { /// If there is already a broadcast with the same path, the new one replaces the active only /// if its hop path is shorter or equal; otherwise it is queued as a backup. /// When the active broadcast closes, the backup with the shortest hop path is promoted and - /// reannounced. Backups that close before being promoted are silently dropped. + /// restartd. Backups that close before being promoted are silently dropped. /// /// Returns false if the broadcast is not allowed to be published. pub fn publish_broadcast(&self, path: impl AsPath, broadcast: BroadcastConsumer) -> bool { @@ -850,10 +903,10 @@ impl OriginConsumer { let mut announced = consumer.announced(); loop { - let (announced_path, broadcast) = announced.next().await?; + let (announced_path, event) = announced.next().await?; // `scope` narrows by prefix, but we only want an exact-path match. if announced_path.as_path() == path { - if let Some(broadcast) = broadcast { + if let Some(broadcast) = event.broadcast() { return Some(broadcast); } } @@ -1031,23 +1084,41 @@ use futures::FutureExt; impl AnnounceConsumer { pub fn assert_next(&mut self, expected: impl AsPath, broadcast: &BroadcastConsumer) { let expected = expected.as_path(); - let (path, active) = self.next().now_or_never().expect("next blocked").expect("no next"); + let (path, event) = self.next().now_or_never().expect("next blocked").expect("no next"); + assert!(matches!(event, Announced::Active(_)), "should be an active announce"); + assert_eq!(path, expected, "wrong path"); + assert!( + event.broadcast().unwrap().is_clone(broadcast), + "should be the same broadcast" + ); + } + + pub fn assert_next_restart(&mut self, expected: impl AsPath, broadcast: &BroadcastConsumer) { + let expected = expected.as_path(); + let (path, event) = self.next().now_or_never().expect("next blocked").expect("no next"); + assert!(matches!(event, Announced::Restart(_)), "should be a restart"); assert_eq!(path, expected, "wrong path"); - assert!(active.unwrap().is_clone(broadcast), "should be the same broadcast"); + assert!( + event.broadcast().unwrap().is_clone(broadcast), + "should be the same broadcast" + ); } pub fn assert_try_next(&mut self, expected: impl AsPath, broadcast: &BroadcastConsumer) { let expected = expected.as_path(); - let (path, active) = self.try_next().expect("no next"); + let (path, event) = self.try_next().expect("no next"); assert_eq!(path, expected, "wrong path"); - assert!(active.unwrap().is_clone(broadcast), "should be the same broadcast"); + assert!( + event.broadcast().unwrap().is_clone(broadcast), + "should be the same broadcast" + ); } pub fn assert_next_none(&mut self, expected: impl AsPath) { let expected = expected.as_path(); - let (path, active) = self.next().now_or_never().expect("next blocked").expect("no next"); + let (path, event) = self.next().now_or_never().expect("next blocked").expect("no next"); assert_eq!(path, expected, "wrong path"); - assert!(active.is_none(), "should be unannounced"); + assert!(event.broadcast().is_none(), "should be unannounced"); } pub fn assert_next_wait(&mut self) { @@ -1179,7 +1250,7 @@ mod tests { origin.publish_broadcast("test", consumer3.clone()); assert!(consumer.get_broadcast("test").is_some()); - // On equal hop lengths, each new publish replaces the active and reannounces. + // On equal hop lengths, each new publish replaces the active and restarts. // Because the cursor hasn't drained between publishes, the stale announces // collapse with their following unannounces and only the final active broadcast // is delivered. @@ -1195,15 +1266,14 @@ mod tests { assert!(consumer.get_broadcast("test").is_some()); announced.assert_next_wait(); - // Drop the active, we should reannounce with the remaining backup. + // Drop the active, we should restart with the remaining backup. drop(broadcast3); // Wait for the async task to run. tokio::time::sleep(tokio::time::Duration::from_millis(1)).await; assert!(consumer.get_broadcast("test").is_some()); - announced.assert_next_none("test"); - announced.assert_next("test", &consumer1); + announced.assert_next_restart("test", &consumer1); // Drop the final broadcast, we should unannounce. drop(broadcast1); @@ -1216,6 +1286,39 @@ mod tests { announced.assert_next_wait(); } + #[tokio::test] + async fn test_restart_after_drain() { + // Replacing the active broadcast (equal hops) after the consumer has drained + // the original announce is delivered as a single atomic restart. + let origin = Origin::random().produce(); + let a = Broadcast::new().produce(); + let b = Broadcast::new().produce(); + + let mut announced = origin.consume().announced(); + origin.publish_broadcast("test", a.consume()); + announced.assert_next("test", &a.consume()); + + origin.publish_broadcast("test", b.consume()); + announced.assert_next_restart("test", &b.consume()); + announced.assert_next_wait(); + } + + #[tokio::test] + async fn test_restart_undrained_stays_active() { + // If the consumer hasn't observed the original announce yet, a restart just + // swaps in the newer broadcast and is still delivered as a fresh Active. + let origin = Origin::random().produce(); + let a = Broadcast::new().produce(); + let b = Broadcast::new().produce(); + + let mut announced = origin.consume().announced(); + origin.publish_broadcast("test", a.consume()); + origin.publish_broadcast("test", b.consume()); + + announced.assert_next("test", &b.consume()); + announced.assert_next_wait(); + } + #[tokio::test] async fn test_duplicate_reverse() { tokio::time::pause(); diff --git a/rs/moq-net/src/stats.rs b/rs/moq-net/src/stats.rs index 60d2ba948..fee117c2e 100644 --- a/rs/moq-net/src/stats.rs +++ b/rs/moq-net/src/stats.rs @@ -1063,8 +1063,8 @@ mod tests { let _t2 = bs2.publisher().track("video"); tokio::time::advance(Duration::from_millis(1)).await; - let (path, broadcast) = consumer.next().await.expect("expected announce"); - assert!(broadcast.is_some()); + let (path, event) = consumer.next().await.expect("expected announce"); + assert!(event.broadcast().is_some()); assert_eq!(path.as_str(), ".stats/node/sjc/1"); } @@ -1078,8 +1078,8 @@ mod tests { let _t = bs.publisher().track("video"); tokio::time::advance(Duration::from_millis(1)).await; - let (path, broadcast) = consumer.next().await.expect("expected announce"); - assert!(broadcast.is_some()); + let (path, event) = consumer.next().await.expect("expected announce"); + assert!(event.broadcast().is_some()); assert_eq!(path.as_str(), ".stats/node"); } @@ -1161,8 +1161,8 @@ mod tests { tokio::time::advance(Duration::from_millis(1100)).await; - let (_path, broadcast) = consumer.next().await.expect("expected announce"); - let broadcast = broadcast.expect("active"); + let (_path, event) = consumer.next().await.expect("expected announce"); + let broadcast = event.broadcast().expect("active"); let track = broadcast .subscribe_track(&Track { name: "publisher.json".into(), @@ -1190,8 +1190,8 @@ mod tests { tokio::time::advance(Duration::from_millis(1100)).await; - let (_path, broadcast) = consumer.next().await.expect("announce"); - let broadcast = broadcast.expect("active"); + let (_path, event) = consumer.next().await.expect("announce"); + let broadcast = event.broadcast().expect("active"); let track = broadcast .subscribe_track(&Track { name: "publisher.json".into(), @@ -1224,8 +1224,8 @@ mod tests { tokio::time::advance(Duration::from_millis(1100)).await; - let (_path, broadcast) = consumer.next().await.expect("announce"); - let broadcast = broadcast.expect("active"); + let (_path, event) = consumer.next().await.expect("announce"); + let broadcast = event.broadcast().expect("active"); let track = broadcast .subscribe_track(&Track { name: "publisher.json".into(), @@ -1302,8 +1302,8 @@ mod tests { drive_ticks(2).await; - let (_path, broadcast) = consumer.next().await.expect("announce"); - let broadcast = broadcast.expect("active"); + let (_path, event) = consumer.next().await.expect("announce"); + let broadcast = event.broadcast().expect("active"); // External publisher slot SHOULD include foo/bar. let pub_track = broadcast diff --git a/rs/moq-relay/src/cluster.rs b/rs/moq-relay/src/cluster.rs index 2fae08108..3f2072917 100644 --- a/rs/moq-relay/src/cluster.rs +++ b/rs/moq-relay/src/cluster.rs @@ -20,7 +20,7 @@ const MESH_PREFIX: &str = ".internal/origins"; const SWEEP_INTERVAL: Duration = Duration::from_secs(30); /// How long a peer must stay unannounced before we abort the dial. Must clear the -/// "prefer shorter hop" reannounce flap (which arrives as unannounce-then-announce +/// "prefer shorter hop" restart flap (which arrives as unannounce-then-announce /// within sub-milliseconds) plus reasonable churn from a peer restart. const STALE_AFTER: Duration = Duration::from_secs(60); @@ -76,7 +76,7 @@ impl DialMap { } /// Clear any pending-unannounce on `peer`. Returns `true` if a timestamp was - /// actually cleared (useful for callers that want to log the reannounce). + /// actually cleared (useful for callers that want to log the restart). fn mark_announced(&self, peer: &str) -> bool { let mut map = self.inner.lock().expect("dial map poisoned"); map.get_mut(peer) @@ -343,7 +343,7 @@ impl Cluster { /// announced URL. Unannounces don't abort immediately — they just mark the /// entry as "pending cleanup" with a timestamp. A periodic sweep evicts /// entries whose unannounce has stuck for [`STALE_AFTER`]. The "prefer - /// shorter hop" path in OriginProducer delivers reannouncements as + /// shorter hop" path in OriginProducer delivers restartments as /// unannounce-then-announce within sub-milliseconds, which clears the /// pending-cleanup timestamp long before the sweep fires. async fn run_discovery(self, self_url: String, token: String, dialed: DialMap) { @@ -361,17 +361,17 @@ impl Cluster { loop { tokio::select! { ann = announced.next() => { - let Some((relative, announced)) = ann else { return; }; + let Some((relative, event)) = ann else { return; }; let peer = relative.as_str(); if peer == self_url { continue; } let peer = peer.to_owned(); - match announced { + match event.broadcast() { Some(_) => { if dialed.contains(&peer) { if dialed.mark_announced(&peer) { - tracing::debug!(%peer, "reannounce within sweep window; keeping dial"); + tracing::debug!(%peer, "restart within sweep window; keeping dial"); } continue; } @@ -525,7 +525,7 @@ mod tests { assert!(dialed.contains("healthy:4443")); } - /// A reannounce after an unannounce clears the pending-sweep timestamp, so + /// A restart after an unannounce clears the pending-sweep timestamp, so /// the entry survives even if the original unannounce was old enough to /// otherwise trigger eviction. #[tokio::test] @@ -637,9 +637,9 @@ mod tests { tokio::time::sleep(tokio::time::Duration::from_millis(10)).await; // The self-registration broadcast must be visible on the origin. - let (path, broadcast) = watcher.try_next().expect("self-registration must be published"); + let (path, event) = watcher.try_next().expect("self-registration must be published"); assert_eq!(path.as_str(), ".internal/origins/rendezvous.example.com:4443"); - assert!(broadcast.is_some()); + assert!(event.broadcast().is_some()); // run() must NOT have returned: dropping the broadcast (via run returning) // would unannounce the registration immediately. Use a short timeout to diff --git a/rs/moq-relay/src/web.rs b/rs/moq-relay/src/web.rs index 60c5e3f11..af4258e51 100644 --- a/rs/moq-relay/src/web.rs +++ b/rs/moq-relay/src/web.rs @@ -466,8 +466,8 @@ async fn serve_announced( let mut announced = origin.consume().announced(); let mut broadcasts = Vec::new(); - while let Some((suffix, active)) = announced.try_next() { - if active.is_some() { + while let Some((suffix, event)) = announced.try_next() { + if event.broadcast().is_some() { broadcasts.push(suffix); } } diff --git a/rs/moq-relay/tests/smoke.rs b/rs/moq-relay/tests/smoke.rs index 8788736b1..2c45a9f80 100644 --- a/rs/moq-relay/tests/smoke.rs +++ b/rs/moq-relay/tests/smoke.rs @@ -158,7 +158,7 @@ async fn relay_websocket_round_trip_uses_newest_version() { .expect("origin closed"); // Auth root for `/smoke` is "smoke"; the broadcast "test" announces underneath. assert_eq!(path.as_str(), "test"); - let bc = bc.expect("expected announce, got unannounce"); + let bc = bc.broadcast().expect("expected announce, got unannounce"); let mut track_sub = bc.subscribe_track(&Track::new("video")).expect("subscribe_track"); let mut group_sub = tokio::time::timeout(TIMEOUT, track_sub.recv_group())