fix(net): a lite-05/06 SUBSCRIBE sends datagrams written since its TRACK_INFO

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. */

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<Error | null>;
+	/** 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");
 	},

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();

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<Uint8Array>({
+			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();
+		}
+	});
+}

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<broadcast.Consumer, Map<string, TrackInfoEntry>>();
 
+	// 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<datagram.Consumer>();
+
 	/**
 	 * 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<void> {
-		const track = front.datagrams(msg.track);
+		const track = this.#adopt(front.datagrams(msg.track));
 		track.setPriority(msg.priority);
 		const updates = (async () => {
 			for (;;) {

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

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;
 }