fix(net)!: report a reader's largest within its local bounds

doc/concept/moq-lite.md

@@ -259,7 +259,10 @@ untimed.
 A relay cancels its upstream subscription once nobody subscribes, but keeps its
 copy of the track for 30 seconds after the last reader leaves, so a returning
 reader or the next fetch finds it. That copy is not live meanwhile: readers get
-nothing from it until the source answers again.
+nothing from it until the source answers again, and its latest group is unknown
+until then too, so a player starting at the latest waits for the answer instead
+of starting from what the ended feed left. A reader's latest group stays within
+the groups it limits itself to, and ends once the track does.
 
 The publisher may declare a retention window per track. Omission sets no limit;
 zero keeps only the live edge. Relays preserve that declaration, and may still

js/net/MIGRATE.md

@@ -4,6 +4,14 @@ How to update your code for each breaking change, newest first.
 
 ## Unreleased
 
+### `largest()` reports within a reader's `setGroups` bounds
+
+`Track.Subscriber.largest()` and `Track.Ordered.largest()` report the track's largest only while it is within the bounds set with `setGroups` or `withGroups`, where they used to ignore the bounds.
+A largest past the cap or below the floor reports nothing; the cache never stands in for it.
+A cap moved below the largest stays pending. Moving the bounds wakes a pending read, which reports the largest once it is within them and newer than the last report.
+Unbounded readers and `Track.Producer.largest()` are unchanged.
+To read the track's edge from a capped reader, lift the cap with `setGroups({})` first.
+
 ### Lite-07 accounts for groups with `SUBSCRIBE_DROP`
 
 Construct `SubscribeEnd(group)` without a stream count; remove reads of `.streams` and calls to `hasStreamCount()`.

js/net/src/internal.ts

