fix(net): wait for subscription-owed groups in JS ordered reads

js/net/MIGRATE.md

@@ -4,6 +4,11 @@ How to update your code for each breaking change, newest first.
 
 ## Unreleased
 
+### Ordered subscriptions wait for owed groups
+
+An ordered reader on a subscribed track now waits at a missing group while the feed may still deliver it, up to `maxDelay` by media time or wall clock. Local tracks still skip gaps, and the zero default waits for nothing.
+With `maxDelay: null`, there is no subscriber deadline. Use a finite budget for bounded waiting: a moq-transport SUBSCRIBE signals no gaps, so an unlimited wait can last until the track ends. Publisher retention still bounds the budget.
+
 ### `maxDelay: null` sets no subscriber limit ([#7403](https://github.com/moq-dev/moq/pull/7403))
 
 A `Track.Subscription` whose `maxDelay` is `null` reaches back as far as the publisher's `Track.Info.maxAge` keeps, as `None` does in Rust. An omitted or `undefined` `maxDelay` is still zero.

js/net/bench/drift.ts

@@ -1,6 +1,7 @@
 /** Sweep publishers, readers, and retained history for a parked frame read's drift check. */
 import { Manual, Milli } from "@moq/clock";
 import { type Group, Time, Track } from "@moq/net";
