diff --git a/js/net/src/datagram.ts b/js/net/src/datagram.ts index ee570c3cf48..0f7cdde831f 100644 --- a/js/net/src/datagram.ts +++ b/js/net/src/datagram.ts @@ -246,6 +246,11 @@ 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; + hooks.observeDatagramConsumer = (consumer) => { + if (consumer.#closed.peek() !== undefined) throw new Error("datagram consumer is closed"); + return consume(consumer.name, consumer.#state, consumer.#cursor, null); + }; } /** The track's publisher properties, once accepted. */ diff --git a/js/net/src/internal.ts b/js/net/src/internal.ts index 6a9dd799bed..c9df2e06cc0 100644 --- a/js/net/src/internal.ts +++ b/js/net/src/internal.ts @@ -459,6 +459,10 @@ export const hooks: { observeDatagrams: (producer: datagram.Producer) => datagram.Consumer; /** 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; + /** Another consumer at the same position that adds no demand, until its priority is set. */ + observeDatagramConsumer: (consumer: datagram.Consumer) => datagram.Consumer; /** * 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 @@ -783,6 +787,12 @@ export const hooks: { datagramClosed: () => { throw new Error("datagram module not loaded"); }, + datagramSameTrack: () => { + throw new Error("datagram module not loaded"); + }, + observeDatagramConsumer: () => { + throw new Error("datagram module not loaded"); + }, endTrack: () => { throw new Error("track.ts not loaded"); }, 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(); diff --git a/js/net/src/lite/publisher.test.ts b/js/net/src/lite/publisher.test.ts index eae0748cf2f..915f281dd3d 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,113 @@ test("lite-06: a datagram request failing behind the shared writer resets rather pair.server.close(); } }); + +// 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, +// and the hold adds no demand 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]) { + test(`${label}, stream track ${stream}: a SUBSCRIBE sends the datagrams written since its TRACK_INFO`, async () => { + const legacy = legacyPublisher(version); + try { + 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); + // Beside a stream track, only the SUBSCRIBE asks for the datagrams. + if (stream) expect(datagrams.priority.peek()).toBe(-5); + } finally { + legacy.close(); + } + }); + } + + // 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 () => { + 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 b78da9de6fc..f28663680fe 100644 --- a/js/net/src/lite/publisher.ts +++ b/js/net/src/lite/publisher.ts @@ -542,6 +542,17 @@ 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, 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 is a passive observer: it keeps a position without adding + // demand, which the SUBSCRIBE adds when it reads the track. + #heldDatagrams = new Set(); + /** * Creates a new Publisher instance. * @param quic - The WebTransport session to use @@ -1086,7 +1097,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 = wireOf(front).liveDatagrams(msg.track); + const live = wireOf(front).liveDatagrams(msg.track); + legacy = live && this.#adopt(live); if (legacy) { if (wireOf(front).localDatagrams) legacy.setPriority(msg.priority); datagrams = this.#runDatagrams(legacy, { sub: msg.id, stamp, priority: legacyPriority }) @@ -1816,6 +1828,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) { @@ -1825,13 +1838,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(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}`); @@ -1913,6 +1934,32 @@ export class Publisher { } } + // Hold `consumer`'s track from its position for the SUBSCRIBE alongside until `hold` + // aborts, as a passive observer in place of `consumer`. + #holdDatagrams(consumer: datagram.Consumer, hold: AbortSignal) { + const held = hooks.observeDatagramConsumer(consumer); + consumer.close(); + this.#heldDatagrams.add(held); + const release = () => { + this.#heldDatagrams.delete(held); + held.close(); + }; + if (hold.aborted) release(); + else hold.addEventListener("abort", release, { once: true }); + } + + // `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 { + 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. // `resolveTrackInfo` triggers a TrackRequest the app answers with accept(TrackInfo); // only the immutable properties are needed (not the groups). Cached because they're @@ -2029,7 +2076,7 @@ export class Publisher { stream: Stream, signal: AbortSignal, ): Promise { - const track = front.datagrams(msg.track); + const track = this.#adopt(front.datagrams(msg.track)); track.setPriority(msg.priority); const updates = (async () => { for (;;) { diff --git a/quest/m0/test-flakes-2/README.md b/quest/m0/test-flakes-2/README.md index 9475129619c..03351e66c46 100644 --- a/quest/m0/test-flakes-2/README.md +++ b/quest/m0/test-flakes-2/README.md @@ -109,7 +109,6 @@ Public API: none. Wire: none. - [The draft-22 wider-fill refusal test is deterministic](/quest/m0/test-flakes-2/ietf-wider-fill.md) - fix its intermittent live-group reset and re-enable both ordering cases without weakening their assertions - [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 - [moq-net's pool-debt flight test holds its bound in parallel runs](/quest/m0/test-flakes-2/pool-debt-flight.md) - `pool_debt_sheds_a_flight_without_its_cached_group` stays under `2 * POOL` when it shares a process with moq-net's other tests 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 525d707ba2b..00000000000 --- a/quest/m0/test-flakes-2/ffi-datagram-roundtrip.md +++ /dev/null @@ -1,66 +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. No retry or longer wait was added. -- `requested_datagrams_cross_a_default_connection` is quarantined too, by - [moq-dev/moq#7904](https://github.com/moq-dev/moq/pull/7904): its single - datagram missed the 10 s wait in that PR's - [Check](https://github.com/moq-dev/moq/actions/runs/42076855917/job/127419261608) - (panic at `rs/moq-ffi/src/test.rs:6111`), the same symptom as the plain - test, then passed 100 of 100 stress runs on an idle host. Re-enable it with - the plain test once this quest's cause is fixed; its other requested - siblings stay enabled. - -## 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 7a44fa1c5e5..022622dad08 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 @@ -6029,7 +6028,6 @@ async fn datagrams_cross_a_default_connection() { /// An on-demand datagram-only track reaches a reader on the default lite-06 path while a /// stream handler leaves the name unanswered: the datagram handler's answer serves it. #[tokio::test] -#[ignore = "flaky under CI load; tracked in quest/m0/test-flakes-2/ffi-datagram-roundtrip.md"] async fn requested_datagrams_cross_a_default_connection() { requested_datagrams_cross(vec![], "moq-lite-06").await; } diff --git a/rs/moq-net/src/lite/publisher.rs b/rs/moq-net/src/lite/publisher.rs index b4ff91c4a1d..2d1bd4b46f1 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, @@ -2086,9 +2089,17 @@ 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), + /// Answered from the datagram track, which stays asked for while held for the SUBSCRIBE + /// alongside. + Held( + #[expect(dead_code, reason = "held for its drop")] datagram::Consumer, + #[expect(dead_code, reason = "held for its drop")] DatagramHold, + ), } impl Request for TrackInfoServe { @@ -2122,7 +2133,7 @@ impl Request for TrackInfoServe { ) -> Result { let legacy = Legacy::new(shared.version, &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)?)), } } @@ -2138,12 +2149,10 @@ 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)) => *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(Ok(_)) => {} Poll::Pending => { if let Poll::Ready(consumer) = Legacy::poll_datagrams(legacy, waiter) { *self = Self::Datagrams(consumer); @@ -2151,8 +2160,8 @@ impl Request for TrackInfoServe { } } } - let querying = match self { - Self::Track(querying, _) => querying, + 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); @@ -2168,11 +2177,20 @@ impl Request for TrackInfoServe { if ietf { writer.buffer(&lite::ietf::TrackProperties(Default::default()))?; } + let hold = shared.held_datagrams.hold(consumer); + *self = Self::Held(consumer.clone(), hold); return Poll::Ready(Ok(ControlFlow::Break(()))); } + 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. + *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); @@ -2339,7 +2357,8 @@ impl Request for SubscribeServe { track: &str, broadcast: crate::broadcast::Consumer, ) -> Result { - let legacy = Legacy::new(shared.version, &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)), @@ -2640,6 +2659,8 @@ 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 { @@ -2650,10 +2671,40 @@ impl Legacy { broadcast: broadcast.clone(), name: name.to_string(), 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(mut self) -> Option { + let answered = self + .datagrams + .take() + .flatten() + .filter(|asked| matches!(asked.poll_info(&kio::Waiter::noop()), Poll::Ready(Ok(_)))); + 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 /// producer answered while no producer serves the stream track. Pending otherwise, /// waking with the stream track's lookup, which the caller polls first. @@ -2679,7 +2730,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 @@ -2690,16 +2744,78 @@ 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)?)) + } +} + +/// 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. +/// +/// 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 is a passive observer: it keeps +/// a position without adding demand, which the SUBSCRIBE adds when it reads the track. +#[derive(Clone, Default)] +struct HeldDatagrams(Arc>); + +#[derive(Default)] +struct HeldState { + next: u64, + /// Oldest first, by key. + holds: Vec<(u64, datagram::Consumer)>, +} + +impl HeldDatagrams { + /// Hold `consumer`'s track from its position until the returned hold drops. + fn hold(&self, consumer: &datagram::Consumer) -> DatagramHold { + let consumer = consumer.observe(); + 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 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().find(|(_, held)| held.is_clone(consumer))?; + Some(held.clone()) + } +} + +/// One answered TRACK's entry in [`HeldDatagrams`]. +struct DatagramHold { + held: HeldDatagrams, + 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. + drop(state); + drop(consumer); } } @@ -2724,17 +2840,9 @@ 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 demand = broadcast.has_local_datagrams(); + let legacy = legacy?; + let demand = legacy.broadcast.has_local_datagrams(); + let mut consumer = legacy.live()?; if demand { consumer.set_priority(priority); } @@ -9407,6 +9515,212 @@ 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. + #[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 (publisher, log) = legacy_publisher(version, &origin); + + 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 subscribe = legacy_request::(&publisher, version, &legacy_subscribe(-5)).await; + assert!(!drive(&mut subscribe).await, "{case}: the subscription is live"); + 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" + ); + } + } + } + } + + /// 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: 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] + 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. diff --git a/rs/moq-net/src/model/datagram.rs b/rs/moq-net/src/model/datagram.rs index 8b44002e0f4..08055a35759 100644 --- a/rs/moq-net/src/model/datagram.rs +++ b/rs/moq-net/src/model/datagram.rs @@ -586,6 +586,19 @@ impl Consumer { } } + /// Another consumer at the same position that adds no demand, until its priority is set. + #[cfg_attr(not(feature = "async"), allow(dead_code, reason = "async-only"))] + pub(crate) fn observe(&self) -> Self { + let slot = self.slot.map(|_| self.state.lock().priorities.insert(None)); + Self { + name: self.name.clone(), + state: self.state.clone(), + slot, + cursor: self.cursor, + stats: self.stats.clone(), + } + } + /// Whether this reads the same track as `other`. pub fn is_clone(&self, other: &Self) -> bool { self.state.same_channel(&other.state) diff --git a/sh/gh/test-inventory.allow b/sh/gh/test-inventory.allow index 45e9fe71542..6c56be00373 100644 --- a/sh/gh/test-inventory.allow +++ b/sh/gh/test-inventory.allow @@ -3,9 +3,7 @@ # 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-ffi::test::concurrent_ready_calls_all_wake /quest/m0/test-flakes-2/ffi-concurrent-ready.md -ignore moq-ffi::test::requested_datagrams_cross_a_default_connection /quest/m0/test-flakes-2/ffi-datagram-roundtrip.md # Need a GPU, a desktop, or a device that no CI runner has. ignore moq-carla-bridge::gpu_video_reaches_the_worker /quest/m5/carla-bridge.md