feat(net)!: support JavaScript subscription ranges

doc/concept/moq-lite.md

@@ -298,7 +298,10 @@ Ranges are both a request and a filter. A subscriber reads only the groups its
 own ranges cover, resolved again whenever it replaces them, even when another
 subscriber of the same track asked for more. Widening them, including lowering
 the start, also serves the cached groups they newly cover, never one already
-served. A group they cover below the live feed that the cache lacks (evicted
+served. A reader that has read every group its ranges cover waits rather than
+ends, since the track goes on: a moq-lite-07 publisher ends a bounded
+subscription once it covered it, and the copy stays cached for the same ranges
+until an update asks for groups it did not cover, wider or shifted. A group they cover below the live feed that the cache lacks (evicted
 before the subscriber first looked for it, skipped, or below where the feed
 starts) is asked of the track's dynamic handler
 as a fill, which a relay serves by fetching upstream one group at a time, skipping
@@ -312,7 +315,8 @@ track, and over moq-lite-07, which carries a relative or time start upstream and
 reports the live edge and media time, so a copy resolves the ends itself. A Rust
 moq-transport copy resolves a relative anchor from the Largest Object its live join
 reports, accepts time anchors its cache can already resolve, and sends group bounds
-upstream; it refuses any other time anchor. Older
+upstream; it refuses any other time anchor. A JavaScript one refuses every time
+anchor. Older
 moq-lite versions accept absolute anchors but refuse time anchors and relative anchors other
 than the live and default starts. An untimed track refuses a time anchor. At the live
 edge, a time end

doc/lib/js/net.md

@@ -52,6 +52,7 @@ publication.announce();
 - **Timing is per track, with no default.** A track that declares a `timescale` carries a `timestamp` on every frame; one without is [untimed](/concept/moq-lite#subscriptions) and its frames carry none. Omitting it means untimed, as in Rust. Nothing stamps a frame for you: the publisher passes the timestamp, and the `Timed` type (`{ value, at }`) carries one into the `@moq/json` producers.
 - **A subscription's `delay` is a `Track.Delay { min, max }`, as in Rust.** Staleness is `delay.max`: it counts from a group's successor, by media time and by wall clock, whichever reaches it first. `delay.min` is the target latency, carried on moq-lite 07; a peer that cannot carry it leaves it `null` (not stated). Passing the old `maxAge` or `maxDelay` key throws a `TypeError` naming `delay.max`. Publisher retention is the separate `Track.Info.maxAge`.
 - **Subscription defaults.** `Track.Subscription.default()` returns the default subscription, whose zero `delay.max` skips a superseded group at once; calling `subscribe()` with no options means the same. An omitted `delay` field means the same zero; `max: null` sets no subscriber limit.
+- **A subscription asks for `ranges`.** Each [range](/concept/moq-lite#subscriptions) starts and ends at an `absolute` group, one `relative` to the latest, or the group containing a `time` (a `Timestamp`), and a reader reads only its own ranges. A read whose ranges all end finishes once every group in them is handed out, dropped, or skipped, as in Rust. A time anchor needs a timed track and `moq-lite-07-wip` over a session. Passing the old `groups` key throws a `TypeError` naming `ranges`; `withRange` only narrows reads locally.
 - **Hidden paths** stay out of discovery unless the announce request opts in. See [hidden broadcasts](/concept/moq-lite#hidden-broadcasts).
 - **Authorization is in band.** On moq-lite 07 (opt in with `versions: ["moq-lite-07-wip", "moq-lite-06"]`, listing every fallback you still want) and on moq-transport draft-18+ with the [MoQ Auth extension](/draft/moq-auth), `auth.grant` watches what the relay lets this side publish and subscribe to, and `auth.add(token)` presents another token without reconnecting. Publishing outside the grant closes the connection with `SessionCode.Unauthorized`, naming the path. On moq-lite, a subscription or track listing the grant stops covering resets with `StreamCode.Unauthorized` and the session stays up. On moq-transport, `auth.add(token)` keeps the token even without the extension, and a broadcast announced with the `Auth.Credential` it returns (`publication.announce({ auth })`) carries it on that announcement's `PUBLISH_NAMESPACE`; announcing again with another renews the request. A `Connection`'s `auth` handle (a getter on a pooled connection, following its URL) outlives its sessions: a token added there is presented on every reconnect and keeps riding the announcements that name it, with no re-add or re-announce. A one-shot session's `auth` ends with it, unless `Connection.connect` was given a `new Auth.Handle()` as `auth`.
 - **A graceful close waits.** `await connection.close()` withdraws announcements and drains for up to one second. `moq-lite-05` and newer wait for the subscriber to receive finished tails; older Lite versions wait only for transport ACKs. IETF media streams are not drained. `abort()` ends immediately.

js/net/MIGRATE.md

@@ -11,6 +11,17 @@ Replace `maxDelay: d` with `delay: { max: d }`. `max` keeps `maxDelay`'s meaning
 `min` is the subscriber's target latency; `null` means not stated, which only a peer on moq-lite before 07 or moq-transport sends. moq-lite-07 SUBSCRIBE and SUBSCRIBE\_UPDATE carry it, and a track's aggregate takes the smallest stated one.
 `FetchGroupOptions.maxDelay` is unchanged.
 
+### `Subscription.groups` is `ranges` ([#6949](https://github.com/moq-dev/moq/pull/6949))
+
+`Track.Subscription` asks for `ranges: Track.Range[]` in place of `groups`, which is now refused at runtime. Each `Track.Range` starts and ends at a `Track.Anchor`: an `absolute` group, one `relative` to the latest, or the group containing a `time`.
+Replace `groups: { start: { included: a }, end: { excluded: b } }` with `ranges: [{ start: { absolute: a }, end: { absolute: b } }]`. An inclusive end `b` becomes `absolute: b + 1`, and an excluded start `a` becomes `absolute: a + 1`.
+A reader reads only its own ranges, whatever another subscriber widens the request to. `withRange` only narrows them; widen with `update({ ranges })`.
+
+### Bounded reads end ([#6949](https://github.com/moq-dev/moq/pull/6949))
+
+A `recvGroup` or `nextGroup` whose ranges all end now returns `undefined` once every group in them was handed out, dropped, skipped as stale, or will never arrive, as Rust's reads do since [#7545](https://github.com/moq-dev/moq/pull/7545). The track goes on: replacing the ranges with `update` reads on from the same subscriber. A `withRange` cap holding a group back keeps the read open, since the cap may rise.
+A loop that read a bounded range until a timeout now ends on its own; one that waited for more groups past the end of its ranges must widen them first.
+
 ### 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 `delay.max` by media time or wall clock. Local tracks still skip gaps, and the zero default waits for nothing.

js/net/api/index.api.md

@@ -575,6 +575,19 @@ export interface Timed<T> {
 }
 
 export namespace Track {
+	export type Anchor = {
+		absolute: number;
+		relative?: never;
+		time?: never;
+	} | {
+		absolute?: never;
+		relative: number;
+		time?: never;
+	} | {
+		absolute?: never;
+		relative?: never;
+		time: Time.Timestamp;
+	};
 	export type Bound = {
 		excluded?: never;
 		included: number;
@@ -665,6 +678,10 @@ export namespace Track {
 		get subscription(): Signals.Getter<Track.Subscription | undefined>;
 		writeFrame(frame: Group.Frame): void;
 	}
+	export interface Range {
+		end?: Track.Anchor;
+		start?: Track.Anchor;
+	}
 	export class Request {
 		accept(info?: Partial<Track.Info>): Track.Producer;
 		demand(): Track.Demand;
@@ -693,9 +710,9 @@ export namespace Track {
 	};
 	export interface Subscription {
 		delay?: Track.Delay;
-		groups?: Track.Groups;
 		order?: Track.Order;
 		priority?: number;
+		ranges?: Track.Range[];
 	}
 	export function infoDefaults(info?: Partial<Track.Info>): Track.Info;
 }

js/net/bench/replay.ts

@@ -10,7 +10,7 @@ const PAYLOAD = 8;
 const APPENDS = 64;
 const REPS = 5;
 // Every retained group, however old: replay is the work being measured.
-const SUBSCRIPTION = { groups: { start: { included: 0 } }, delay: { max: null } };
+const SUBSCRIPTION = { ranges: [{ start: { absolute: 0 } }], delay: { max: null } };
 
 // The arrival order of sequences 0..count-1: ascending, descending, or a seeded shuffle.
 function arrivals(count: number, order: Order): number[] {

js/net/src/broadcast.test.ts

@@ -319,20 +319,20 @@ test("a request exposes the aggregate subscription options", async () => {
 	consumer.track("video").subscribe({
 		priority: 3,
 		delay: { max: Milli(100) },
-		groups: { start: { included: 10 }, end: { excluded: 20 } },
+		ranges: [{ start: { absolute: 10 }, end: { absolute: 20 } }],
 	});
 	consumer.track("video").subscribe({
 		priority: 7,
 		delay: { max: Milli(250) },
-		groups: { start: { included: 0 }, end: { excluded: 30 } },
+		ranges: [{ start: { absolute: 0 }, end: { absolute: 30 } }],
 	});
 
 	const request = await pendingRequest(consumer);
 	expect(request?.subscription).toEqual({
 		priority: 7,
 		delay: { min: Milli.zero, max: Milli(250) },
 		order: "descending",
-		groups: { start: { included: 0 }, end: { excluded: 30 } },
+		ranges: [{ start: { absolute: 0 }, end: { absolute: 30 } }],
 	});
 	expect(request?.priority).toBe(7);
 });

js/net/src/delivered.ts

@@ -66,6 +66,14 @@ export class Delivered {
 		this.#runs.splice(first, last - first + 1, merged);
 	}
 
+	/** A record of what either this or `other` handed out. */
+	union(other: Delivered): Delivered {
+		const union = new Delivered();
+		union.#runs = this.#runs.map(([first, last]) => [first, last]);
+		for (const [first, last] of other.#runs) union.addRun(first, last + 1);
+		return union;
+	}
+
 	/** The parts of `[start, end)` not handed out, in order. */
 	gaps(start: number, end: number): { start: number; end: number }[] {
 		const gaps: { start: number; end: number }[] = [];

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

@@ -2387,7 +2387,7 @@ test("draft-22: an absolute filter trims the range it serves", async () => {
 		// The request forwarded upstream carries the model's exclusive end, one past the
 		// filter's inclusive last group.
 		expect(track.subscription.peek()).toMatchObject({
-			groups: { start: { included: 1 }, end: { excluded: 3 } },
+			ranges: [{ start: { absolute: 1 }, end: { absolute: 3 } }],
 		});
 
 		const first = await nextUni(fx.uni);
@@ -3441,7 +3441,7 @@ test("REQUEST_UPDATE applies priority and preserves it when omitted", async () =
 		expect(track.subscription.peek()?.priority).toBe(117);
 		// Only the priority changes; retention and the group range survive.
 		expect(track.subscription.peek()?.delay?.max).toBe(before?.delay?.max);
-		expect(track.subscription.peek()?.groups).toEqual(before?.groups);
+		expect(track.subscription.peek()?.ranges).toEqual(before?.ranges);
 		await client.writer.write(new Uint8Array([0x02, 0, 2, 4, 0]));
 		expect(await client.reader.u53()).toBe(RequestOk.id);
 		await RequestOk.decode(client.reader, fx.version);
@@ -3945,7 +3945,7 @@ test("IETF draft-18: range updates preserve the shared subscription", async () =
 			await new SubscribeUpdate({ requestId: id, parameters }).encode(client.writer, fx.version);
 			expect(await client.reader.u53()).toBe(RequestOk.id);
 			await RequestOk.decode(client.reader, fx.version);
-			expect(track.subscription.peek()?.groups).toEqual({ start: { included: 6 }, end: { excluded: 8 } });
+			expect(track.subscription.peek()?.ranges).toEqual([{ start: { absolute: 6 }, end: { absolute: 8 } }]);
 		}
 	} finally {
 		client.close();

js/net/src/ietf/publisher.ts

@@ -29,7 +29,7 @@ import { sendPayload } from "../payload.ts";
 import { Priority, sendOrder } from "../priority.ts";
 import { type Stream, Writer } from "../stream.ts";
 import { Timescale } from "../time.ts";
-import type { Groups, Info, Position as TrackPosition, Subscriber as TrackSubscriber } from "../track.ts";
+import type { Groups, Info, Range, Position as TrackPosition, Subscriber as TrackSubscriber } from "../track.ts";
 import { untilAborted } from "../util/abort.ts";
 import { nextChange } from "../util/change.ts";
 import { Deadline, REQUEST_TIMEOUT_MS } from "../util/deadline.ts";
@@ -497,14 +497,17 @@ export class Publisher {
 			// A range that resolves without the live edge has a fixed start, which is no join:
 			// bound the subscription before subscribing so the refresh ratio never judges it.
 			const fixed = subscribeRange(msg, {}, version);
-			return source.track(msg.trackName).subscribe({
-				priority: fromWire(msg.subscriberPriority),
-				order: msg.groupOrder,
-				// An absent budget preserves everything retained by the producer.
-				// moq-transport has no field for a minimum.
-				delay: { max: Milli(msg.maxAge ?? Varint.MAX_U53), min: null },
-				groups: fixed.start ? rangeGroups(fixed) : undefined,
-			});
+			// Parked past a bounded window's end: the peer reads this stream's end as the track's.
+			return hooks.parked(
+				source.track(msg.trackName).subscribe({
+					priority: fromWire(msg.subscriberPriority),
+					order: msg.groupOrder,
+					// An absent budget preserves everything retained by the producer.
+					// moq-transport has no field for a minimum.
+					delay: { max: Milli(msg.maxAge ?? Varint.MAX_U53), min: null },
+					ranges: fixed.start ? serveRanges(fixed) : undefined,
+				}),
+			);
 		};
 		let track = subscribe(broadcast);
 
@@ -710,7 +713,7 @@ export class Publisher {
 				order,
 				// moq-transport has no field for a minimum.
 				delay: { max: Milli(msg.maxAge ?? Varint.MAX_U53), min: null },
-				groups: rangeGroups(range),
+				ranges: serveRanges(range),
 			});
 			const startGroup = range.start ? Number(range.start.group) : hooks.latestSequence(track);
 			if (startGroup !== undefined) track.withRange({ ...rangeGroups(range), start: { included: startGroup } });
@@ -864,7 +867,7 @@ export class Publisher {
 						range.start = tail.start;
 						range.end = tail.end;
 						bounds.set(requested, true);
-						track.update({ ...track.subscription.peek(), groups: rangeGroups(range) });
+						track.update({ ...track.subscription.peek(), ranges: serveRanges(range) });
 						// Below its floor the live cursor already passed every group or never owned it,
 						// so lowering it for a fill-only update would hand those groups out a second time.
 						let start = range.start ? Number(range.start.group) : 0;
@@ -2849,6 +2852,14 @@ async function servingGroup(track: TrackSubscriber): Promise<group.Consumer | un
 }
 
 /** The model's group bounds for a {@link ServeRange}: an absent start joins at the latest group. */
+/** The model ranges a {@link ServeRange} asks for, whole groups through its last one. */
+function serveRanges(range: ServeRange): Range[] {
+	const out: Range = {};
+	if (range.start) out.start = { absolute: Number(range.start.group) };
+	if (range.end) out.end = { absolute: Number(range.end.group) + 1 };
+	return [out];
+}
+
 function rangeGroups(range: ServeRange): Groups {
 	return {
 		start: range.start ? { included: Number(range.start.group) } : undefined,

js/net/src/ietf/subscriber.ts

@@ -21,16 +21,7 @@ import {
 import type * as netGroup from "../group.ts";
 import { Cost, type Route, routesEqual, UNKNOWN_HOP } from "../hop.ts";
 import * as Instance from "../instance.ts";
-import {
-	finalSequence,
-	groupBounds,
-	hiddenBelow,
-	hooks,
-	maxDelayOf,
-	scopeCaptures,
-	scopeHead,
-	scopeOverlaps,
-} from "../internal.ts";
+import { finalSequence, hiddenBelow, hooks, maxDelayOf, scopeCaptures, scopeHead, scopeOverlaps } from "../internal.ts";
 import type { Lister } from "../listing.ts";
 import * as Path from "../path.ts";
 import { receiveFrame, skipPayload } from "../payload.ts";
@@ -43,6 +34,7 @@ import { closeUnwanted } from "../util/announce.ts";
 import { Deadline, requestOptions } from "../util/deadline.ts";
 import { abandoned as demandAbandoned, isAbandoned, unwanted } from "../util/demand.ts";
 import { LIMITS, type Limits, SlotMap, Slots } from "../util/limits.ts";
+import { wireSpan } from "../window.ts";
 import { overrideBroadcastWire, wireOf } from "../wire.ts";
 import { DuplicateTrackAlias, RetiredTrackAlias, TrackAliases } from "./aliases.ts";
 import * as Cluster from "./cluster.ts";
@@ -1288,10 +1280,9 @@ export class Subscriber {
 		// The live join starts at the publisher's edge, whatever floor the demand asked for.
 		// With no Largest it skips nothing, so every group is owed: a lower one's stream may
 		// still arrive after a higher one's.
-		const groups = request.subscription.groups;
 		hooks.startTrack(producer, {
 			start: ok.largest ? Number(ok.largest.groupId) : 0,
-			requested: groups?.start === undefined ? undefined : groupBounds(groups).start,
+			requested: wireSpan(request.subscription.ranges ?? [{}]).start,
 		});
 		staged.commit();
 		state.fill?.(subscription);