+import { hooks } from "../src/internal.ts";
 
 /** Read live frames after a retained window, checking every delivered payload. */
 export async function run(): Promise<void> {
@@ -42,8 +43,10 @@ export async function run(): Promise<void> {
 						);
 						const sources: Group.Producer[] = [];
 						const consumers: Group.Consumer[] = [];
+						const waiting: Track.Ordered[] = [];
 						try {
 							for (const track of tracks) {
+								hooks.receiveTrack(track);
 								for (let sequence = 0; sequence < retained; sequence++) {
 									const group = track.appendGroup();
 									group.writeFrame({
@@ -73,6 +76,10 @@ export async function run(): Promise<void> {
 									if (!group) throw new Error("live group missing");
 									consumers.push(group);
 								}
+								// A frameless candidate beyond one owed group keeps the ordered
+								// cursor parked while frames advance its presentation edge.
+								track.createGroup(retained + 2);
+								waiting.push(...ordered);
 							}
 							let start = 0;
 							for (let iteration = 0; iteration < iterations + warmup; iteration++) {
@@ -85,6 +92,8 @@ export async function run(): Promise<void> {
 									iteration,
 								);
 								for (const source of sources) source.writeFrame({ payload, timestamp });
+								for (const reader of waiting)
+									if (reader.tryNextGroup()) throw new Error("an owed hole was skipped");
 								for (const frame of await Promise.all(pending))
 									if (
 										frame?.payload[0] !== 7 ||

js/net/src/ietf/subscriber.test.ts

@@ -1260,9 +1260,9 @@ test.each([
 		new Reader(undefined, encodeObjects([0]), version),
 	);
 
-	const ordered = track.ordered();
-	for (const sequence of expected) expect((await ordered.nextGroup())?.sequence).toBe(sequence);
-	ordered.close();
+	// This tests stream framing, independent of gaps below these groups. An ordered
+	// subscription with this budget would wait for the unreceived groups 0..2.
+	for (const sequence of expected) expect((await track.recvGroup())?.sequence).toBe(sequence);
 	track.close();
 });
 

js/net/src/ietf/subscriber.ts

@@ -1263,6 +1263,7 @@ export class Subscriber {
 			ietf: version,
 		});
 		const producer = staged.producer;
+		hooks.receiveTrack(producer);
 		const tail = new Tail();
 		const subscription: Subscription = { track: producer, tail, timescale: ok.properties.timescale, fetchIds: [] };
 		// The subscription starts after Largest, so the joining FETCH or the fill brings its

js/net/src/internal.ts

@@ -410,6 +410,8 @@ export const hooks: {
 	 * the drop. Mirrors the Rust `drop_groups`; assigned by `track.ts`.
 	 */
 	dropTrackGroups: (producer: Producer, start: number, end: number, error: Error) => void;
+	/** Mark a track as fed by a wire subscription, which owes missing groups. Assigned by `track.ts`. */
+	receiveTrack: (producer: Producer) => void;
 	/** Create a received group at its wire offset before publishing it to readers. Assigned by `track.ts`. */
 	receiveGroup: (producer: Producer, group: { sequence: number; start: number }) => GroupProducer;
 	/** The producer's largest reported location and its subsequent changes. */
@@ -485,10 +487,10 @@ export const hooks: {
 	 * kept for it, so a resumed reader takes groups the plain way and drops from here.
 	 */
 	tryDrop: (subscriber: Subscriber) => (Recv & { kind: "drop" }) | undefined;
-	/** The sequence {@link tryNextGroup} (`ordered`) or {@link tryRecvGroup} would take next, without taking it. */
-	peekGroup: (subscriber: Subscriber, ordered: boolean) => number | undefined;
+	/** The lowest buffered candidate, which an ordered take may wait below; `floor` passes sequences taken on other copies. */
+	peekGroup: (subscriber: Subscriber, ordered: boolean, floor?: number) => number | undefined;
 	/** Take the next group in sequence order without waiting, as {@link tryRecvGroup} does in arrival order. */
-	tryNextGroup: (subscriber: Subscriber) => Recv;
+	tryNextGroup: (subscriber: Subscriber, floor?: number) => Recv;
 	/** Wake once a subscriber's datagram cursor may read differently; fires at most once. */
 	datagramChanged: (subscriber: Subscriber, fn: () => void) => Dispose;
 	/** Take the next datagram in arrival order without waiting. */
@@ -766,6 +768,9 @@ export const hooks: {
 	bindGroupCompression: () => {
 		throw new Error("group.ts not loaded");
 	},
+	receiveTrack: () => {
+		throw new Error("track.ts not loaded");
+	},
 	receiveGroup: () => {
 		throw new Error("track.ts not loaded");
 	},

js/net/src/lite/subscriber.ts

@@ -1163,6 +1163,7 @@ export class Subscriber {
 		request: track.Request,
 		state: SubscribeSetup,
 	): SubscribeEntry {
+		hooks.receiveTrack(producer);
 		// Lite-06+ resolves where the feed starts only with SUBSCRIBE_START, so a relay serving
 		// this copy holds its first group until the answer arrives.
 		if (resolvesStart(this.version)) hooks.startTrack(producer, { requested: msg.startGroup });

js/net/src/lite/tail.test.ts

@@ -197,20 +197,25 @@ describe.each([Version.DRAFT_05, Version.DRAFT_06, Version.DRAFT_07])("%s", (ver
 			await fin();
 
 			// Group 0 has no header, so accounting still waits, and its stream may yet
-			// land behind group 1's. The arrival reader holds at the hole; the sequence
-			// reader is already past it and ends at the boundary.
+			// land behind group 1's. Both readers wait, with the ordered reader holding
+			// group 1 until its budget gives up on the owed hole.
 			expect((await reader.recvGroup())?.sequence).toBe(1);
 			let ended = false;
 			const held = reader.recvGroup().finally(() => (ended = true));
-			expect(await ordered.readString()).toBe("1.0");
-			expect(await ordered.readString()).toBeUndefined();
+			const orderedHeld = ordered.readString();
+			let orderedReady = false;
+			void orderedHeld.then(() => (orderedReady = true));
+			await settle();
+			expect(orderedReady).toBe(false);
 			await settle();
 			expect(ended).toBe(false);
 			expect(reader.closed.peek()).toBeUndefined();
 			expect(hooks.subscriberEnd(reader)).toEqual({ group: 2, frame: 0 });
 
 			// The grace gives up on group 0, and the settled subscription ends the reader.
+			expect(await clock.drive(orderedHeld)).toBe("1.0");
 			expect(await clock.drive(held)).toBeUndefined();
+			expect(await ordered.readString()).toBeUndefined();
 			expect(await reader.closed).toBeNull();
 			ordered.close();
 		} finally {

js/net/src/resume.test.ts

@@ -1042,6 +1042,52 @@ test("an ordered reader takes a replaced copy's buffered group before the new co
 	b.close();
 });
 
+for (const resolution of ["arrival", "drop", "other copy", "budget"] as const) {
+	test(`an ordered resumed reader keeps an older copy's owed hole until ${resolution}`, async () => {
+		const clock = new Manual();
+		const installed = clock.install();
+		const front = new Front();
+		const a = new BroadcastProducer();
+		const b = new BroadcastProducer();
+		const ta = copy(a);
+		const tb = copy(b);
+		hooks.receiveTrack(ta.producer);
+		hooks.receiveTrack(tb.producer);
+		front.serve(a.consume());
+		const ordered = front.consumer
+			.track("seq")
+			.subscribe({ maxDelay: Milli(1000) })
+			.ordered();
+		try {
+			ta.write(4);
+			expect((await ordered.nextGroup())?.sequence).toBe(4);
+			ta.write(6);
+			tb.write(7);
+			hooks.startTrack(tb.producer, { start: 7, requested: 4 });
+			front.serve(b.consume());
+			await settle();
+			const held = ordered.nextGroup();
+			expect(await isPending(held)).toBe(true);
+			if (resolution === "arrival") ta.write(5);
+			if (resolution === "other copy") {
+				tb.write(5);
+				tb.write(6);
+			}
+			if (resolution === "drop") hooks.dropTrackGroups(ta.producer, 5, 6, new StreamError(StreamCode.Old));
+			if (resolution === "budget") await clock.advance(Nano.fromMilli(Milli(1000)));
+			const expected = resolution === "arrival" || resolution === "other copy" ? [5, 6, 7] : [6, 7];
+			expect((await held)?.sequence).toBe(expected[0]);
+			for (const sequence of expected.slice(1)) expect((await ordered.nextGroup())?.sequence).toBe(sequence);
+		} finally {
+			ordered.close();
+			front.consumer.close();
+			a.close();
+			b.close();
+			installed.close();
+		}
+	});
+}
+
 test("a late subscriber's first route is checked against properties an earlier source gave", async () => {
 	const front = new Front();
 	const a = new BroadcastProducer();

js/net/src/resume.ts

@@ -281,8 +281,8 @@ function subscribe(wire: BroadcastWire, name: string, subscription: track.Subscr
 }
 
 /** Take a copy's next group, in sequence or arrival order. */
-function take(copy: Copy, ordered: boolean): Recv {
-	return ordered ? hooks.tryNextGroup(copy.sub) : hooks.tryRecvGroup(copy.sub, false);
+function take(copy: Copy, ordered: boolean, floor?: number): Recv {
+	return ordered ? hooks.tryNextGroup(copy.sub, floor) : hooks.tryRecvGroup(copy.sub, false);
 }
 
 /** The first group past where `copy` was replaced: 0 for one replaced before any group. */
@@ -650,18 +650,21 @@ class Reader implements ResumeSource {
 		// lower group that truly arrives later.
 		const next = this.#lowest(ordered, delivered);
 		if (next) {
-			const recv = take(next, ordered);
+			const recv = take(next, ordered, this.#newest === undefined ? undefined : this.#newest + 1);
 			if (recv.kind === "group") {
 				this.#observe(next, recv.group.sequence);
 				return { kind: "group", group: this.#handOut(next, recv.group) };
 			}
+			// The lowest buffered group keeps its place while its copy waits on a hole.
+			// A newer copy must not pass it and permanently discard it from sequence order.
+			if (ordered && recv.kind === "idle") return recv;
 		}
 		// The serving copy first: it is the one that keeps delivering.
 		for (let index = this.#copies.length - 1; index >= 0; index--) {
 			const copy = this.#copies[index];
 			if (!this.#feeds(copy)) continue;
 			for (;;) {
-				const recv = take(copy, ordered);
+				const recv = take(copy, ordered, this.#newest === undefined ? undefined : this.#newest + 1);
 				if (recv.kind === "group") {
 					if (!this.#deliverable(copy, recv.group.sequence, delivered)) {
 						recv.group.close();
@@ -745,15 +748,19 @@ class Reader implements ResumeSource {
 		for (const copy of this.#copies) {
 			if (!this.#feeds(copy)) continue;
 			for (;;) {
-				const sequence = hooks.peekGroup(copy.sub, ordered);
+				const sequence = hooks.peekGroup(
+					copy.sub,
+					ordered,
+					this.#newest === undefined ? undefined : this.#newest + 1,
+				);
 				if (sequence === undefined) break;
 				if (this.#deliverable(copy, sequence, delivered)) {
 					// A tie goes to the newer copy, so the serving one hands out its first group and
 					// bounds the copies it replaced.
 					if (!lowest || sequence <= lowest.sequence) lowest = { copy, sequence };
 					break;
 				}
-				const recv = take(copy, ordered);
+				const recv = take(copy, ordered, this.#newest === undefined ? undefined : this.#newest + 1);
 				if (recv.kind !== "group") break;
 				recv.group.close();
 			}