From b3295ca5bda656ce41eadf72080f9fee9c206597 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 5 Nov 2026 22:27:54 -0800 Subject: [PATCH 1/8] chore(quest): claim quest/m0/test-flakes-2/ffi-datagram-roundtrip.md Quest-Claim: ceeefdf066d11d31 From fab404a8eca74024e92171f91d2e28d3f28d303c Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 6 Nov 2026 00:10:55 -0800 Subject: [PATCH 2/8] fix(net): a lite-05/06 SUBSCRIBE sends datagrams written since its TRACK_INFO On lite-05/06 a requester reads TRACK_INFO as the publisher accepting a name, but the datagrams flow on the SUBSCRIBE it pipelines on another stream. That SUBSCRIBE joined the datagram track only when it arrived, so a datagram written in between was never sent. Under CI load this timed out the FFI datagram connection tests. An answered TRACK now holds the name's datagram track, and a SUBSCRIBE of the same name on the session reads from a clone of that hold. The requester keeps its TRACK open until the SUBSCRIBE has a response, so the hold outlasts the SUBSCRIBE's arrival. Re-enables datagrams_cross_a_default_connection and finishes its quest. Co-Authored-By: Claude Opus 5.5 --- quest/m0/test-flakes-2/README.md | 1 - .../test-flakes-2/ffi-datagram-roundtrip.md | 59 ----- rs/moq-ffi/src/test.rs | 1 - rs/moq-net/src/lite/publisher.rs | 215 ++++++++++++++++-- sh/gh/test-inventory.allow | 1 - 5 files changed, 190 insertions(+), 87 deletions(-) delete mode 100644 quest/m0/test-flakes-2/ffi-datagram-roundtrip.md diff --git a/quest/m0/test-flakes-2/README.md b/quest/m0/test-flakes-2/README.md index 0f3622170f7..d3059a72be5 100644 --- a/quest/m0/test-flakes-2/README.md +++ b/quest/m0/test-flakes-2/README.md @@ -104,6 +104,5 @@ Public API: none. Wire: none. - [moq-gst's interleaved restart test passes under load](/quest/m0/test-flakes-2/gst-reconnect-interleave.md) - `restarts_under_reconnects_interleave_on_one_pad` passes in loaded runs and the merge queue, its missed delivery and its stray `Eos` each fixed at the cause - [reload.test.ts's close-drain tests pass under load](/quest/m0/test-flakes-2/reload-close-drain.md) - the drain cases in `js/net`'s reload tests pass in loaded multi-file runs - [dispatch-query's terminate test survives load](/quest/m0/test-flakes-2/dispatch-query-terminate.md) - `test_terminating_the_run_stops_the_running_phase` passes reliably under host load, re-enabled once fixed -- [Diagnose and re-enable the FFI datagram connection test](/quest/m0/test-flakes-2/ffi-datagram-roundtrip.md) - diagnose the default-connection timeout and re-enable the quarantined test - [Test PulseAudio servers stay off the desktop's D-Bus](/quest/m0/test-flakes-2/pulse-dbus.md) - concurrent media runs on one desktop never collide on `org.PulseAudio1` - [The P2P data-channel test transfers its frame](/quest/m0/test-flakes-2/p2p-data-channel.md) - Interop's two-peer data-channel test stops stalling after a failed send diff --git a/quest/m0/test-flakes-2/ffi-datagram-roundtrip.md b/quest/m0/test-flakes-2/ffi-datagram-roundtrip.md deleted file mode 100644 index 8629ae1993e..00000000000 --- a/quest/m0/test-flakes-2/ffi-datagram-roundtrip.md +++ /dev/null @@ -1,59 +0,0 @@ -# [S] Diagnose and re-enable the FFI datagram connection test - -## Goal - -`datagrams_cross_a_default_connection` and its `requested_` sibling -(`rs/moq-ffi/src/test.rs`) pass reliably under CI load after the timeout -cause is confirmed and fixed. - -## Plan - -Decided 2026-11-05 (maintainer, follow-up from -[moq-dev/moq#7914](https://github.com/moq-dev/moq/pull/7914)): ✅ m0. - -Decided 2026-11-05 (maintainer, /quest-plan): ✅ Keep only -`datagrams_cross_a_default_connection` quarantined and land the quarantine in -[moq-dev/moq#7954](https://github.com/moq-dev/moq/pull/7954). Re-diagnose under -CI load after [moq-dev/moq#7951](https://github.com/moq-dev/moq/pull/7951) -lands. Re-enabling now is rejected. Do not add repeated sending without -evidence of ordinary loss. - -The test sends one datagram over a real connection and waits 10 s for it; -it timed out in Check on -[moq-dev/moq#7914](https://github.com/moq-dev/moq/pull/7914) while passing -30 of 30 runs on an idle host. Confirm the cause with send/route diagnostics -before choosing the fix and re-enabling the test. - -Investigation 2026-11-05 on `main` at `419b799c75`: - -- The unchanged three connection cases pass locally. With temporary diagnostic - prints at `DatagramSend::send` and `Subscriber::route_datagram`, - `quest gate just rs test -p moq-ffi --lib -E 'test(datagrams_cross)' - --stress-count 100 --no-capture` passes all 300 cases. Every send returns - `Ok(())`, and all 300 received datagrams find a route. The diagnostic - prints were removed. This does not establish the cause of the CI timeout. -- All three tests await `consumer.info()` before appending. In - `DatagramServe::poll_serve`, the answer decodes before `request.accept` - exposes that info; route registration follows synchronously on that driver - before its next datagram receive pass. The answer race in - [moq-dev/moq#7951](https://github.com/moq-dev/moq/pull/7951) is real, - but was not reproduced by these tests. Do not claim it caused this timeout. -- `datagrams_cross_a_default_connection` is disabled pending diagnosis, per - the repository's flaky-test rule. Its requested siblings have not failed - in this investigation and remain enabled. No retry or longer wait was added. - -## Waiting - -- moq-dev/moq: wait for - [moq-dev/moq#7951](https://github.com/moq-dev/moq/pull/7951) to merge, then - capture a failing run under CI load with send/route diagnostics. Check the - PR's merged state with `gh pr view 7951 --json state,mergedAt`; once merged, - run the ignored case locally under CI-like concurrent build/test load, - with diagnostics at `DatagramSend::send` and `Subscriber::route_datagram`, - retaining a failing trace. Explicitly include the quarantine with - `quest gate just rs test -p moq-ffi --lib - -E 'test(datagrams_cross_a_default_connection)' --run-ignored all --no-capture`; - normal Check and queue runs intentionally skip it. Use it to distinguish ordinary transport loss, a route miss, or - another lifecycle bug and choose the corresponding fix before re-enabling. - Passing stress runs alone do not clear this wait or establish transport loss. - No workflow dispatch or host change is authorized for this investigation. diff --git a/rs/moq-ffi/src/test.rs b/rs/moq-ffi/src/test.rs index 7f5c2e452cd..b511bf69cc6 100644 --- a/rs/moq-ffi/src/test.rs +++ b/rs/moq-ffi/src/test.rs @@ -5986,7 +5986,6 @@ async fn datagram_connection( /// A native datagram-only reader reaches a publisher on the default lite-06 path. #[tokio::test] -#[ignore = "flaky under CI load; tracked in quest/m0/test-flakes-2/ffi-datagram-roundtrip.md"] async fn datagrams_cross_a_default_connection() { let (broadcast, remote, session, _server) = datagram_connection("sensor", vec![]).await; let producer = broadcast diff --git a/rs/moq-net/src/lite/publisher.rs b/rs/moq-net/src/lite/publisher.rs index 352701606d9..a897668b59f 100644 --- a/rs/moq-net/src/lite/publisher.rs +++ b/rs/moq-net/src/lite/publisher.rs @@ -96,6 +96,8 @@ struct Shared { peer_targets: Mutex, /// Numbers each PROBE stream for `peer_targets`. probe_streams: AtomicU64, + /// Lite-05/06: the datagram tracks answered TRACKs hold for their SUBSCRIBEs. + held_datagrams: HeldDatagrams, } /// The latest target each open PROBE stream sent, so the peer's target is the newest @@ -262,6 +264,7 @@ impl Publisher { peer_target: kio::Producer::new(crate::bandwidth::Rate::ZERO), peer_targets: Default::default(), probe_streams: AtomicU64::new(0), + held_datagrams: Default::default(), }), runtime: config.runtime, accept, @@ -2088,7 +2091,11 @@ impl Drop for RequestServe { enum TrackInfoServe { Track(track::Resolving, Option>), /// Lite-05/06: a name no stream track has, answered for its datagram track. - Datagrams(datagram::Consumer), + Datagrams(datagram::Consumer, crate::broadcast::Consumer), + /// Answered, holding the name's datagram track for the SUBSCRIBE alongside on Lite-05/06. + Answered { + _hold: Option, + }, } impl Request for TrackInfoServe { @@ -2120,10 +2127,10 @@ impl Request for TrackInfoServe { track: &str, broadcast: crate::broadcast::Consumer, ) -> Result { - let legacy = Legacy::new(shared.version, &broadcast, track); + let legacy = Legacy::new(shared, &broadcast, track); match broadcast.track(track) { Ok(track) => Ok(Self::Track(track.info().into_inner(), legacy)), - Err(err) => Ok(Self::Datagrams(Legacy::datagrams(legacy, err)?)), + Err(err) => Ok(Self::Datagrams(Legacy::datagrams(legacy, err)?, broadcast)), } } @@ -2140,20 +2147,23 @@ impl Request for TrackInfoServe { let ietf = ready!(shared.peer_setup.poll_ietf(waiter)); if let Self::Track(querying, legacy) = self { match querying.poll_ok(waiter) { - Poll::Ready(Err(err)) => *self = Self::Datagrams(Legacy::datagrams(legacy.take(), err)?), - // The stream track answers: let a pending datagram ask go, rather than hold it - // as long as the requester keeps the stream open. - Poll::Ready(Ok(_)) => drop(legacy.take()), + Poll::Ready(Err(err)) => { + let broadcast = legacy.as_ref().map(|legacy| legacy.broadcast.clone()); + let consumer = Legacy::datagrams(legacy.take(), err)?; + *self = Self::Datagrams(consumer, broadcast.expect("only a fallback serves datagrams")); + } + Poll::Ready(Ok(_)) => {} Poll::Pending => { + let broadcast = legacy.as_ref().map(|legacy| legacy.broadcast.clone()); if let Poll::Ready(consumer) = Legacy::poll_datagrams(legacy, waiter) { - *self = Self::Datagrams(consumer); + *self = Self::Datagrams(consumer, broadcast.expect("only a fallback serves datagrams")); } } } } - let querying = match self { - Self::Track(querying, _) => querying, - Self::Datagrams(consumer) => { + let (querying, legacy) = match self { + Self::Track(querying, legacy) => (querying, legacy), + Self::Datagrams(consumer, _) => { let info = ready!(consumer.poll_info(waiter))?; writer.set_priority(crate::coding::TRACK_INFO_SEND_ORDER); writer.buffer(&lite::TrackInfo { @@ -2168,11 +2178,25 @@ impl Request for TrackInfoServe { if ietf { writer.buffer(&lite::ietf::TrackProperties(Default::default()))?; } + let Self::Datagrams(consumer, broadcast) = std::mem::replace(self, Self::Answered { _hold: None }) + else { + unreachable!() + }; + *self = Self::Answered { + _hold: Some(shared.held_datagrams.hold(&broadcast, consumer)), + }; return Poll::Ready(Ok(ControlFlow::Break(()))); } + Self::Answered { .. } => unreachable!("answered once"), }; let info = ready!(querying.poll_ok(waiter))?; crate::ietf::flate::egress(&info, shared.version.has_compression())?; + // The datagrams beside the stream track, read from this answer on. A pending ask goes, + // rather than held as long as the requester keeps the stream open. + let hold = legacy.take().and_then(|legacy| { + let broadcast = legacy.broadcast.clone(); + Some(shared.held_datagrams.hold(&broadcast, legacy.live()?)) + }); // Ahead of the track's groups, which a subscriber holds unread until it lands. writer.set_priority(crate::coding::TRACK_INFO_SEND_ORDER); @@ -2181,6 +2205,7 @@ impl Request for TrackInfoServe { if ietf { writer.buffer(&lite::ietf::TrackProperties(info.properties.clone()))?; } + *self = Self::Answered { _hold: hold }; Poll::Ready(Ok(ControlFlow::Break(()))) } } @@ -2339,7 +2364,7 @@ impl Request for SubscribeServe { track: &str, broadcast: crate::broadcast::Consumer, ) -> Result { - let legacy = Legacy::new(shared.version, &broadcast, track); + let legacy = Legacy::new(shared, &broadcast, track); let track = match broadcast.track(track) { Ok(track) => track, Err(err) => return Ok(Self::datagrams(Legacy::datagrams(legacy, err)?, &msg, None)), @@ -2643,17 +2668,31 @@ struct Legacy { } impl Legacy { - /// The fallback for a request on `version`, or `None` when the stream track's answer stands. - fn new(version: Version, broadcast: &crate::broadcast::Consumer, name: &str) -> Option> { + /// The fallback for a request on the session's version, or `None` when the stream + /// track's answer stands. It starts from the datagram track an answered TRACK of the + /// name holds, if any (see [`HeldDatagrams`]). + fn new(shared: &Shared, broadcast: &crate::broadcast::Consumer, name: &str) -> Option> { + let version = shared.version; (version.has_datagrams() && !version.has_datagram_tracks()).then(|| { Box::new(Self { broadcast: broadcast.clone(), name: name.to_string(), - datagrams: None, + datagrams: shared.held_datagrams.find(broadcast, name).map(Some), }) }) } + /// The name's live datagram track, asking no handler: the request's own ask once + /// answered, or one published or already asked for. An unanswered ask is dropped, + /// which withdraws it. + fn live(self) -> Option { + let answered = self + .datagrams + .flatten() + .filter(|asked| matches!(asked.poll_info(&kio::Waiter::noop()), Poll::Ready(Ok(_)))); + answered.or_else(|| self.broadcast.datagrams_live(&self.name)) + } + /// The datagram track to serve before the stream track answers: one a local handler or /// producer answered while no producer serves the stream track. Pending otherwise, /// waking with the stream track's lookup, which the caller polls first. @@ -2703,6 +2742,48 @@ impl Legacy { } } +/// Lite-05/06: the datagram tracks this session's answered TRACKs hold, read from each +/// answer on. The requester reads TRACK_INFO as the publisher accepting the name, but the +/// datagrams flow on the SUBSCRIBE it sends alongside, a separate stream that can arrive +/// later. That SUBSCRIBE reads from a clone of the hold, keeping its position, rather than +/// from its own arrival, which would lose what was written in between. The requester keeps +/// its TRACK open until the SUBSCRIBE has a response, so the hold outlasts its arrival. +#[derive(Clone, Default)] +struct HeldDatagrams(Arc>>); + +impl HeldDatagrams { + /// Hold `consumer` of `broadcast` until the returned hold drops. + fn hold(&self, broadcast: &crate::broadcast::Consumer, consumer: datagram::Consumer) -> DatagramHold { + let key = self.0.lock().unwrap().insert((broadcast.clone(), consumer)); + DatagramHold { + held: self.clone(), + key, + } + } + + /// A clone of the datagram track `name` of `broadcast` a TRACK holds, if any. + fn find(&self, broadcast: &crate::broadcast::Consumer, name: &str) -> Option { + let held = self.0.lock().unwrap(); + held.iter() + .find(|(_, (held, consumer))| consumer.name() == name && held.is_clone(broadcast)) + .map(|(_, (_, consumer))| consumer.clone()) + } +} + +/// One answered TRACK's entry in [`HeldDatagrams`]. +struct DatagramHold { + held: HeldDatagrams, + key: usize, +} + +impl Drop for DatagramHold { + fn drop(&mut self) { + // Dropped outside the lock. + let consumer = self.held.0.lock().unwrap().remove(self.key); + drop(consumer); + } +} + /// Lite-05/06: a name's datagram track carried on the subscription of its stream track, as /// one subscription carries both kinds there. A subscriber keeps the kind it asked for. struct LegacyDatagrams { @@ -2723,16 +2804,7 @@ impl LegacyDatagrams { /// subscribes to, so this asks no handler: a request would subscribe to the name again. /// The request's own ask joins only once answered, the track then being live. fn join(legacy: Option>, id: u64, priority: i8, timescale: Option) -> Option { - let Legacy { - broadcast, - name, - datagrams, - } = *legacy?; - // An unanswered ask is dropped, which withdraws it. - let answered = datagrams - .flatten() - .filter(|asked| matches!(asked.poll_info(&kio::Waiter::noop()), Poll::Ready(Ok(_)))); - let mut consumer = answered.or_else(|| broadcast.datagrams_live(&name))?; + let mut consumer = legacy?.live()?; consumer.set_priority(priority); Some(Self { consumer, @@ -9400,6 +9472,99 @@ mod tests { assert!(crate::coding::TRACK_INFO_SEND_ORDER > top); } + /// Lite-05/06: a requester reads TRACK_INFO as the name being accepted, and its SUBSCRIBE, + /// on a stream of its own, can arrive after a datagram written meanwhile. That SUBSCRIBE + /// still sends it, whether the datagram track answered the TRACK or rides beside a stream + /// track of the same name. + #[kio::test] + async fn a_subscribe_sends_the_datagrams_written_since_its_track_info() { + for version in [Version::Lite05, Version::Lite06] { + for stream in [false, true] { + let case = format!("{version:?}, stream track {stream}"); + let origin = crate::origin::Config::new(crate::Hop::new(1).unwrap()).produce(); + let broadcast = origin.route("room", crate::origin::RouteOptions::hidden()).unwrap(); + let mut datagrams = broadcast + .broadcast() + .create_datagrams("events", datagram::Info::default()) + .unwrap(); + let _track = stream.then(|| { + broadcast + .broadcast() + .create_track("events", crate::track::Info::timed()) + .unwrap() + }); + broadcast.announce().unwrap(); + + let peer_setup = crate::lite::PeerSetup::new(false, crate::transport::Address::Setup); + peer_setup.set(crate::lite::Setup::default()).unwrap(); + let (_, goaway) = crate::goaway::Handle::new(true); + let sink = SinkSession::new(crate::lite::test_transport::Log::default()).with_datagrams(1200); + let log = sink.log.clone(); + let publisher = Publisher::new(PublisherConfig { + estimate: kio::Producer::default().weak(), + runtime: crate::time::sim(), + session: crate::driver::Transport::new(sink), + origin: origin.consume(), + version, + peer_setup, + goaway, + peer_hop: None, + subscriptions: Default::default(), + auth: crate::auth::Link::new(false), + peer_grant: Default::default(), + client: false, + }); + let open = |script: Vec| async move { + let session = ScriptedSession::new(script); + let (send, recv) = kio::wait(|waiter| { + ::poll_open_bi( + &session, + waiter, + Default::default(), + ) + }) + .await + .unwrap(); + Stream:: { + writer: Writer::new(crate::transport::boxed::SendStream::new(send), version), + reader: crate::coding::Reader::new(crate::transport::boxed::RecvStream::new(recv), version), + } + }; + + let mut script = Vec::new(); + lite::Track { + instance: None, + broadcast: crate::Path::from_static("room"), + track: "events".into(), + } + .encode(&mut crate::coding::Encoder::new(&mut script, version.into()), version) + .unwrap(); + let mut track_info = RequestServe::::new(publisher.shared.clone(), open(script).await); + assert!(!drive(&mut track_info).await, "{case}: the TRACK stream is held"); + + datagrams.append(None, b"between".as_slice()).unwrap(); + + let mut script = Vec::new(); + lite::Subscribe { + order: crate::track::Order::Descending, + instance: None, + id: 0, + broadcast: crate::Path::from_static("room"), + track: "events".into(), + priority: 0, + max_delay: Duration::ZERO, + min_delay: None, + ranges: vec![lite::Range::span(None, 0, None, None)], + } + .encode(&mut crate::coding::Encoder::new(&mut script, version.into()), version) + .unwrap(); + let mut subscribe = RequestServe::::new(publisher.shared.clone(), open(script).await); + assert!(!drive(&mut subscribe).await, "{case}: the subscription is live"); + assert_eq!(log.datagrams.lock().unwrap().len(), 1, "{case}"); + } + } + } + /// An answered TRACK stream holds its track until the requester closes its side, by /// FIN or reset, and the hold owes nothing a draining close would wait for. A requester /// that leaves while the reply is still blocked on flow control lets the track go too. diff --git a/sh/gh/test-inventory.allow b/sh/gh/test-inventory.allow index 033ae5389de..6b646c3fd6a 100644 --- a/sh/gh/test-inventory.allow +++ b/sh/gh/test-inventory.allow @@ -3,7 +3,6 @@ # orphan missing here, and fails on an entry that has a caller. # Flaky under load; the quest makes it reliable, then removes its `#[ignore]`. -ignore moq-ffi::test::datagrams_cross_a_default_connection /quest/m0/test-flakes-2/ffi-datagram-roundtrip.md ignore moq-gst::source::imp::session_tests::restarts_under_reconnects_interleave_on_one_pad /quest/m0/test-flakes-2/gst-reconnect-interleave.md # Need a GPU, a desktop, or a device that no CI runner has. From ad599d0334e6ed0555c7c80a0fe979de76fa3448 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 6 Nov 2026 01:17:03 -0800 Subject: [PATCH 3/8] fix(net): an answered TRACK keeps its query beside the held datagrams Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/lite/publisher.rs | 31 ++++++++++++++----------------- 1 file changed, 14 insertions(+), 17 deletions(-) diff --git a/rs/moq-net/src/lite/publisher.rs b/rs/moq-net/src/lite/publisher.rs index a897668b59f..60cf165e503 100644 --- a/rs/moq-net/src/lite/publisher.rs +++ b/rs/moq-net/src/lite/publisher.rs @@ -2089,13 +2089,13 @@ impl Drop for RequestServe { /// The open stream is interest in the track: the query that answered it stays until the /// requester closes its side, so demand holds while the requester moves on to SUBSCRIBE. enum TrackInfoServe { - Track(track::Resolving, Option>), + /// The stream track's query, its Lite-05/06 fallback, and once answered, the name's + /// datagram track held for the SUBSCRIBE alongside. + Track(track::Resolving, Option>, Option), /// Lite-05/06: a name no stream track has, answered for its datagram track. Datagrams(datagram::Consumer, crate::broadcast::Consumer), - /// Answered, holding the name's datagram track for the SUBSCRIBE alongside on Lite-05/06. - Answered { - _hold: Option, - }, + /// Answered from the datagram track, now held for the SUBSCRIBE alongside. + Held(#[expect(dead_code, reason = "held for its drop")] DatagramHold), } impl Request for TrackInfoServe { @@ -2129,7 +2129,7 @@ impl Request for TrackInfoServe { ) -> Result { let legacy = Legacy::new(shared, &broadcast, track); match broadcast.track(track) { - Ok(track) => Ok(Self::Track(track.info().into_inner(), legacy)), + Ok(track) => Ok(Self::Track(track.info().into_inner(), legacy, None)), Err(err) => Ok(Self::Datagrams(Legacy::datagrams(legacy, err)?, broadcast)), } } @@ -2145,7 +2145,7 @@ impl Request for TrackInfoServe { waiter: &kio::Waiter, ) -> Poll, Error>> { let ietf = ready!(shared.peer_setup.poll_ietf(waiter)); - if let Self::Track(querying, legacy) = self { + if let Self::Track(querying, legacy, _) = self { match querying.poll_ok(waiter) { Poll::Ready(Err(err)) => { let broadcast = legacy.as_ref().map(|legacy| legacy.broadcast.clone()); @@ -2161,8 +2161,8 @@ impl Request for TrackInfoServe { } } } - let (querying, legacy) = match self { - Self::Track(querying, legacy) => (querying, legacy), + let (querying, legacy, held) = match self { + Self::Track(querying, legacy, held) => (querying, legacy, held), Self::Datagrams(consumer, _) => { let info = ready!(consumer.poll_info(waiter))?; writer.set_priority(crate::coding::TRACK_INFO_SEND_ORDER); @@ -2178,22 +2178,20 @@ impl Request for TrackInfoServe { if ietf { writer.buffer(&lite::ietf::TrackProperties(Default::default()))?; } - let Self::Datagrams(consumer, broadcast) = std::mem::replace(self, Self::Answered { _hold: None }) - else { + let consumer = consumer.clone(); + let Self::Datagrams(_, broadcast) = &*self else { unreachable!() }; - *self = Self::Answered { - _hold: Some(shared.held_datagrams.hold(&broadcast, consumer)), - }; + *self = Self::Held(shared.held_datagrams.hold(broadcast, consumer)); return Poll::Ready(Ok(ControlFlow::Break(()))); } - Self::Answered { .. } => unreachable!("answered once"), + Self::Held(_) => unreachable!("answered once"), }; let info = ready!(querying.poll_ok(waiter))?; crate::ietf::flate::egress(&info, shared.version.has_compression())?; // The datagrams beside the stream track, read from this answer on. A pending ask goes, // rather than held as long as the requester keeps the stream open. - let hold = legacy.take().and_then(|legacy| { + *held = legacy.take().and_then(|legacy| { let broadcast = legacy.broadcast.clone(); Some(shared.held_datagrams.hold(&broadcast, legacy.live()?)) }); @@ -2205,7 +2203,6 @@ impl Request for TrackInfoServe { if ietf { writer.buffer(&lite::ietf::TrackProperties(info.properties.clone()))?; } - *self = Self::Answered { _hold: hold }; Poll::Ready(Ok(ControlFlow::Break(()))) } } From 2877bacd334b7c8672ef7c24b6d49dd4e4f82270 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 6 Nov 2026 02:08:06 -0800 Subject: [PATCH 4/8] fix(net): mirror the lite-05/06 TRACK datagram hold in the JS publisher The JS publisher closed the datagram consumer that answered a TRACK and had its SUBSCRIBE join afresh, losing datagrams written in between. An answered TRACK now holds the name's datagram track until the requester closes it, and a SUBSCRIBE of the name reads from a clone of the hold. Co-Authored-By: Claude Opus 5.5 --- js/net/src/lite/publisher.test.ts | 54 +++++++++++++++++++++++++++- js/net/src/lite/publisher.ts | 58 +++++++++++++++++++++++++++---- 2 files changed, 105 insertions(+), 7 deletions(-) diff --git a/js/net/src/lite/publisher.test.ts b/js/net/src/lite/publisher.test.ts index eae0748cf2f..e814bd5143a 100644 --- a/js/net/src/lite/publisher.test.ts +++ b/js/net/src/lite/publisher.test.ts @@ -28,7 +28,7 @@ import { type SubscribeFinish, SubscribeUpdate, } from "./subscribe.ts"; -import { Track as TrackMessage } from "./track.ts"; +import { TrackInfo, Track as TrackMessage } from "./track.ts"; import { ALPN_05, ALPN_06, ALPN_07_WIP, hasRanges, keepsSubscribeOpen, Version } from "./version.ts"; function publish(origin: OriginProducer, path: Path.Valid) { @@ -3404,3 +3404,55 @@ test("lite-06: a datagram request failing behind the shared writer resets rather pair.server.close(); } }); + +// Lite-05/06: a requester reads TRACK_INFO as the name being accepted, and its SUBSCRIBE, on a +// stream of its own, can arrive after a datagram written meanwhile. That SUBSCRIBE still sends +// it, whether the datagram track answered the TRACK or rides beside a stream track of the name. +for (const version of [Version.DRAFT_05, Version.DRAFT_06]) { + for (const stream of [false, true]) { + const label = version === Version.DRAFT_06 ? "lite-06" : "lite-05"; + test(`${label}, stream track ${stream}: a SUBSCRIBE sends the datagrams written since its TRACK_INFO`, async () => { + const pair = createMockTransportPair(version === Version.DRAFT_06 ? ALPN_06 : ALPN_05); + const sent: Uint8Array[] = []; + Object.defineProperty(pair.server.datagrams, "writable", { + value: new WritableStream({ + write(chunk) { + sent.push(chunk); + }, + }), + }); + const origin = new OriginProducer(); + const publisher = new Publisher(pair.server, version, randomHop(), origin.consume()); + const broadcast = publish(origin, Path.from("test")); + const datagrams = broadcast.createDatagrams("events"); + if (stream) broadcast.createTrack("events", { timescale: Timescale.MILLI }); + + const track = await Stream.open(pair.client, { version }); + const trackServer = await Stream.accept(pair.server, version); + if (!trackServer) throw new Error("publisher never accepted the track stream"); + void publisher.runTrackInfo(new TrackMessage(Path.from("test"), "events"), trackServer); + await TrackInfo.decode(track.reader, version); + + datagrams.append(undefined, Uint8Array.of(0xaa)); + + const subscribe = await Stream.open(pair.client, { version }); + const subscribeServer = await Stream.accept(pair.server, version); + if (!subscribeServer) throw new Error("publisher never accepted the subscribe stream"); + try { + void publisher.runSubscribe( + new Subscribe({ id: 0n, broadcast: Path.from("test"), track: "events", priority: 0 }), + subscribeServer, + ); + for (let i = 0; i < 20 && sent.length === 0; i++) await new Promise((resolve) => setImmediate(resolve)); + expect(sent.length).toBe(1); + } finally { + track.close(); + subscribe.close(); + publisher.close(); + origin.close(); + pair.client.close(); + pair.server.close(); + } + }); + } +} diff --git a/js/net/src/lite/publisher.ts b/js/net/src/lite/publisher.ts index 4e75c4430c3..fad1f1cee54 100644 --- a/js/net/src/lite/publisher.ts +++ b/js/net/src/lite/publisher.ts @@ -542,6 +542,14 @@ export class Publisher { // A rejected lookup is evicted so a retry can re-probe. #trackInfo = new WeakMap>(); + // Lite-05/06: the datagram tracks this session's answered TRACKs hold, read from each + // answer on. The requester reads TRACK_INFO as the publisher accepting the name, but the + // datagrams flow on the SUBSCRIBE it sends alongside, which can arrive later. That + // SUBSCRIBE reads from a clone of the hold, keeping its position, rather than from its own + // arrival, which would lose what was written in between. The requester keeps its TRACK open + // until the SUBSCRIBE has a response, so the hold outlasts its arrival. + #heldDatagrams = new WeakMap>>(); + /** * Creates a new Publisher instance. * @param quic - The WebTransport session to use @@ -1085,7 +1093,7 @@ export class Publisher { // Serve datagrams concurrently with groups whenever the transport carries them // (the writer exists iff so). No group fallback: otherwise they simply aren't sent. if (this.#datagramWriter && !hasDatagramTracks(this.version)) { - legacy = wireOf(front).liveDatagrams(msg.track); + legacy = this.#held(front, msg.track)?.clone() ?? wireOf(front).liveDatagrams(msg.track); if (legacy) { legacy.setPriority(msg.priority); datagrams = this.#runDatagrams(legacy, { sub: msg.id, stamp }) @@ -1814,6 +1822,7 @@ export class Publisher { if (!front) throw new NotFound(`broadcast ${msg.broadcast}`); let info: TrackInfoMessage; + let datagrams: datagram.Consumer | undefined; try { info = await this.#resolveTrackInfo(front, msg.track, hold.signal); } catch (err) { @@ -1823,13 +1832,21 @@ export class Publisher { toStreamCode(error(err)) !== StreamCode.NotFound ) throw err; - const datagrams = front.datagrams(msg.track); + const asked = front.datagrams(msg.track); try { - info = trackInfoMessage(await untilAborted(datagrams.info(), hold.signal), this.version); - } finally { - datagrams.close(); + info = trackInfoMessage(await untilAborted(asked.info(), hold.signal), this.version); + } catch (err) { + asked.close(); + throw err; } + datagrams = asked; } + // Lite-05/06: the name's datagram track, read from this answer on, held for the + // SUBSCRIBE alongside, beside a stream track too. + if (hasDatagrams(this.version) && !hasDatagramTracks(this.version)) { + datagrams ??= wireOf(front).liveDatagrams(msg.track); + } + if (datagrams) this.#holdDatagrams(front, datagrams, hold.signal); if (watch.revoked()) throw unauthorized(msg.broadcast); await info.encode(stream.writer, this.version); console.debug(`track info: broadcast=${msg.broadcast} track=${msg.track}`); @@ -1911,6 +1928,35 @@ export class Publisher { } } + // Hold `consumer` of `front` for the SUBSCRIBE alongside until `hold` aborts. + #holdDatagrams(front: broadcast.Consumer, consumer: datagram.Consumer, hold: AbortSignal) { + let names = this.#heldDatagrams.get(front); + if (!names) { + names = new Map(); + this.#heldDatagrams.set(front, names); + } + let held = names.get(consumer.name); + if (!held) { + held = new Set(); + names.set(consumer.name, held); + } + const set = held; + set.add(consumer); + const release = () => { + set.delete(consumer); + if (set.size === 0 && names.get(consumer.name) === set) names.delete(consumer.name); + consumer.close(); + }; + if (hold.aborted) release(); + else hold.addEventListener("abort", release, { once: true }); + } + + // The datagram track `name` of `front` an answered TRACK holds, if any. + #held(front: broadcast.Consumer, name: string): datagram.Consumer | undefined { + const held = this.#heldDatagrams.get(front)?.get(name); + return held?.values().next().value; + } + // Resolve (and cache) a track's immutable TRACK_INFO by asking the application. // `resolveTrackInfo` triggers a TrackRequest the app answers with accept(TrackInfo); // only the immutable properties are needed (not the groups). Cached because they're @@ -2027,7 +2073,7 @@ export class Publisher { stream: Stream, signal: AbortSignal, ): Promise { - const track = front.datagrams(msg.track); + const track = this.#held(front, msg.track)?.clone() ?? front.datagrams(msg.track); track.setPriority(msg.priority); const updates = (async () => { for (;;) { From ccd385598fa811e4620eaa8fc39a5507f48b8a38 Mon Sep 17 00:00:00 2001 From: "pullfrog[bot]" <226033991+pullfrog[bot]@users.noreply.github.com> Date: Fri, 6 Nov 2026 11:22:58 +0000 Subject: [PATCH 5/8] test(net): keep the datagram end ahead of the session close under load --- js/net/src/lite/datagram-tracks.test.ts | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/js/net/src/lite/datagram-tracks.test.ts b/js/net/src/lite/datagram-tracks.test.ts index a79af11067f..3a843622e5f 100644 --- a/js/net/src/lite/datagram-tracks.test.ts +++ b/js/net/src/lite/datagram-tracks.test.ts @@ -464,15 +464,20 @@ for (const alpn of [ALPN_05, ALPN_06, ALPN_07_WIP]) { const remote = wireOf(client).consume(path); const reader = remote.datagrams("audio"); reader.setPriority(5); + // The end has to reach the client before the abort, and the mock delivers it in + // microtasks. A write-budget yield is a host task that the drain below can't advance, + // so under CPU load the publisher's answer could still be stalled when the abort lands. + const budget = spyOn(Budget.prototype, "poll").mockImplementation(() => undefined); try { await reader.info(); while (datagrams.priority.peek() !== 5) await datagrams.priority.changed(); datagrams.finish(); // The mock delivers the end in microtasks, all within this host task. - for (let i = 0; i < 1000; i++) await Promise.resolve(); + for (let i = 0; i < 10000; i++) await Promise.resolve(); client.abort(); expect(await reader.recv()).toBeUndefined(); } finally { + budget.mockRestore(); reader.close(); remote.close(); client.abort(); From 39af688dc917ac0a98ffaff9df5ef8dd1923f500 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 6 Nov 2026 08:20:46 -0800 Subject: [PATCH 6/8] fix(net): adopt a datagram hold of the same track when a SUBSCRIBE picks it A lite-05/06 SUBSCRIBE now looks for a hold when it picks its datagram consumer rather than when it starts, so one parked on a relayed stream track still reads from the TRACK answered meanwhile. A hold is matched by the track it reads, not the name, so a replaced track's hold never stands in for the current one, and it sits at the lowest priority so it never outranks the SUBSCRIBE. JS mirrors both. Co-Authored-By: Claude Opus 5.5 --- js/net/src/datagram.ts | 1 + js/net/src/internal.ts | 5 + js/net/src/lite/publisher.test.ts | 121 +++++++---- js/net/src/lite/publisher.ts | 54 +++-- rs/moq-net/src/lite/publisher.rs | 333 ++++++++++++++++++++---------- 5 files changed, 336 insertions(+), 178 deletions(-) diff --git a/js/net/src/datagram.ts b/js/net/src/datagram.ts index 2e111d3847d..3345dcc41ab 100644 --- a/js/net/src/datagram.ts +++ b/js/net/src/datagram.ts @@ -244,6 +244,7 @@ export class Consumer { consume = (name, state, cursor, priority) => new Consumer(name, state, cursor, priority); hooks.datagramPriority = (consumer) => consumer.#priority; hooks.datagramClosed = (consumer) => consumer.#closed; + hooks.datagramSameTrack = (a, b) => a.#state === b.#state; } /** The track's publisher properties, once accepted. */ diff --git a/js/net/src/internal.ts b/js/net/src/internal.ts index 1f068aa70e5..23e043f657c 100644 --- a/js/net/src/internal.ts +++ b/js/net/src/internal.ts @@ -457,6 +457,8 @@ export const hooks: { datagramPriority: (consumer: datagram.Consumer) => Getter; /** Settles once this consumer is closed, the end of whatever request it serves. */ datagramClosed: (consumer: datagram.Consumer) => GetPromise; + /** Whether two consumers read the same datagram track. */ + datagramSameTrack: (a: datagram.Consumer, b: datagram.Consumer) => boolean; /** * Declare the end a wire subscription received: repeating it is a no-op, and a mid-group * end may refine the group boundary it falls just before. Throws `ProtocolViolation` on @@ -776,6 +778,9 @@ export const hooks: { datagramClosed: () => { throw new Error("datagram module not loaded"); }, + datagramSameTrack: () => { + throw new Error("datagram module not loaded"); + }, endTrack: () => { throw new Error("track.ts not loaded"); }, diff --git a/js/net/src/lite/publisher.test.ts b/js/net/src/lite/publisher.test.ts index e814bd5143a..d3352cb0606 100644 --- a/js/net/src/lite/publisher.test.ts +++ b/js/net/src/lite/publisher.test.ts @@ -3405,54 +3405,95 @@ test("lite-06: a datagram request failing behind the shared writer resets rather } }); +// A lite-05/06 publisher of broadcast "test", recording the datagrams it sends. +function legacyPublisher(version: Version) { + const pair = createMockTransportPair(version === Version.DRAFT_06 ? ALPN_06 : ALPN_05); + const sent: Uint8Array[] = []; + Object.defineProperty(pair.server.datagrams, "writable", { + value: new WritableStream({ + write(chunk) { + sent.push(chunk); + }, + }), + }); + const origin = new OriginProducer(); + const publisher = new Publisher(pair.server, version, randomHop(), origin.consume()); + const broadcast = publish(origin, Path.from("test")); + const streams: Stream[] = []; + return { + sent, + broadcast, + // Send a TRACK for "events" and read its TRACK_INFO. + async track() { + const track = await Stream.open(pair.client, { version }); + streams.push(track); + const server = await Stream.accept(pair.server, version); + if (!server) throw new Error("publisher never accepted the track stream"); + void publisher.runTrackInfo(new TrackMessage(Path.from("test"), "events"), server); + await TrackInfo.decode(track.reader, version); + }, + // SUBSCRIBE to "events" and wait for its first datagram. + async subscribe(priority: number) { + const subscribe = await Stream.open(pair.client, { version }); + streams.push(subscribe); + const server = await Stream.accept(pair.server, version); + if (!server) throw new Error("publisher never accepted the subscribe stream"); + void publisher.runSubscribe( + new Subscribe({ id: 0n, broadcast: Path.from("test"), track: "events", priority }), + server, + ); + for (let i = 0; i < 20 && sent.length === 0; i++) await new Promise((resolve) => setImmediate(resolve)); + }, + close() { + for (const stream of streams) stream.close(); + publisher.close(); + origin.close(); + pair.client.close(); + pair.server.close(); + }, + }; +} + // Lite-05/06: a requester reads TRACK_INFO as the name being accepted, and its SUBSCRIBE, on a // stream of its own, can arrive after a datagram written meanwhile. That SUBSCRIBE still sends -// it, whether the datagram track answered the TRACK or rides beside a stream track of the name. +// it, whether the datagram track answered the TRACK or rides beside a stream track of the name, +// and the hold keeps no priority of its own. for (const version of [Version.DRAFT_05, Version.DRAFT_06]) { + const label = version === Version.DRAFT_06 ? "lite-06" : "lite-05"; for (const stream of [false, true]) { - const label = version === Version.DRAFT_06 ? "lite-06" : "lite-05"; test(`${label}, stream track ${stream}: a SUBSCRIBE sends the datagrams written since its TRACK_INFO`, async () => { - const pair = createMockTransportPair(version === Version.DRAFT_06 ? ALPN_06 : ALPN_05); - const sent: Uint8Array[] = []; - Object.defineProperty(pair.server.datagrams, "writable", { - value: new WritableStream({ - write(chunk) { - sent.push(chunk); - }, - }), - }); - const origin = new OriginProducer(); - const publisher = new Publisher(pair.server, version, randomHop(), origin.consume()); - const broadcast = publish(origin, Path.from("test")); - const datagrams = broadcast.createDatagrams("events"); - if (stream) broadcast.createTrack("events", { timescale: Timescale.MILLI }); - - const track = await Stream.open(pair.client, { version }); - const trackServer = await Stream.accept(pair.server, version); - if (!trackServer) throw new Error("publisher never accepted the track stream"); - void publisher.runTrackInfo(new TrackMessage(Path.from("test"), "events"), trackServer); - await TrackInfo.decode(track.reader, version); - - datagrams.append(undefined, Uint8Array.of(0xaa)); - - const subscribe = await Stream.open(pair.client, { version }); - const subscribeServer = await Stream.accept(pair.server, version); - if (!subscribeServer) throw new Error("publisher never accepted the subscribe stream"); + const legacy = legacyPublisher(version); try { - void publisher.runSubscribe( - new Subscribe({ id: 0n, broadcast: Path.from("test"), track: "events", priority: 0 }), - subscribeServer, - ); - for (let i = 0; i < 20 && sent.length === 0; i++) await new Promise((resolve) => setImmediate(resolve)); - expect(sent.length).toBe(1); + const datagrams = legacy.broadcast.createDatagrams("events"); + if (stream) legacy.broadcast.createTrack("events", { timescale: Timescale.MILLI }); + await legacy.track(); + datagrams.append(undefined, Uint8Array.of(0xaa)); + await legacy.subscribe(-5); + expect(legacy.sent.length).toBe(1); + expect(datagrams.priority.peek()).toBe(-5); } finally { - track.close(); - subscribe.close(); - publisher.close(); - origin.close(); - pair.client.close(); - pair.server.close(); + legacy.close(); } }); } + + // A hold reads the datagram track that answered its TRACK. Once that track is replaced, a + // SUBSCRIBE reads the replacement, even while the older TRACK stays open. + test(`${label}: a SUBSCRIBE reads the current datagram track past an older hold`, async () => { + const legacy = legacyPublisher(version); + try { + const first = legacy.broadcast.createDatagrams("events"); + await legacy.track(); + first.append(undefined, Uint8Array.of(0xaa)); + first.finish(); + const current = legacy.broadcast.createDatagrams("events"); + await legacy.track(); + current.append(undefined, Uint8Array.of(0xbb)); + await legacy.subscribe(0); + expect(legacy.sent.length).toBe(1); + expect(legacy.sent[0]?.at(-1)).toBe(0xbb); + } finally { + legacy.close(); + } + }); } diff --git a/js/net/src/lite/publisher.ts b/js/net/src/lite/publisher.ts index fad1f1cee54..626a197b345 100644 --- a/js/net/src/lite/publisher.ts +++ b/js/net/src/lite/publisher.ts @@ -543,12 +543,15 @@ export class Publisher { #trackInfo = new WeakMap>(); // Lite-05/06: the datagram tracks this session's answered TRACKs hold, read from each - // answer on. The requester reads TRACK_INFO as the publisher accepting the name, but the - // datagrams flow on the SUBSCRIBE it sends alongside, which can arrive later. That - // SUBSCRIBE reads from a clone of the hold, keeping its position, rather than from its own - // arrival, which would lose what was written in between. The requester keeps its TRACK open - // until the SUBSCRIBE has a response, so the hold outlasts its arrival. - #heldDatagrams = new WeakMap>>(); + // answer on, oldest first. The requester reads TRACK_INFO as the publisher accepting the + // name, but the datagrams flow on the SUBSCRIBE it sends alongside, which can arrive later. + // That SUBSCRIBE reads from a clone of the hold, keeping its position, rather than from its + // own arrival, which would lose what was written in between. The requester keeps its TRACK + // open until the SUBSCRIBE has a response, so the hold outlasts its arrival. A hold is + // matched by the track it reads, not by name, so one left on a replaced track never stands + // in for the current one. It sits at the lowest priority, keeping a position without + // outranking the subscriptions that read the track. + #heldDatagrams = new Set(); /** * Creates a new Publisher instance. @@ -1093,7 +1096,8 @@ export class Publisher { // Serve datagrams concurrently with groups whenever the transport carries them // (the writer exists iff so). No group fallback: otherwise they simply aren't sent. if (this.#datagramWriter && !hasDatagramTracks(this.version)) { - legacy = this.#held(front, msg.track)?.clone() ?? wireOf(front).liveDatagrams(msg.track); + const live = wireOf(front).liveDatagrams(msg.track); + legacy = live && this.#adopt(live); if (legacy) { legacy.setPriority(msg.priority); datagrams = this.#runDatagrams(legacy, { sub: msg.id, stamp }) @@ -1846,7 +1850,7 @@ export class Publisher { if (hasDatagrams(this.version) && !hasDatagramTracks(this.version)) { datagrams ??= wireOf(front).liveDatagrams(msg.track); } - if (datagrams) this.#holdDatagrams(front, datagrams, hold.signal); + if (datagrams) this.#holdDatagrams(datagrams, hold.signal); if (watch.revoked()) throw unauthorized(msg.broadcast); await info.encode(stream.writer, this.version); console.debug(`track info: broadcast=${msg.broadcast} track=${msg.track}`); @@ -1928,33 +1932,25 @@ export class Publisher { } } - // Hold `consumer` of `front` for the SUBSCRIBE alongside until `hold` aborts. - #holdDatagrams(front: broadcast.Consumer, consumer: datagram.Consumer, hold: AbortSignal) { - let names = this.#heldDatagrams.get(front); - if (!names) { - names = new Map(); - this.#heldDatagrams.set(front, names); - } - let held = names.get(consumer.name); - if (!held) { - held = new Set(); - names.set(consumer.name, held); - } - const set = held; - set.add(consumer); + // Hold `consumer` for the SUBSCRIBE alongside until `hold` aborts. + #holdDatagrams(consumer: datagram.Consumer, hold: AbortSignal) { + consumer.setPriority(-128); + this.#heldDatagrams.add(consumer); const release = () => { - set.delete(consumer); - if (set.size === 0 && names.get(consumer.name) === set) names.delete(consumer.name); + this.#heldDatagrams.delete(consumer); consumer.close(); }; if (hold.aborted) release(); else hold.addEventListener("abort", release, { once: true }); } - // The datagram track `name` of `front` an answered TRACK holds, if any. - #held(front: broadcast.Consumer, name: string): datagram.Consumer | undefined { - const held = this.#heldDatagrams.get(front)?.get(name); - return held?.values().next().value; + // `consumer`, or a clone of the newest hold of the same track, read from that hold's answer on. + #adopt(consumer: datagram.Consumer): datagram.Consumer { + let newest: datagram.Consumer | undefined; + for (const held of this.#heldDatagrams) if (hooks.datagramSameTrack(held, consumer)) newest = held; + if (!newest) return consumer; + consumer.close(); + return newest.clone(); } // Resolve (and cache) a track's immutable TRACK_INFO by asking the application. @@ -2073,7 +2069,7 @@ export class Publisher { stream: Stream, signal: AbortSignal, ): Promise { - const track = this.#held(front, msg.track)?.clone() ?? front.datagrams(msg.track); + const track = this.#adopt(front.datagrams(msg.track)); track.setPriority(msg.priority); const updates = (async () => { for (;;) { diff --git a/rs/moq-net/src/lite/publisher.rs b/rs/moq-net/src/lite/publisher.rs index 60cf165e503..a420cff5e54 100644 --- a/rs/moq-net/src/lite/publisher.rs +++ b/rs/moq-net/src/lite/publisher.rs @@ -2093,7 +2093,7 @@ enum TrackInfoServe { /// datagram track held for the SUBSCRIBE alongside. Track(track::Resolving, Option>, Option), /// Lite-05/06: a name no stream track has, answered for its datagram track. - Datagrams(datagram::Consumer, crate::broadcast::Consumer), + Datagrams(datagram::Consumer), /// Answered from the datagram track, now held for the SUBSCRIBE alongside. Held(#[expect(dead_code, reason = "held for its drop")] DatagramHold), } @@ -2127,10 +2127,10 @@ impl Request for TrackInfoServe { track: &str, broadcast: crate::broadcast::Consumer, ) -> Result { - let legacy = Legacy::new(shared, &broadcast, track); + let legacy = Legacy::new(shared.version, &broadcast, track); match broadcast.track(track) { Ok(track) => Ok(Self::Track(track.info().into_inner(), legacy, None)), - Err(err) => Ok(Self::Datagrams(Legacy::datagrams(legacy, err)?, broadcast)), + Err(err) => Ok(Self::Datagrams(Legacy::datagrams(legacy, err)?)), } } @@ -2147,23 +2147,18 @@ impl Request for TrackInfoServe { let ietf = ready!(shared.peer_setup.poll_ietf(waiter)); if let Self::Track(querying, legacy, _) = self { match querying.poll_ok(waiter) { - Poll::Ready(Err(err)) => { - let broadcast = legacy.as_ref().map(|legacy| legacy.broadcast.clone()); - let consumer = Legacy::datagrams(legacy.take(), err)?; - *self = Self::Datagrams(consumer, broadcast.expect("only a fallback serves datagrams")); - } + Poll::Ready(Err(err)) => *self = Self::Datagrams(Legacy::datagrams(legacy.take(), err)?), Poll::Ready(Ok(_)) => {} Poll::Pending => { - let broadcast = legacy.as_ref().map(|legacy| legacy.broadcast.clone()); if let Poll::Ready(consumer) = Legacy::poll_datagrams(legacy, waiter) { - *self = Self::Datagrams(consumer, broadcast.expect("only a fallback serves datagrams")); + *self = Self::Datagrams(consumer); } } } } let (querying, legacy, held) = match self { Self::Track(querying, legacy, held) => (querying, legacy, held), - Self::Datagrams(consumer, _) => { + Self::Datagrams(consumer) => { let info = ready!(consumer.poll_info(waiter))?; writer.set_priority(crate::coding::TRACK_INFO_SEND_ORDER); writer.buffer(&lite::TrackInfo { @@ -2178,11 +2173,7 @@ impl Request for TrackInfoServe { if ietf { writer.buffer(&lite::ietf::TrackProperties(Default::default()))?; } - let consumer = consumer.clone(); - let Self::Datagrams(_, broadcast) = &*self else { - unreachable!() - }; - *self = Self::Held(shared.held_datagrams.hold(broadcast, consumer)); + *self = Self::Held(shared.held_datagrams.hold(consumer.clone())); return Poll::Ready(Ok(ControlFlow::Break(()))); } Self::Held(_) => unreachable!("answered once"), @@ -2191,10 +2182,10 @@ impl Request for TrackInfoServe { crate::ietf::flate::egress(&info, shared.version.has_compression())?; // The datagrams beside the stream track, read from this answer on. A pending ask goes, // rather than held as long as the requester keeps the stream open. - *held = legacy.take().and_then(|legacy| { - let broadcast = legacy.broadcast.clone(); - Some(shared.held_datagrams.hold(&broadcast, legacy.live()?)) - }); + *held = legacy + .take() + .and_then(|legacy| legacy.live()) + .map(|consumer| shared.held_datagrams.hold(consumer)); // Ahead of the track's groups, which a subscriber holds unread until it lands. writer.set_priority(crate::coding::TRACK_INFO_SEND_ORDER); @@ -2361,7 +2352,8 @@ impl Request for SubscribeServe { track: &str, broadcast: crate::broadcast::Consumer, ) -> Result { - let legacy = Legacy::new(shared, &broadcast, track); + let legacy = + Legacy::new(shared.version, &broadcast, track).map(|legacy| legacy.adopting(&shared.held_datagrams)); let track = match broadcast.track(track) { Ok(track) => track, Err(err) => return Ok(Self::datagrams(Legacy::datagrams(legacy, err)?, &msg, None)), @@ -2662,32 +2654,50 @@ struct Legacy { /// The name's datagram track, asked for beside the stream track: `None` until asked, as /// a relayed broadcast's stays, and `Some(None)` once a handler refused it. datagrams: Option>, + /// A SUBSCRIBE's: the holds its datagram track is read from (see [`HeldDatagrams`]). + held: Option, } impl Legacy { - /// The fallback for a request on the session's version, or `None` when the stream - /// track's answer stands. It starts from the datagram track an answered TRACK of the - /// name holds, if any (see [`HeldDatagrams`]). - fn new(shared: &Shared, broadcast: &crate::broadcast::Consumer, name: &str) -> Option> { - let version = shared.version; + /// The fallback for a request on `version`, or `None` when the stream track's answer stands. + fn new(version: Version, broadcast: &crate::broadcast::Consumer, name: &str) -> Option> { (version.has_datagrams() && !version.has_datagram_tracks()).then(|| { Box::new(Self { broadcast: broadcast.clone(), name: name.to_string(), - datagrams: shared.held_datagrams.find(broadcast, name).map(Some), + datagrams: None, + held: None, }) }) } + /// Serve a SUBSCRIBE: the datagram track it picks is read from the position of a hold of + /// that same track, if any. Checked as it picks, since a pipelined SUBSCRIBE can wait on + /// its stream track while the TRACK beside it is answered. + fn adopting(mut self: Box, held: &HeldDatagrams) -> Box { + self.held = Some(held.clone()); + self + } + + /// `consumer`, or a hold's clone of the same track, read from that hold's answer on. + fn adopt(&self, consumer: datagram::Consumer) -> datagram::Consumer { + self.held + .as_ref() + .and_then(|held| held.find(&consumer)) + .unwrap_or(consumer) + } + /// The name's live datagram track, asking no handler: the request's own ask once /// answered, or one published or already asked for. An unanswered ask is dropped, /// which withdraws it. - fn live(self) -> Option { + fn live(mut self) -> Option { let answered = self .datagrams + .take() .flatten() .filter(|asked| matches!(asked.poll_info(&kio::Waiter::noop()), Poll::Ready(Ok(_)))); - answered.or_else(|| self.broadcast.datagrams_live(&self.name)) + let consumer = answered.or_else(|| self.broadcast.datagrams_live(&self.name))?; + Some(self.adopt(consumer)) } /// The datagram track to serve before the stream track answers: one a local handler or @@ -2715,7 +2725,10 @@ impl Legacy { }; match ready!(consumer.poll_info(waiter)) { Ok(_) if this.broadcast.publishes_track(&this.name) => Poll::Pending, - Ok(_) => Poll::Ready(asked.take().expect("checked above")), + Ok(_) => { + let consumer = asked.take().expect("checked above"); + Poll::Ready(this.adopt(consumer)) + } Err(_) => { *asked = None; Poll::Pending @@ -2726,16 +2739,17 @@ impl Legacy { /// The datagram track to serve instead when the stream track's lookup failed with `err`, /// or `err` itself. fn datagrams(legacy: Option>, err: Error) -> Result { - let Some(legacy) = legacy else { + let Some(mut legacy) = legacy else { return Err(err); }; if !matches!(err, Error::NotFound | Error::Stream(StreamError::NotFound)) { return Err(err); } - let asked = legacy - .datagrams - .unwrap_or_else(|| legacy.broadcast.datagrams(&legacy.name).ok()); - asked.ok_or(err) + let asked = match legacy.datagrams.take() { + Some(asked) => asked, + None => legacy.broadcast.datagrams(&legacy.name).ok(), + }; + Ok(legacy.adopt(asked.ok_or(err)?)) } } @@ -2745,38 +2759,55 @@ impl Legacy { /// later. That SUBSCRIBE reads from a clone of the hold, keeping its position, rather than /// from its own arrival, which would lose what was written in between. The requester keeps /// its TRACK open until the SUBSCRIBE has a response, so the hold outlasts its arrival. +/// +/// A hold is matched by the track it reads, not by name, so one left on a track a route +/// change replaced never stands in for the current one. It sits at the lowest priority, so +/// it keeps a position without outranking the subscriptions that read the track. #[derive(Clone, Default)] -struct HeldDatagrams(Arc>>); +struct HeldDatagrams(Arc>); + +#[derive(Default)] +struct HeldState { + next: u64, + /// Oldest first, by key. + holds: Vec<(u64, datagram::Consumer)>, +} impl HeldDatagrams { - /// Hold `consumer` of `broadcast` until the returned hold drops. - fn hold(&self, broadcast: &crate::broadcast::Consumer, consumer: datagram::Consumer) -> DatagramHold { - let key = self.0.lock().unwrap().insert((broadcast.clone(), consumer)); + /// Hold `consumer` until the returned hold drops. + fn hold(&self, mut consumer: datagram::Consumer) -> DatagramHold { + consumer.set_priority(i8::MIN); + let mut state = self.0.lock().unwrap(); + let key = state.next; + state.next += 1; + state.holds.push((key, consumer)); DatagramHold { held: self.clone(), key, } } - /// A clone of the datagram track `name` of `broadcast` a TRACK holds, if any. - fn find(&self, broadcast: &crate::broadcast::Consumer, name: &str) -> Option { - let held = self.0.lock().unwrap(); - held.iter() - .find(|(_, (held, consumer))| consumer.name() == name && held.is_clone(broadcast)) - .map(|(_, (_, consumer))| consumer.clone()) + /// A clone of the newest hold reading the same track as `consumer`, if any. + fn find(&self, consumer: &datagram::Consumer) -> Option { + let state = self.0.lock().unwrap(); + let (_, held) = state.holds.iter().rev().find(|(_, held)| held.is_clone(consumer))?; + Some(held.clone()) } } /// One answered TRACK's entry in [`HeldDatagrams`]. struct DatagramHold { held: HeldDatagrams, - key: usize, + key: u64, } impl Drop for DatagramHold { fn drop(&mut self) { + let mut state = self.held.0.lock().unwrap(); + let index = state.holds.iter().position(|(key, _)| *key == self.key); + let consumer = index.map(|index| state.holds.remove(index)); // Dropped outside the lock. - let consumer = self.held.0.lock().unwrap().remove(self.key); + drop(state); drop(consumer); } } @@ -9469,10 +9500,81 @@ mod tests { assert!(crate::coding::TRACK_INFO_SEND_ORDER > top); } + /// A lite-05/06 publisher serving `origin`, with the log of the datagrams it sends. + fn legacy_publisher( + version: Version, + origin: &crate::origin::Producer, + ) -> (Publisher, Arc>>) { + let peer_setup = crate::lite::PeerSetup::new(false, crate::transport::Address::Setup); + peer_setup.set(crate::lite::Setup::default()).unwrap(); + let (_, goaway) = crate::goaway::Handle::new(true); + let sink = SinkSession::new(crate::lite::test_transport::Log::default()).with_datagrams(1200); + let log = sink.log.datagrams.clone(); + let publisher = Publisher::new(PublisherConfig { + estimate: kio::Producer::default().weak(), + runtime: crate::time::sim(), + session: crate::driver::Transport::new(sink), + origin: origin.consume(), + version, + peer_setup, + goaway, + peer_hop: None, + subscriptions: Default::default(), + auth: crate::auth::Link::new(false), + peer_grant: Default::default(), + client: false, + }); + (publisher, log) + } + + /// Serve `msg` as the first message of a request stream on `publisher`. + async fn legacy_request( + publisher: &Publisher, + version: Version, + msg: &impl Encode, + ) -> RequestServe { + let mut script = Vec::new(); + msg.encode(&mut crate::coding::Encoder::new(&mut script, version.into()), version) + .unwrap(); + let session = ScriptedSession::new(script); + let (send, recv) = kio::wait(|waiter| { + ::poll_open_bi(&session, waiter, Default::default()) + }) + .await + .unwrap(); + let stream = Stream:: { + writer: Writer::new(crate::transport::boxed::SendStream::new(send), version), + reader: crate::coding::Reader::new(crate::transport::boxed::RecvStream::new(recv), version), + }; + RequestServe::new(publisher.shared.clone(), stream) + } + + fn legacy_track() -> lite::Track<'static> { + lite::Track { + instance: None, + broadcast: crate::Path::from_static("room"), + track: "events".into(), + } + } + + fn legacy_subscribe(priority: i8) -> lite::Subscribe<'static> { + lite::Subscribe { + order: crate::track::Order::Descending, + instance: None, + id: 0, + broadcast: crate::Path::from_static("room"), + track: "events".into(), + priority, + max_delay: Duration::ZERO, + min_delay: None, + ranges: vec![lite::Range::span(None, 0, None, None)], + } + } + /// Lite-05/06: a requester reads TRACK_INFO as the name being accepted, and its SUBSCRIBE, /// on a stream of its own, can arrive after a datagram written meanwhile. That SUBSCRIBE /// still sends it, whether the datagram track answered the TRACK or rides beside a stream - /// track of the same name. + /// track of the same name, and the hold keeps no priority of its own. #[kio::test] async fn a_subscribe_sends_the_datagrams_written_since_its_track_info() { for version in [Version::Lite05, Version::Lite06] { @@ -9491,77 +9593,90 @@ mod tests { .unwrap() }); broadcast.announce().unwrap(); + let (publisher, log) = legacy_publisher(version, &origin); - let peer_setup = crate::lite::PeerSetup::new(false, crate::transport::Address::Setup); - peer_setup.set(crate::lite::Setup::default()).unwrap(); - let (_, goaway) = crate::goaway::Handle::new(true); - let sink = SinkSession::new(crate::lite::test_transport::Log::default()).with_datagrams(1200); - let log = sink.log.clone(); - let publisher = Publisher::new(PublisherConfig { - estimate: kio::Producer::default().weak(), - runtime: crate::time::sim(), - session: crate::driver::Transport::new(sink), - origin: origin.consume(), - version, - peer_setup, - goaway, - peer_hop: None, - subscriptions: Default::default(), - auth: crate::auth::Link::new(false), - peer_grant: Default::default(), - client: false, - }); - let open = |script: Vec| async move { - let session = ScriptedSession::new(script); - let (send, recv) = kio::wait(|waiter| { - ::poll_open_bi( - &session, - waiter, - Default::default(), - ) - }) - .await - .unwrap(); - Stream:: { - writer: Writer::new(crate::transport::boxed::SendStream::new(send), version), - reader: crate::coding::Reader::new(crate::transport::boxed::RecvStream::new(recv), version), - } - }; - - let mut script = Vec::new(); - lite::Track { - instance: None, - broadcast: crate::Path::from_static("room"), - track: "events".into(), - } - .encode(&mut crate::coding::Encoder::new(&mut script, version.into()), version) - .unwrap(); - let mut track_info = RequestServe::::new(publisher.shared.clone(), open(script).await); + let mut track_info = legacy_request::(&publisher, version, &legacy_track()).await; assert!(!drive(&mut track_info).await, "{case}: the TRACK stream is held"); datagrams.append(None, b"between".as_slice()).unwrap(); - let mut script = Vec::new(); - lite::Subscribe { - order: crate::track::Order::Descending, - instance: None, - id: 0, - broadcast: crate::Path::from_static("room"), - track: "events".into(), - priority: 0, - max_delay: Duration::ZERO, - min_delay: None, - ranges: vec![lite::Range::span(None, 0, None, None)], - } - .encode(&mut crate::coding::Encoder::new(&mut script, version.into()), version) - .unwrap(); - let mut subscribe = RequestServe::::new(publisher.shared.clone(), open(script).await); + let mut subscribe = legacy_request::(&publisher, version, &legacy_subscribe(-5)).await; assert!(!drive(&mut subscribe).await, "{case}: the subscription is live"); - assert_eq!(log.datagrams.lock().unwrap().len(), 1, "{case}"); + assert_eq!(log.lock().unwrap().len(), 1, "{case}"); + assert_eq!( + datagrams.priority(), + Some(-5), + "{case}: the SUBSCRIBE's priority stands" + ); } } } + /// Lite-05/06: a SUBSCRIBE sent alongside its TRACK can arrive first and wait on a + /// relayed stream track's info. Once the TRACK is answered and a datagram written, the + /// waiting SUBSCRIBE still sends that datagram. + #[kio::test] + async fn a_waiting_subscribe_sends_the_datagrams_written_since_its_track_info() { + for version in [Version::Lite05, Version::Lite06] { + let origin = crate::origin::Config::new(crate::Hop::new(1).unwrap()).produce(); + let broadcast = origin.route("room", crate::origin::RouteOptions::hidden()).unwrap(); + let _relayed = crate::model::broadcast::SourceGuard::new(broadcast.broadcast().clone()); + let mut datagrams = broadcast + .broadcast() + .create_datagrams("events", datagram::Info::default()) + .unwrap(); + let reserved = broadcast.broadcast().reserve_track("events").unwrap(); + broadcast.announce().unwrap(); + let (publisher, log) = legacy_publisher(version, &origin); + + let mut subscribe = legacy_request::(&publisher, version, &legacy_subscribe(0)).await; + assert!(!drive(&mut subscribe).await, "{version:?}: waits on the track's info"); + let mut track_info = legacy_request::(&publisher, version, &legacy_track()).await; + assert!(!drive(&mut track_info).await, "{version:?}: waits on the track's info"); + + let _track = reserved.accept(crate::track::Info::timed()); + assert!(!drive(&mut track_info).await, "{version:?}: the TRACK stream is held"); + datagrams.append(None, b"between".as_slice()).unwrap(); + assert!(!drive(&mut subscribe).await, "{version:?}: the subscription is live"); + assert_eq!(log.lock().unwrap().len(), 1, "{version:?}"); + } + } + + /// Lite-05/06: a hold reads the datagram track that answered its TRACK. Once that track + /// is replaced, a SUBSCRIBE reads the replacement, even while the older TRACK stays open. + #[kio::test] + async fn a_subscribe_reads_the_current_datagram_track_past_an_older_hold() { + for version in [Version::Lite05, Version::Lite06] { + let origin = crate::origin::Config::new(crate::Hop::new(1).unwrap()).produce(); + let broadcast = origin.route("room", crate::origin::RouteOptions::hidden()).unwrap(); + let mut first = broadcast + .broadcast() + .create_datagrams("events", datagram::Info::default()) + .unwrap(); + broadcast.announce().unwrap(); + let (publisher, log) = legacy_publisher(version, &origin); + + let mut older = legacy_request::(&publisher, version, &legacy_track()).await; + assert!(!drive(&mut older).await, "{version:?}: the older TRACK stream is held"); + first.append(None, b"first".as_slice()).unwrap(); + first.finish(); + + let mut current = broadcast + .broadcast() + .create_datagrams("events", datagram::Info::default()) + .unwrap(); + let mut track_info = legacy_request::(&publisher, version, &legacy_track()).await; + assert!(!drive(&mut track_info).await, "{version:?}: the TRACK stream is held"); + current.append(None, b"current".as_slice()).unwrap(); + + let mut subscribe = legacy_request::(&publisher, version, &legacy_subscribe(0)).await; + assert!(!drive(&mut subscribe).await, "{version:?}: the subscription is live"); + let sent = log.lock().unwrap().clone(); + assert_eq!(sent.len(), 1, "{version:?}"); + assert!(sent[0].ends_with(b"current"), "{version:?}: {sent:?}"); + } + } + /// An answered TRACK stream holds its track until the requester closes its side, by /// FIN or reset, and the hold owes nothing a draining close would wait for. A requester /// that leaves while the reply is still blocked on flow control lets the track go too. From 3cef3b40e0d7f0ad57a5eec6a2123b4541b9bec4 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 6 Nov 2026 08:40:40 -0800 Subject: [PATCH 7/8] style(net): format the passive hold Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/lite/publisher.rs | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/rs/moq-net/src/lite/publisher.rs b/rs/moq-net/src/lite/publisher.rs index 20659bfa3e3..b1ea0570276 100644 --- a/rs/moq-net/src/lite/publisher.rs +++ b/rs/moq-net/src/lite/publisher.rs @@ -9618,7 +9618,11 @@ mod tests { assert_eq!(log.lock().unwrap().len(), 1, "{case}"); // Beside a stream track, only the SUBSCRIBE asks for the datagrams. if stream { - assert_eq!(datagrams.priority(), Some(-5), "{case}: the SUBSCRIBE's priority stands"); + assert_eq!( + datagrams.priority(), + Some(-5), + "{case}: the SUBSCRIBE's priority stands" + ); } } } From 47484f0b37115e9e6410d39d3a0ed856e6ade547 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 6 Nov 2026 09:54:20 -0800 Subject: [PATCH 8/8] fix(net): a SUBSCRIBE adopts the oldest matching datagram hold With two TRACKs of one track open, the newest hold skipped datagrams written between the two answers. Read from the earliest answer instead; a replay only duplicates. Co-Authored-By: Claude Opus 5.5 --- js/net/src/lite/publisher.test.ts | 16 ++++++++++++++++ js/net/src/lite/publisher.ts | 15 +++++++++------ rs/moq-net/src/lite/publisher.rs | 32 +++++++++++++++++++++++++++++-- 3 files changed, 55 insertions(+), 8 deletions(-) diff --git a/js/net/src/lite/publisher.test.ts b/js/net/src/lite/publisher.test.ts index 23ca02a5fa0..915f281dd3d 100644 --- a/js/net/src/lite/publisher.test.ts +++ b/js/net/src/lite/publisher.test.ts @@ -3478,6 +3478,22 @@ for (const version of [Version.DRAFT_05, Version.DRAFT_06]) { }); } + // With two TRACKs of one name open, a SUBSCRIBE reads from the older answer, so a datagram + // written between the two still reaches the SUBSCRIBE of the first. + test(`${label}: a SUBSCRIBE reads from the oldest hold`, async () => { + const legacy = legacyPublisher(version); + try { + const datagrams = legacy.broadcast.createDatagrams("events"); + await legacy.track(); + datagrams.append(undefined, Uint8Array.of(0xaa)); + await legacy.track(); + await legacy.subscribe(0); + expect(legacy.sent.length).toBe(1); + } finally { + legacy.close(); + } + }); + // A hold reads the datagram track that answered its TRACK. Once that track is replaced, a // SUBSCRIBE reads the replacement, even while the older TRACK stays open. test(`${label}: a SUBSCRIBE reads the current datagram track past an older hold`, async () => { diff --git a/js/net/src/lite/publisher.ts b/js/net/src/lite/publisher.ts index 12eb0613bba..f28663680fe 100644 --- a/js/net/src/lite/publisher.ts +++ b/js/net/src/lite/publisher.ts @@ -1948,13 +1948,16 @@ export class Publisher { else hold.addEventListener("abort", release, { once: true }); } - // `consumer`, or a clone of the newest hold of the same track, read from that hold's answer on. + // `consumer`, or a clone of the oldest hold of the same track, read from that hold's answer on. + // Several TRACKs of one track may be open, and the SUBSCRIBE could follow any of them, so it + // reads from the earliest answer: a replay only duplicates, a later one would lose. #adopt(consumer: datagram.Consumer): datagram.Consumer { - let newest: datagram.Consumer | undefined; - for (const held of this.#heldDatagrams) if (hooks.datagramSameTrack(held, consumer)) newest = held; - if (!newest) return consumer; - consumer.close(); - return newest.clone(); + for (const held of this.#heldDatagrams) { + if (!hooks.datagramSameTrack(held, consumer)) continue; + consumer.close(); + return held.clone(); + } + return consumer; } // Resolve (and cache) a track's immutable TRACK_INFO by asking the application. diff --git a/rs/moq-net/src/lite/publisher.rs b/rs/moq-net/src/lite/publisher.rs index b1ea0570276..2d1bd4b46f1 100644 --- a/rs/moq-net/src/lite/publisher.rs +++ b/rs/moq-net/src/lite/publisher.rs @@ -2792,10 +2792,12 @@ impl HeldDatagrams { } } - /// A clone of the newest hold reading the same track as `consumer`, if any. + /// A clone of the oldest hold reading the same track as `consumer`, if any. Several + /// TRACKs of one track may be open, and the SUBSCRIBE could follow any of them, so it + /// reads from the earliest answer: a replay only duplicates, a later one would lose. fn find(&self, consumer: &datagram::Consumer) -> Option { let state = self.0.lock().unwrap(); - let (_, held) = state.holds.iter().rev().find(|(_, held)| held.is_clone(consumer))?; + let (_, held) = state.holds.iter().find(|(_, held)| held.is_clone(consumer))?; Some(held.clone()) } } @@ -9658,6 +9660,32 @@ mod tests { } } + /// Lite-05/06: with two TRACKs of one name open, a SUBSCRIBE reads from the older answer, + /// so a datagram written between the two still reaches the SUBSCRIBE of the first. + #[kio::test] + async fn a_subscribe_reads_from_the_oldest_hold() { + for version in [Version::Lite05, Version::Lite06] { + let origin = crate::origin::Config::new(crate::Hop::new(1).unwrap()).produce(); + let broadcast = origin.route("room", crate::origin::RouteOptions::hidden()).unwrap(); + let mut datagrams = broadcast + .broadcast() + .create_datagrams("events", datagram::Info::default()) + .unwrap(); + broadcast.announce().unwrap(); + let (publisher, log) = legacy_publisher(version, &origin); + + let mut older = legacy_request::(&publisher, version, &legacy_track()).await; + assert!(!drive(&mut older).await, "{version:?}: the older TRACK stream is held"); + datagrams.append(None, b"between".as_slice()).unwrap(); + let mut newer = legacy_request::(&publisher, version, &legacy_track()).await; + assert!(!drive(&mut newer).await, "{version:?}: the newer TRACK stream is held"); + + let mut subscribe = legacy_request::(&publisher, version, &legacy_subscribe(0)).await; + assert!(!drive(&mut subscribe).await, "{version:?}: the subscription is live"); + assert_eq!(log.lock().unwrap().len(), 1, "{version:?}"); + } + } + /// Lite-05/06: a hold reads the datagram track that answered its TRACK. Once that track /// is replaced, a SUBSCRIBE reads the replacement, even while the older TRACK stays open. #[kio::test]