@@ -304,7 +304,10 @@ export interface ResumeSource {
 	bufferedLargest(): Location | undefined;
 	/** The serving copy's largest location now, and whether it can still advance. */
 	peekLargest(): { largest?: Location; ended: boolean };
-	/** The serving copy's largest location once newer than `seen`, or undefined once the logical track ends. */
+	/**
+	 * The serving copy's largest location once within the reader's cursor and newer than `seen`,
+	 * or undefined once the logical track ends.
+	 */
 	largest(seen: Location | undefined): Promise<Location | undefined>;
 	/** An independent subscriber to the serving copy. */
 	fork(options?: Subscription): Subscriber;

js/net/src/resume.test.ts

@@ -20,6 +20,17 @@ async function settle() {
 	await new Promise((resolve) => setTimeout(resolve, 0));
 }
 
+/** Whether `promise` is still pending once queued work settles. */
+async function isPending(promise: Promise<unknown>): Promise<boolean> {
+	let settled = false;
+	void promise.then(
+		() => (settled = true),
+		() => (settled = true),
+	);
+	await settle();
+	return !settled;
+}
+
 /** A stand-in session announcing `path` under the epoch at `cost`, serving `source`. */
 function serve(origin: Producer, source: BroadcastProducer, cost: bigint) {
 	const handle = wireOf(origin).receive(path, Route.normalize({ epoch: EPOCH, cost }));
@@ -1151,6 +1162,36 @@ test("a fetch still out on the old route moves to the new one", async () => {
 	b.close();
 });
 
+// A front's reader reports the serving copy's largest only within its bounds: the cap it sets
+// and the floor its subscription names.
+test("the largest honours the reader's bounds on the serving copy", async () => {
+	const front = new Front();
+	const a = new BroadcastProducer();
+	const ta = copy(a);
+	front.serve(a.consume());
+	const capped = front.consumer
+		.track("seq")
+		.subscribe()
+		.withGroups({ end: { excluded: 3 } });
+	const floored = front.consumer.track("seq").subscribe({ groups: { start: { included: 8 } } });
+	for (let sequence = 0; sequence < 5; sequence++) ta.write(sequence);
+
+	const next = capped.largest();
+	expect(await isPending(next)).toBe(true);
+	capped.setGroups({});
+	expect(await next).toEqual({ group: 4, frame: 0 });
+
+	const floor = floored.largest();
+	expect(await isPending(floor)).toBe(true);
+	ta.write(8);
+	expect(await floor).toEqual({ group: 8, frame: 0 });
+
+	capped.close();
+	floored.close();
+	front.consumer.close();
+	a.close();
+});
+
 test("the largest follows the serving copy across a swap without moving back, and ends with the route", async () => {
 	const front = new Front();
 	const a = new BroadcastProducer();

js/net/src/resume.ts

@@ -651,9 +651,12 @@ class Reader implements ResumeSource {
 			const serving = this.#serving();
 			const view = serving && hooks.peekLargest(serving.sub);
 			const largest = view?.largest;
+			const { start, end } = this.#shell.cursor.peek();
 			// A replacement copy can start below what an earlier one reported.
 			if (
 				largest &&
+				largest.group >= start &&
+				(end === undefined || largest.group < end) &&
 				(!seen || largest.group > seen.group || (largest.group === seen.group && largest.frame > seen.frame))
 			)
 				return largest;
@@ -666,7 +669,7 @@ class Reader implements ResumeSource {
 					for (const dispose of disposes) dispose();
 					resolve();
 				};
-				disposes.push(this.changed(wake), this.#shell.closed.changed(wake));
+				disposes.push(this.changed(wake), this.#shell.closed.changed(wake), this.#shell.cursor.changed(wake));
 				if (serving) disposes.push(hooks.largestChanged(serving.sub, wake));
 			});
 		}

js/net/src/track.test.ts

@@ -2027,6 +2027,92 @@ test("a datagram raises the largest", async () => {
 	producer.close();
 });
 
+/** Write untimed group `sequence` with `frames` frames, left open. */
+function framedGroup(producer: TrackProducer, sequence: number, frames: number): GroupProducer {
+	const group = new GroupProducer(sequence);
+	producer.writeGroup(group);
+	for (let i = 0; i < frames; i++) group.writeFrame({ payload: enc.encode("x") });
+	return group;
+}
+
+/** Whether `promise` is still pending once queued work settles. */
+async function isPending(promise: Promise<unknown>): Promise<boolean> {
+	let settled = false;
+	void promise.then(
+		() => (settled = true),
+		() => (settled = true),
+	);
+	await settle();
+	return !settled;
+}
+
+// The largest reports only while it is within the bounds: a cap below it or a floor above it
+// reports nothing, rather than a cached group standing in for it.
+test("largest reports only within the bounds", async () => {
+	const producer = new TrackProducer("test").accept({});
+	for (let sequence = 0; sequence < 6; sequence++) framedGroup(producer, sequence, 2);
+
+	const capped = producer.subscribe().withGroups({ end: { excluded: 3 } });
+	expect(await isPending(capped.largest())).toBe(true);
+
+	// An inclusive cap reports the group it names.
+	const ordered = producer
+		.subscribe()
+		.ordered()
+		.withGroups({ end: { included: 5 } });
+	expect(await ordered.largest()).toEqual({ group: 5, frame: 1 });
+
+	const floored = producer.subscribe().withGroups({ start: { included: 8 } });
+	const next = floored.largest();
+	expect(await isPending(next)).toBe(true);
+	framedGroup(producer, 8, 0);
+	expect(await next).toEqual({ group: 8, frame: 0 });
+	producer.close();
+});
+
+// Moving the bounds wakes a pending read, which reports the largest once it is within them;
+// bounds that leave it where it was report nothing new.
+test("largest follows the bounds as they move", async () => {
+	const producer = new TrackProducer("test").accept({});
+	for (let sequence = 0; sequence < 6; sequence++) framedGroup(producer, sequence, 1);
+
+	const reader = producer
+		.subscribe()
+		.ordered()
+		.withGroups({ end: { excluded: 3 } });
+	let next = reader.largest();
+	expect(await isPending(next)).toBe(true);
+	reader.setGroups({});
+	expect(await next).toEqual({ group: 5, frame: 0 });
+
+	next = reader.largest();
+	expect(await isPending(next)).toBe(true);
+	reader.setGroups({ end: { excluded: 2 } });
+	expect(await isPending(next)).toBe(true);
+
+	// Lifting it again leaves the last report standing.
+	reader.setGroups({});
+	expect(await isPending(next)).toBe(true);
+	framedGroup(producer, 6, 0);
+	expect(await next).toEqual({ group: 6, frame: 0 });
+	reader.setGroups({ start: { included: 3 } });
+	expect(await isPending(reader.largest())).toBe(true);
+	producer.close();
+});
+
+// Once the stream ended, widening the bounds over a largest it never reported does not bring
+// it back.
+test("largest stays ended when the bounds widen", async () => {
+	const producer = new TrackProducer("test").accept({});
+	framedGroup(producer, 5, 1).close();
+	producer.close();
+
+	const reader = producer.subscribe().withGroups({ end: { excluded: 3 } });
+	expect(await reader.largest()).toBeUndefined();
+	reader.setGroups({});
+	expect(await reader.largest()).toBeUndefined();
+});
+
 // A reader that starts at the largest group while it is still empty is at the live edge: it
 // gets that group, which ends with nothing, then the next one.
 test("a live-edge reader starts on an empty largest group", async () => {
@@ -2682,8 +2768,10 @@ test("a mid-group end survives eviction and a filtered late reader", async () =>
 		group.close();
 		producer.close();
 		await clock.advance(10_000 + CACHE_WINDOW_MS + 1);
+		expect(await producer.subscribe().largest()).toEqual({ group: 5, frame: 4 });
+		// The cap excludes the only group, so nothing is within it.
 		const reader = producer.subscribe().withGroups({ end: { excluded: 5 } });
-		expect(await reader.largest()).toEqual({ group: 5, frame: 4 });
+		expect(await reader.largest()).toBeUndefined();
 		expect(reader.end()).toEqual({ group: 5, frame: 5 });
 		reader.close();
 	} finally {

js/net/src/track.ts

@@ -494,17 +494,19 @@ async function resolveInfo(state: TrackState): Promise<Info> {
 	}
 }
 
+// Whether `a` is past `b`.
+function after(a: Location, b: Location): boolean {
+	return a.group > b.group || (a.group === b.group && a.frame > b.frame);
+}
+
 async function nextLargest(
 	state: TrackState,
 	seen: Location | undefined,
 	commit: (largest: Location) => void,
 ): Promise<Location | undefined> {
 	for (;;) {
 		const largest = state.largest.peek();
-		if (
-			largest &&
-			(!seen || largest.group > seen.group || (largest.group === seen.group && largest.frame > seen.frame))
-		) {
+		if (largest && (!seen || after(largest, seen))) {
 			commit(largest);
 			return largest;
 		}
@@ -1100,12 +1102,7 @@ export class Producer {
 	#advanceLargest(largest: Location | undefined): void {
 		if (!largest) return;
 		const current = this.#state.largest.peek();
-		if (
-			current &&
-			(current.group > largest.group || (current.group === largest.group && current.frame >= largest.frame))
-		) {
-			return;
-		}
+		if (current && !after(largest, current)) return;
 		this.#largestOpen = true;
 		this.#state.largest.set(largest);
 		for (const sink of this.#sinks) sink.largest.set(largest);
@@ -1594,7 +1591,10 @@ export class Subscriber {
 	#delivered = new Delivered();
 	// Whether a resumed reader has read groups, so a datagram read keeps copies they still need.
 	#groupsRead = false;
+	// What {@link largest} last reported, and whether it ended: moving the bounds afterwards
+	// does not bring it back.
 	#largestSeen?: Location;
+	#largestDone = false;
 	#nextSequence = 0;
 	#cursor = new Signal<{ start: number; end?: number }>({ start: 0 });
 	// A FETCH's own budget in place of the subscription's; unlike SUBSCRIBE's zero
@@ -1951,18 +1951,39 @@ export class Subscriber {
 	}
 
 	/**
-	 * The track's largest location, its newest group or datagram at the newest frame (0 while
-	 * empty): the current one at first, then each newer one (skipping any in between), and
-	 * `undefined` once the track ends. Seeded by the publisher's report, so it can name a frame
-	 * before its group arrives.
+	 * The largest location within {@link setGroups}: the current one at first, then each newer
+	 * one (skipping any in between), and `undefined` once the track ends.
+	 *
+	 * The largest is the track's newest group or datagram at the newest frame (0 while empty), as
+	 * the publisher reported it and its feed delivered since, so it can name a frame before its
+	 * group arrives. A largest past the cap or below the floor reports nothing: the cache does not
+	 * stand in for it, so a cap moved below the largest stays pending. Moving the bounds wakes a
+	 * pending read, which reports the largest once it is within them and newer than the last report.
 	 */
 	async largest(): Promise<Location | undefined> {
+		if (this.#largestDone) return undefined;
 		if (this.#resume) {
 			const largest = await this.#resume.largest(this.#largestSeen);
 			if (largest) this.#largestSeen = largest;
+			else this.#largestDone = true;
 			return largest;
 		}
-		return nextLargest(this.#state, this.#largestSeen, (largest) => (this.#largestSeen = largest));
+		for (;;) {
+			const { start, end } = this.#cursor.peek();
+			const largest = this.#state.largest.peek();
+			const within =
+				largest && largest.group >= start && (end === undefined || largest.group < end) ? largest : undefined;
+			const seen = this.#largestSeen;
+			if (within && (!seen || after(within, seen))) {
+				this.#largestSeen = within;
+				return within;
+			}
+			if (this.#state.closed.peek() !== undefined || this.#state.largestEnded.peek()) {
+				this.#largestDone = true;
+				return undefined;
+			}
+			await Signal.race(this.#state.largest, this.#state.closed, this.#state.largestEnded, this.#cursor);
+		}
 	}
 
 	// The newest frame buffered for this subscriber, which a wire answer names as its edge:
@@ -2014,6 +2035,7 @@ export class Subscriber {
 	 * Limit subsequent reads without rewinding read progress or changing the wire request.
 	 * An omitted start preserves the current floor; an omitted end removes the cap.
 	 * Raising the end makes unread buffered groups available again.
+	 * {@link largest} reports within these bounds.
 	 */
 	setGroups(groups: Groups): void {
 		const { start, end } = groupBounds(groups);
@@ -2526,7 +2548,7 @@ export class Ordered {
 		return this.#subscriber.subscription;
 	}
 
-	/** The track's largest location; see {@link Subscriber.largest}. */
+	/** The largest location within {@link setGroups}; see {@link Subscriber.largest}. */
 	largest(): Promise<Location | undefined> {
 		return this.#subscriber.largest();
 	}

quest/m1/README.md

@@ -122,6 +122,7 @@ lands before the cut.
 - [Gateway linger](/quest/m1/linger-gateways.md) - HLS, RTMP, SRT and RTC imports linger at a clean end, cut short by a newer epoch at the same path
 - [Exporters follow a rendition to a new track](/quest/m1/export-rendition-rebind.md) - the fMP4, TS, MKV and FLV exporters rebuild a rendition whose resolved `track` or `broadcast` changes under the same ID
 - [lite-06 max sequence](/quest/m1/lite06-max-sequence.md) - group sequence 2^62-1 reaches a lite-06 subscriber or is refused loudly, instead of vanishing
+- [Track live signal](/quest/m1/track-live.md) - a reader can wait for its track's feed to be live, so a player skips old cached groups while a fresh upstream subscription starts
 
 - [C++ and OBS package proof](/quest/m1/cpp-package-check.md) - the installed artifacts pass before their first release