From 10dd1d7556fcbdd963b24ff198fdf6615fafdf0a Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 7 Oct 2026 20:37:52 -0700 Subject: [PATCH 1/2] fix(net): give an anonymous dial its own hop (#5025) (cherry picked from commit 2c0cf672829ffee492248a7b14264f6d6593ab65) Co-authored-by: Grok 4.7 Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/client.rs | 76 +++++---- rs/moq-net/src/ietf/publisher.rs | 8 +- rs/moq-net/src/ietf/subscriber.rs | 8 +- rs/moq-net/src/lite/publisher.rs | 7 +- rs/moq-net/src/lite/subscriber.rs | 12 +- rs/moq-net/tests/dial_split_horizon.rs | 205 +++++++++++++++++++++++++ rs/moq-tokio/src/client.rs | 4 +- 7 files changed, 265 insertions(+), 55 deletions(-) create mode 100644 rs/moq-net/tests/dial_split_horizon.rs diff --git a/rs/moq-net/src/client.rs b/rs/moq-net/src/client.rs index 7574b768ca..e4a6497ca5 100644 --- a/rs/moq-net/src/client.rs +++ b/rs/moq-net/src/client.rs @@ -108,41 +108,47 @@ impl Client { self } - /// Assign an origin (hop) id to the peer, used whenever the peer doesn't declare - /// one itself. + /// Assign the identity this peer's routes are attributed to, overriding the fresh + /// per-dial default. /// - /// Some relays never declare their identity: moq-lite peers without the hops - /// extension, and moq-transport peers that don't negotiate the MoQ Cluster - /// extension (or predate it, on `moqt-16` and earlier). - /// Broadcasts received from such a peer are normally attributed to the reserved - /// Hop ID 0 ("unknown"), which identifies nothing: it never proves continuity, - /// so their advertisements neither splice nor survive a restart in place. This - /// knob pins a real identity instead, exactly as if the peer had declared it: + /// A dialed session whose peer declares no hop gets a random one for that + /// connection, the same way an accepted session does. That keeps a route + /// learned from the peer off the session that learned it, and keeps a + /// SUBSCRIBE that arrives on it from being routed back to it. An identity the + /// peer declares on the wire still wins. /// - /// - broadcasts received from the peer carry `origin` in their hop chains, so - /// every session dialing the same relay (with the same id) resolves to one - /// route and loop checks can recognize it; - /// - broadcasts whose hop chain already contains `origin` are neither announced - /// nor served back to the peer, preventing an echo through a relay that does - /// no loop detection of its own. - /// - /// An identity the peer does declare wins over this one. + /// Pass an id only for a peer whose identity the caller has actually + /// established. Two dials given the same hop are treated as one endpoint: + /// routes learned from either are kept off both, and content arriving on + /// either is interchangeable with the other's. That is the point when they + /// really are one peer reconnecting or running redundant links, and a bug + /// otherwise. Derive it from the authenticated identity, never from something + /// coarser like the remote address. pub fn with_peer_hop(mut self, hop: crate::Hop) -> Self { self.peer_hop = Some(hop); self } + /// The hop this connection attributes the peer to when the peer declares none. + /// + /// A caller-supplied id is stable across dials of this client. Otherwise each + /// connection gets a fresh one, matching the per-session default an accepted + /// session already has. + fn assigned_hop(&self) -> crate::Hop { + self.peer_hop.unwrap_or_else(crate::Hop::random) + } + /// The origin pair a session attaches, tagged and filtered. /// /// Reads through the publish (egress) consumer and writes through the /// subscribe (ingress) producer are attributed by the model through the /// stats context; one shared context, so presence and viewer counts are - /// never double-attributed across the two halves. An assigned peer identity - /// means subscriptions from the peer resolve to a source whose hop chain - /// excludes it, the same split-horizon rule applied when a peer declares - /// its own id; announce filtering is per-protocol and handled inside each - /// publisher. - fn origins(&self) -> (Option, Option) { + /// never double-attributed across the two halves. `peer_hop` is the identity + /// this connection attributes the peer to when the peer declares none, so + /// subscriptions from the peer resolve to a source whose route excludes it. + /// A declared identity still wins inside each publisher; announce filtering + /// is per-protocol and handled there too. + fn origins(&self, peer_hop: crate::Hop) -> (Option, Option) { if self.publish.is_none() && self.subscribe.is_none() { tracing::warn!("not publishing or consuming anything"); } @@ -151,7 +157,7 @@ impl Client { .subscribe .clone() .map(|origin| origin.with_stats(self.stats.clone())); - let publish = publish.map(|origin| origin.excluding(self.peer_hop.unwrap_or(crate::Hop::UNKNOWN))); + let publish = publish.map(|origin| origin.excluding(peer_hop)); (publish, subscribe) } @@ -162,11 +168,12 @@ impl Client { runtime: Clock, session: S, version: lite::Version, + peer_hop: crate::Hop, ) -> Result<(Session, crate::Driver), Error> where S: crate::transport::poll::Session, { - let (publish, subscribe) = self.origins(); + let (publish, subscribe) = self.origins(peer_hop); // Advertise our capabilities (we report what the transport measures; we // don't pad) plus the request path on URI-less transports, and the @@ -192,7 +199,7 @@ impl Client { setup_stream: None, publish, subscribe, - peer_hop: self.peer_hop, + peer_hop: Some(peer_hop), version, our_setup, peer_setup: None, @@ -231,7 +238,7 @@ impl Client { _ => return Err(Error::Version), }; self.versions.select(Version::Lite(version)).ok_or(Error::Version)?; - self.start_lite(runtime, session, version) + self.start_lite(runtime, session, version, self.assigned_hop()) } /// Perform the MoQ handshake, returning the [`Session`] and its [`Driver`](crate::Driver). @@ -242,7 +249,8 @@ impl Client { S: crate::transport::poll::Boxable, { let runtime = Clock::new(now); - let (publish, subscribe) = self.origins(); + let peer_hop = self.assigned_hop(); + let (publish, subscribe) = self.origins(peer_hop); // If ALPN was used to negotiate the version, use the appropriate encoding. // Default to IETF 14 if no ALPN was used and we'll negotiate the version later. @@ -269,7 +277,7 @@ impl Client { client: true, publish: publish.clone(), subscribe: subscribe.clone(), - peer_hop: self.peer_hop, + peer_hop: Some(peer_hop), cost: self.cost, version: draft, path: self.setup_path.clone(), @@ -317,19 +325,19 @@ impl Client { _ => lite::Version::Lite05, }; self.versions.select(Version::Lite(version)).ok_or(Error::Version)?; - return self.start_lite(runtime, session, version); + return self.start_lite(runtime, session, version, peer_hop); } Some(ALPN_LITE_04) => { self.versions .select(Version::Lite(lite::Version::Lite04)) .ok_or(Error::Version)?; - return self.start_lite(runtime, session, lite::Version::Lite04); + return self.start_lite(runtime, session, lite::Version::Lite04, peer_hop); } Some(ALPN_LITE_03) => { self.versions .select(Version::Lite(lite::Version::Lite03)) .ok_or(Error::Version)?; - return self.start_lite(runtime, session, lite::Version::Lite03); + return self.start_lite(runtime, session, lite::Version::Lite03, peer_hop); } Some(ALPN_LITE) | None => { let supported = self.versions.filter(&NEGOTIATED.into()).ok_or(Error::Version)?; @@ -381,7 +389,7 @@ impl Client { setup_stream: Some(stream), publish: publish.clone(), subscribe: subscribe.clone(), - peer_hop: self.peer_hop, + peer_hop: Some(peer_hop), version: v, // This path only handles versions negotiated via the bidi SETUP exchange // (pre-lite-05), which have no Setup Stream. @@ -418,7 +426,7 @@ impl Client { client: true, publish: publish.clone(), subscribe: subscribe.clone(), - peer_hop: self.peer_hop, + peer_hop: Some(peer_hop), cost: self.cost, version: v, path: None, diff --git a/rs/moq-net/src/ietf/publisher.rs b/rs/moq-net/src/ietf/publisher.rs index d42749c8ea..3659fbe8e3 100644 --- a/rs/moq-net/src/ietf/publisher.rs +++ b/rs/moq-net/src/ietf/publisher.rs @@ -313,8 +313,8 @@ pub(super) struct Publisher { // origin we consume so it matches the local relay identity across every session, // which is what makes cross-session loop detection work. self_origin: crate::Hop, - // The identity assigned to the peer by the caller (`Client::with_peer_hop`, or - // the per-session default a server hands every request), used when the peer declares + // The identity assigned to the peer (a fresh per-session id, or one the caller + // pinned with `with_peer_hop`), used when the peer declares // none itself. A peer that negotiates the MoQ Cluster extension declares its own, // which wins unless it withheld it as the reserved 0. peer_hop: Option, @@ -420,8 +420,8 @@ where /// The Hop ID whose paths must not be advertised (or served) back to this peer. /// /// A peer that declared an identity supplies its own; otherwise fall back to the one - /// we assigned it (`Client::with_peer_hop` when dialing, `Request::with_peer_hop` - /// or a fresh per-session id when accepting), since moq-transport carries no identity + /// we assigned it (a fresh per-session id, or one pinned with `Client::with_peer_hop` + /// or `Request::with_peer_hop`), since moq-transport carries no identity /// of its own. A peer that declared the reserved 0 declared no identity, so it takes /// the fallback like any other anonymous peer. fn exclude(&self, peer: &cluster::Peer) -> crate::Hop { diff --git a/rs/moq-net/src/ietf/subscriber.rs b/rs/moq-net/src/ietf/subscriber.rs index 1a3e128b7b..78d04114ad 100644 --- a/rs/moq-net/src/ietf/subscriber.rs +++ b/rs/moq-net/src/ietf/subscriber.rs @@ -467,12 +467,12 @@ pub(super) struct Subscriber { // The origin naming this link for split-horizon (`Route.via`) when the peer // declares none of its own (see `session_route`). Base moq-transport carries no // hop ids, so a peer only has an identity if it negotiated the MoQ Cluster - // extension or the caller assigned it one (`Client::with_peer_hop`). + // extension; otherwise this is the one the session assigned it (a fresh id per + // dialed or accepted session, unless the caller pinned one with `with_peer_hop`), + // or `Hop::UNKNOWN` (0) when none was assigned. // - // Otherwise this is `Hop::UNKNOWN` (0), the reserved "no identity" value. // The assigned id stays local: it is never written into a hop chain, so a peer - // that withheld an identity is not named on the wire. A server answers it per - // accepted session; a client only when it knows the peer. + // that withheld an identity is not named on the wire. session_origin: crate::Hop, // A random Hop ID of this connection's own, written as the first hop of any path // that arrives naming no publisher, so a publisher that reconnects reads downstream diff --git a/rs/moq-net/src/lite/publisher.rs b/rs/moq-net/src/lite/publisher.rs index abc7f01875..4ee625a571 100644 --- a/rs/moq-net/src/lite/publisher.rs +++ b/rs/moq-net/src/lite/publisher.rs @@ -51,8 +51,8 @@ struct Shared { // peer from a source whose chain excludes them, keeping the data plane on // the same split-horizon rule as the announces we send them. peer_setup: super::PeerSetup, - // The identity assigned to the peer by the caller (`Client::with_peer_hop`, or - // the per-session default a server hands every request), standing in wherever the + // The identity assigned to the peer (a fresh per-session id, or one the caller + // pinned with `with_peer_hop`), standing in wherever the // peer declines to declare one. Backs both the announce filter and the serving // origin, so a peer that names itself nowhere on the wire is still split-horizoned. peer_hop: Option, @@ -528,8 +528,7 @@ impl AnnounceServe { // The identity whose routes we filter out. Lite-04/05 carry it per // announce stream; lite-06+ reads the session-wide SETUP Hop // parameter, the same identity the subscribe path excludes. A peer that - // declares nothing falls back to the identity the caller assigned it - // (`with_peer_hop`), if any. + // declares nothing falls back to the identity the session assigned it. let assigned = self.shared.peer_hop.map(|origin| origin.id()).unwrap_or(0); if self.shared.version.has_exclude_hop() { let exclude_hop = match interest.exclude_hop { diff --git a/rs/moq-net/src/lite/subscriber.rs b/rs/moq-net/src/lite/subscriber.rs index 92b01d1423..2d49255bae 100644 --- a/rs/moq-net/src/lite/subscriber.rs +++ b/rs/moq-net/src/lite/subscriber.rs @@ -61,13 +61,11 @@ pub(super) struct Subscriber { // carry real hop ids on the wire (Lite01/02/03), and for a peer that reports // 0 in AnnounceOk. Lite03 placeholders stay 0 and count as anonymous. // - // This is the peer's assigned identity (`peer_hop`) when the caller gave - // it one. Otherwise it is `Hop::UNKNOWN` (0), the reserved "no identity" value. - // - // Assigning one is the caller's call, not this layer's: a server gives every - // accepted session a fresh id so its routes are at least distinguishable from - // another session's, while a client only assigns one it knows out of band. The - // assigned id stays local and is never written into a hop chain. + // This is the peer's assigned identity (`peer_hop`): a fresh id per dialed or + // accepted session, so its routes are distinguishable from another session's, + // unless the caller pinned a stable one with `with_peer_hop`. Without one it is + // `Hop::UNKNOWN` (0), the reserved "no identity" value. The assigned id stays + // local and is never written into a hop chain. session_origin: crate::Hop, // A random Hop ID of this connection's own, written as the first hop of any chain // that names no publisher (Lite01/02, a Lite03 placeholder, or a peer reporting 0), diff --git a/rs/moq-net/tests/dial_split_horizon.rs b/rs/moq-net/tests/dial_split_horizon.rs new file mode 100644 index 0000000000..402d01ba98 --- /dev/null +++ b/rs/moq-net/tests/dial_split_horizon.rs @@ -0,0 +1,205 @@ +//! A dialed moq-transport session whose peer declares no hop must not be offered +//! a route learned from that peer, and a SUBSCRIBE that arrives for that route +//! must not be sent back to the peer that announced it. +//! +//! Draft 16 has no cluster extension, so neither side declares a hop. That is the +//! anonymous dial: the accepting side already stamps one, and the dialing side +//! has to as well. + +mod support; + +use std::time::Duration; + +use moq_net::{Client, Hop, Server, Session, Version, origin, stats}; +use support::{ + harness::{now, spawn}, + mock::create_mock_session_pair, +}; + +/// Long enough, in virtual time, for an announce or subscribe to cross the mock link. +const SETTLE: Duration = Duration::from_secs(1); + +fn version() -> Version { + "moq-transport-16".parse().unwrap() +} + +fn produce(hop: u64) -> origin::Producer { + let (producer, driver) = origin::Producer::new(origin::Config::new(Hop::new(hop).unwrap())); + spawn(driver); + producer +} + +/// One dial from `dialer` to `edge`, held open for the test. +struct Link { + _client: Session, + _server: Session, + /// Stats on the accepted session, so an announce or subscribe that arrives + /// there is visible without reading the wire. + registry: stats::Registry, +} + +async fn dial(dialer: &origin::Producer, edge: &origin::Producer) -> Link { + let version = version(); + let registry = stats::Registry::new(stats::Config::new()); + let session_stats = registry.tier(stats::Tier::default()).session(""); + let (client_transport, server_transport) = create_mock_session_pair(Some(version.alpn())); + + // Cluster dials publish and subscribe through a peer handle, which is what + // records the announcing session on the route. + let dialer_origin = dialer.clone().peer(); + let edge_origin = edge.clone().peer(); + let client = Client::new() + .with_versions(version.into()) + .with_publisher(dialer_origin.consume().with_hidden(true)) + .with_subscriber(dialer_origin); + let server = Server::new() + .with_versions(version.into()) + .with_stats(session_stats) + .with_publisher(edge_origin.consume().with_hidden(true)) + .with_subscriber(edge_origin); + + let client = async { + let (session, driver) = client + .connect(now(), client_transport) + .await + .expect("client handshake failed"); + spawn(driver); + session + }; + let server = async { + let (session, driver) = server + .accept(now(), server_transport) + .await + .expect("server handshake failed"); + spawn(driver); + session + }; + let (client, server) = futures::join!(client, server); + Link { + _client: client, + _server: server, + registry, + } +} + +fn traffic(registry: &stats::Registry, role: stats::Role) -> stats::Traffic { + registry + .snapshot() + .traffic() + .into_iter() + .find(|(_, row, _)| *row == role) + .map(|(_, _, traffic)| traffic) + .unwrap_or_default() +} + +/// The hop the dialer attributed `path` to, once that announce has arrived. +async fn learned_hop(dialer: &origin::Producer, path: &str) -> Hop { + let mut announced = dialer.consume().announced(); + let event = moq_net_sim::timeout(SETTLE, async { + loop { + let event = announced.next().await.expect("announce cursor closed"); + let announce = match event { + moq_net::announce::Event::Start(announce) | moq_net::announce::Event::Update(announce) => announce, + moq_net::announce::Event::End(_) => continue, + }; + if announce.prefix.as_str() == path { + return announce; + } + } + }) + .await + .unwrap_or_else(|_| panic!("{path}: the dialer never learned the namespace")); + + match event.route.source() { + origin::Source::Peer(hop) => hop, + origin::Source::Local => panic!("{path}: a route learned from the peer looks local"), + } +} + +/// Draft-16 edges dialed by one origin each get their own hop. The edge that +/// announced a namespace is not told about it, and the other edge is. A +/// subscribe from that other edge is fetched from the announcer and is not +/// sent back to the subscriber. +#[moq_net_sim::test] +async fn dialed_anonymous_session_does_not_echo_or_route_subscribe_back() { + let dialer = produce(1); + let publisher = produce(2); + let subscriber = produce(3); + let to_publisher = dial(&dialer, &publisher).await; + let to_subscriber = dial(&dialer, &subscriber).await; + + let broadcast = publisher.create_broadcast("room").unwrap(); + let track = broadcast.create_track("video", None).unwrap(); + broadcast.announce(origin::Route::default()).unwrap(); + + let publisher_hop = learned_hop(&dialer, "room").await; + assert_ne!( + publisher_hop, + Hop::UNKNOWN, + "an anonymous dial still needs a hop, or its routes echo" + ); + + // The other edge learns the namespace, attributed to a different hop, so the + // two dials are not one endpoint. + let mut subscriber_announced = subscriber.consume().announced(); + let subscriber_saw = moq_net_sim::timeout(SETTLE, subscriber_announced.next()) + .await + .expect("the other edge never learned the namespace") + .expect("announce cursor closed"); + let moq_net::announce::Event::Start(subscriber_saw) = subscriber_saw else { + panic!("the other edge's first event was not the namespace: {subscriber_saw:?}"); + }; + assert_eq!(subscriber_saw.prefix.as_str(), "room"); + let subscriber_hop = match subscriber_saw.route.source() { + origin::Source::Peer(hop) => hop, + origin::Source::Local => panic!("the forwarded route looks local"), + }; + assert_ne!(subscriber_hop, publisher_hop, "each dial gets its own hop"); + assert_ne!(subscriber_hop, Hop::UNKNOWN); + + moq_net_sim::sleep(SETTLE).await; + + let echoed = traffic(&to_publisher.registry, stats::Role::Subscriber); + assert_eq!( + echoed.announces_started, 0, + "the announcing edge was offered its own namespace back" + ); + let forwarded = traffic(&to_subscriber.registry, stats::Role::Subscriber); + assert!( + forwarded.announces_started >= 1, + "the other edge was not offered the namespace" + ); + + // The subscribe arrives on the dialer's session with the other edge. It has + // to be fetched from the announcer, and not turned into a subscribe on the + // session it arrived on. + let viewed = subscriber + .consume() + .request_broadcast("room") + .await + .expect("subscriber resolves the namespace"); + let subscription = moq_net::track::Subscription::default(); + let _reader = viewed + .track("video") + .unwrap() + .subscribe(subscription) + .await + .expect("subscriber subscribes"); + moq_net_sim::timeout(SETTLE, track.demand().used()) + .await + .expect("the subscribe never reached the announcer") + .unwrap(); + + moq_net_sim::sleep(SETTLE).await; + + let from_announcer = traffic(&to_publisher.registry, stats::Role::Publisher); + assert_eq!( + from_announcer.subscriptions_started, 1, + "the subscribe was not routed to the edge that announced it" + ); + let back_to_subscriber = traffic(&to_subscriber.registry, stats::Role::Publisher); + assert_eq!( + back_to_subscriber.subscriptions_started, 0, + "the subscribe was routed back to the edge that sent it" + ); +} diff --git a/rs/moq-tokio/src/client.rs b/rs/moq-tokio/src/client.rs index d7c517d258..8b97eaf778 100644 --- a/rs/moq-tokio/src/client.rs +++ b/rs/moq-tokio/src/client.rs @@ -220,8 +220,8 @@ impl Client { self } - /// Assign an origin (hop) id to the peers this client dials, used whenever a - /// peer doesn't declare one itself; see [`moq_net::Client::with_peer_hop`]. + /// Pin the identity a dialed peer's routes are attributed to, for a peer whose + /// identity the caller has established; see [`moq_net::Client::with_peer_hop`]. pub fn with_peer_hop(mut self, hop: moq_net::Hop) -> Self { self.moq = self.moq.with_peer_hop(hop); self From 61825510a0a869cb5239c2658e53ffd814f3d8cf Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 9 Oct 2026 14:55:07 -0700 Subject: [PATCH 2/2] test(net): adapt dial_split_horizon to release's test harness Release predates moq_net_sim and announce::Event, so the backported test runs on paused tokio time and reads announce::Update/Kind instead. Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/tests/dial_split_horizon.rs | 42 ++++++++++++-------------- 1 file changed, 20 insertions(+), 22 deletions(-) diff --git a/rs/moq-net/tests/dial_split_horizon.rs b/rs/moq-net/tests/dial_split_horizon.rs index 402d01ba98..914b52a9af 100644 --- a/rs/moq-net/tests/dial_split_horizon.rs +++ b/rs/moq-net/tests/dial_split_horizon.rs @@ -10,9 +10,9 @@ mod support; use std::time::Duration; -use moq_net::{Client, Hop, Server, Session, Version, origin, stats}; +use moq_net::{Client, Hop, Server, Session, Version, announce, origin, stats}; use support::{ - harness::{now, spawn}, + harness::{now, run}, mock::create_mock_session_pair, }; @@ -25,7 +25,7 @@ fn version() -> Version { fn produce(hop: u64) -> origin::Producer { let (producer, driver) = origin::Producer::new(origin::Config::new(Hop::new(hop).unwrap())); - spawn(driver); + tokio::spawn(run(driver)); producer } @@ -63,7 +63,7 @@ async fn dial(dialer: &origin::Producer, edge: &origin::Producer) -> Link { .connect(now(), client_transport) .await .expect("client handshake failed"); - spawn(driver); + tokio::spawn(run(driver)); session }; let server = async { @@ -71,10 +71,10 @@ async fn dial(dialer: &origin::Producer, edge: &origin::Producer) -> Link { .accept(now(), server_transport) .await .expect("server handshake failed"); - spawn(driver); + tokio::spawn(run(driver)); session }; - let (client, server) = futures::join!(client, server); + let (client, server) = tokio::join!(client, server); Link { _client: client, _server: server, @@ -95,15 +95,11 @@ fn traffic(registry: &stats::Registry, role: stats::Role) -> stats::Traffic { /// The hop the dialer attributed `path` to, once that announce has arrived. async fn learned_hop(dialer: &origin::Producer, path: &str) -> Hop { let mut announced = dialer.consume().announced(); - let event = moq_net_sim::timeout(SETTLE, async { + let event = tokio::time::timeout(SETTLE, async { loop { - let event = announced.next().await.expect("announce cursor closed"); - let announce = match event { - moq_net::announce::Event::Start(announce) | moq_net::announce::Event::Update(announce) => announce, - moq_net::announce::Event::End(_) => continue, - }; - if announce.prefix.as_str() == path { - return announce; + let update = announced.next().await.expect("announce cursor closed"); + if update.kind.is_active() && update.prefix.as_str() == path { + return update; } } }) @@ -120,7 +116,7 @@ async fn learned_hop(dialer: &origin::Producer, path: &str) -> Hop { /// announced a namespace is not told about it, and the other edge is. A /// subscribe from that other edge is fetched from the announcer and is not /// sent back to the subscriber. -#[moq_net_sim::test] +#[tokio::test(start_paused = true)] async fn dialed_anonymous_session_does_not_echo_or_route_subscribe_back() { let dialer = produce(1); let publisher = produce(2); @@ -142,13 +138,15 @@ async fn dialed_anonymous_session_does_not_echo_or_route_subscribe_back() { // The other edge learns the namespace, attributed to a different hop, so the // two dials are not one endpoint. let mut subscriber_announced = subscriber.consume().announced(); - let subscriber_saw = moq_net_sim::timeout(SETTLE, subscriber_announced.next()) + let subscriber_saw = tokio::time::timeout(SETTLE, subscriber_announced.next()) .await .expect("the other edge never learned the namespace") .expect("announce cursor closed"); - let moq_net::announce::Event::Start(subscriber_saw) = subscriber_saw else { - panic!("the other edge's first event was not the namespace: {subscriber_saw:?}"); - }; + assert_eq!( + subscriber_saw.kind, + announce::Kind::Announced, + "the other edge's first event was not the namespace: {subscriber_saw:?}" + ); assert_eq!(subscriber_saw.prefix.as_str(), "room"); let subscriber_hop = match subscriber_saw.route.source() { origin::Source::Peer(hop) => hop, @@ -157,7 +155,7 @@ async fn dialed_anonymous_session_does_not_echo_or_route_subscribe_back() { assert_ne!(subscriber_hop, publisher_hop, "each dial gets its own hop"); assert_ne!(subscriber_hop, Hop::UNKNOWN); - moq_net_sim::sleep(SETTLE).await; + tokio::time::sleep(SETTLE).await; let echoed = traffic(&to_publisher.registry, stats::Role::Subscriber); assert_eq!( @@ -185,12 +183,12 @@ async fn dialed_anonymous_session_does_not_echo_or_route_subscribe_back() { .subscribe(subscription) .await .expect("subscriber subscribes"); - moq_net_sim::timeout(SETTLE, track.demand().used()) + tokio::time::timeout(SETTLE, track.demand().used()) .await .expect("the subscribe never reached the announcer") .unwrap(); - moq_net_sim::sleep(SETTLE).await; + tokio::time::sleep(SETTLE).await; let from_announcer = traffic(&to_publisher.registry, stats::Role::Publisher); assert_eq!(