feat(net): a timed track refuses a group that rewinds

drafts/draft-lcurley-moq-lite.md

@@ -174,7 +174,9 @@ A Group Stream normally starts at frame 0, but MAY start later when the publishe
 
 Each frame of a timed Track carries a presentation timestamp expressed in the Track's `Timescale` (see [TRACK_INFO](#track-info)), used by the moq-lite layer for [expiration](#expiration) decisions.
 A Track without a `Timescale` is untimed: its frames and datagrams carry no timestamp.
+A publisher MUST NOT start a timed Group's frame 0 earlier than that of any lower-sequence Group it publishes.
 lite-05 and lite-06 cannot encode an untimed Track, so a publisher on those versions declares a `Timescale` of 1000 for it and stamps every frame and datagram with its send time in milliseconds.
+A receiver cannot tell such send times from presentation timestamps, and they need not rise with sequence, so it does not check received Groups against this rule.
 
 ## Positions {#positions}
 A Position is a (Group Sequence, Frame Index) pair identifying one frame within a Track.
@@ -1582,6 +1584,7 @@ The `Message Length` describes the payload size on the wire.
 
 ## moq-lite-07
 
+- A publisher must not start a timed Group's frame 0 earlier than a lower-sequence Group's. Receivers do not check it, since a received timestamp may be a send time.
 - Added the reliable ordered RTCDataChannel binding, with one Qmux Record per binary message and no datagrams.
 
 - Added `Group Order` to SUBSCRIBE and SUBSCRIBE_UPDATE: descending (0x0, the previous fixed order) or ascending (0x1). A relay requests descending upstream unless every subscription it serves is ascending.

js/hang/src/container/consumer.test.ts

@@ -526,10 +526,16 @@ function encodeLegacy(timestamp: Micro): Uint8Array {
 	return data;
 }
 
-function writeGroupWithLegacyFrames(track: Track.Producer, sequence: number, timestamps: Micro[]) {
+// `stamp` is the transport timestamp, the container's own by default.
+function writeGroupWithLegacyFrames(
+	track: Track.Producer,
+	sequence: number,
+	timestamps: Micro[],
+	stamp: (ts: Micro) => Time.Timestamp = (ts) => Time.Timestamp.fromMicros(ts),
+) {
 	const group = new Group.Producer(sequence);
 	for (const ts of timestamps) {
-		group.writeFrame({ payload: encodeLegacy(ts), timestamp: Time.Timestamp.now() });
+		group.writeFrame({ payload: encodeLegacy(ts), timestamp: stamp(ts) });
 	}
 	group.close();
 	track.writeGroup(group);
@@ -953,12 +959,16 @@ async function nextFrame(consumer: Consumer) {
 	}
 }
 
+// The track refuses a transport timestamp that rewinds, so these rewind only the container's
+// own timestamps, under wall-clock transport stamps that rise in write order.
+const wallClock = () => Time.Timestamp.now();
+
 test("Consumer aborts a group below the previous group start", async () => {
 	const track = new Track.Producer("test");
 	const consumer = new Consumer(replay(track), { format: new LegacyFormat("data"), maxDelay: 30_000 as Milli });
 
-	writeGroupWithLegacyFrames(track, 0, [10_000_000 as Micro]);
-	writeGroupWithLegacyFrames(track, 1, [0 as Micro, 100_000 as Micro]);
+	writeGroupWithLegacyFrames(track, 0, [10_000_000 as Micro], wallClock);
+	writeGroupWithLegacyFrames(track, 1, [0 as Micro, 100_000 as Micro], wallClock);
 	await settle();
 
 	expect((await nextFrame(consumer))?.frame?.timestamp).toBe(10_000_000 as Micro);
@@ -971,13 +981,13 @@ test("Consumer aborts a rewind carried by a later arrival", async () => {
 	const track = new Track.Producer("test");
 	const consumer = new Consumer(replay(track), { format: new LegacyFormat("data"), maxDelay: 30_000 as Milli });
 
-	writeGroupWithLegacyFrames(track, 0, [10_000_000 as Micro]);
+	writeGroupWithLegacyFrames(track, 0, [10_000_000 as Micro], wallClock);
 	await settle();
 	expect((await nextFrame(consumer))?.frame?.timestamp).toBe(10_000_000 as Micro);
 
 	const pending = nextFrame(consumer);
 	await settle();
-	writeGroupWithLegacyFrames(track, 1, [0 as Micro, 100_000 as Micro]);
+	writeGroupWithLegacyFrames(track, 1, [0 as Micro, 100_000 as Micro], wallClock);
 
 	await expect(pending).rejects.toThrow("below the previous group start");
 
@@ -1047,10 +1057,10 @@ test("Consumer reads no gap when retiring a group holding only its endpoint", as
 	const consumer = new Consumer(replay(track), { format: new LegacyFormat("video"), maxDelay: 150 as Milli });
 
 	const group = new Group.Producer(0);
-	group.writeFrame({ payload: encodeLegacy(0 as Micro), timestamp: Time.Timestamp.now() });
+	group.writeFrame({ payload: encodeLegacy(0 as Micro), timestamp: Time.Timestamp.fromMicros(0) });
 	group.writeFrame({
 		payload: encodeLegacyFrame(33_000 as Micro, new Uint8Array()),
-		timestamp: Time.Timestamp.now(),
+		timestamp: Time.Timestamp.fromMicros(33_000),
 	});
 	group.close();
 	track.writeGroup(group);
@@ -1073,7 +1083,7 @@ test("Consumer reads a gap when retiring a truncated group", async () => {
 	const consumer = new Consumer(replay(track), { format: new LegacyFormat("audio"), maxDelay: 150 as Milli });
 
 	const group = new Group.Producer(0);
-	group.writeFrame({ payload: encodeLegacy(0 as Micro), timestamp: Time.Timestamp.now() });
+	group.writeFrame({ payload: encodeLegacy(0 as Micro), timestamp: Time.Timestamp.fromMicros(0) });
 	track.writeGroup(group);
 	await settle();
 	expect(frameOf(await consumer.next()).timestamp).toBe(0 as Micro);
@@ -1096,7 +1106,7 @@ test("Consumer reads a gap when retiring an open group", async () => {
 	const consumer = new Consumer(replay(track), { format: new LegacyFormat("audio"), maxDelay: 150 as Milli });
 
 	const group = new Group.Producer(0);
-	group.writeFrame({ payload: encodeLegacy(0 as Micro), timestamp: Time.Timestamp.now() });
+	group.writeFrame({ payload: encodeLegacy(0 as Micro), timestamp: Time.Timestamp.fromMicros(0) });
 	track.writeGroup(group);
 	await settle();
 	expect(frameOf(await consumer.next()).timestamp).toBe(0 as Micro);
@@ -1723,7 +1733,7 @@ test("Consumer reads a shed marker group as a gap", async () => {
 	track.writeGroup(group0);
 	group0.writeFrame({
 		payload: encodeLegacy(0 as Micro),
-		timestamp: Time.Timestamp.now(),
+		timestamp: Time.Timestamp.fromMicros(0),
 	});
 	await settle();
 	expect((await nextFrame(consumer))?.frame?.timestamp).toBe(0 as Micro);

js/net/src/broadcast.test.ts

@@ -1,7 +1,7 @@
 import { expect, setSystemTime, test } from "bun:test";
 import { Milli } from "@moq/clock";
 import { Consumer as BroadcastConsumer, Producer as BroadcastProducer } from "./broadcast.ts";
-import { GroupTooLarge, NotFound } from "./error.ts";
+import { GroupTooLarge, NotFound, TimestampMismatch } from "./error.ts";
 import { Producer as GroupProducer, MAX_GROUP_FRAMES } from "./group.ts";
 import { TrackSequences } from "./internal.ts";
 import { textFrame } from "./mock.ts";
@@ -55,6 +55,29 @@ test("consumer dedupes repeat subscriptions onto one upstream request", async ()
 	expect((await pendingRequest(consumer))?.name).toBe("video");
 });
 
+// A replacement continues the name's timeline, so its groups can't start before the highest one an
+// earlier producer checked, while a new broadcast starts over.
+test("a replacement producer refuses a rewind past its predecessor", () => {
+	const broadcast = new BroadcastProducer();
+	const first = broadcast.createTrack("media", { timescale: Timescale.MILLI });
+	first.appendGroup().writeFrame(textFrame("a", Timestamp.fromMillis(1000)));
+	first.close();
+
+	const second = broadcast.createTrack("media", { timescale: Timescale.MILLI });
+	const group = new GroupProducer(1);
+	second.writeGroup(group);
+	expect(() => group.writeFrame(textFrame("b", Timestamp.fromMillis(500)))).toThrow(TimestampMismatch);
+	group.writeFrame(textFrame("b", Timestamp.fromMillis(1000)));
+
+	const next = new BroadcastProducer();
+	const track = next.createTrack("media", { timescale: Timescale.MILLI });
+	track.appendGroup().writeFrame(textFrame("c", Timestamp.fromMillis(500)));
+
+	second.close();
+	broadcast.close();
+	next.close();
+});
+
 // A name keeps one sequence namespace across producer replacements. The application owns where a
 // replacement starts: an append that would reuse an earlier producer's sequence throws.
 test("a replacement producer refuses to append a sequence the name already wrote", async () => {

js/net/src/error.ts

@@ -257,8 +257,8 @@ export class GroupTooLarge extends Stream {
 }
 
 /**
- * A frame's timedness doesn't match its track: a timestamp on a track with no timescale, or
- * none on a track that has one.
+ * A frame's timestamp doesn't fit its track: a timestamp on a track with no timescale, none
+ * on a track that has one, or a group's first frame starting before the group below it.
  *
  * Raised locally by a frame or datagram write, and decoded from a moq-lite peer's
  * `TIMESTAMP_MISMATCH` reset. Mirrors the Rust `Error::TimestampMismatch`.
@@ -269,7 +269,7 @@ export class TimestampMismatch extends Stream {
 	constructor(options?: { cause?: unknown }) {
 		super(StreamCode.TimestampMismatch, {
 			...options,
-			message: "frame timestamp doesn't match track timescale",
+			message: "frame timestamp doesn't fit its track",
 		});
 		this.name = "TimestampMismatch";
 	}

js/net/src/group.ts

@@ -69,6 +69,9 @@ class GroupState {
 	timestamp?: Timestamp;
 	// The newest frame's timestamp: where a reader that has taken every frame sits.
 	latest?: Timestamp;
+	// Accepted below the track's eviction watermark, where nothing proves where it starts,
+	// so this mirror drives no media-time expiry.
+	unverified = false;
 	// When the group joined the track this mirror belongs to, at this hop: the wall clock a
 	// superseded predecessor ages from. Per track, since one group can join several.
 	arrived = Instant.zero;
@@ -159,6 +162,10 @@ export class Producer {
 	// not from its creation, so a group still being filled is never reclaimed.
 	#activity = now();
 
+	// The tracks this group joined before its first frame, each checking that frame's
+	// timestamp against its neighbors' starts.
+	#admits?: ((start: Timestamp) => () => void)[];
+
 	constructor(sequence: number) {
 		this.#state = new GroupState(sequence);
 		this.sequence = sequence;
@@ -174,6 +181,11 @@ export class Producer {
 			for (const frame of state.frames.peek()) checkTimed(timed, frame);
 			state.timed = timed;
 		};
+		hooks.admitGroupStart = (group, admit) => {
+			group.#admits ??= [];
+			group.#admits.push(admit);
+		};
+		hooks.groupStart = (group) => (group.#state.total.peek() > 0 ? group.#state.timestamp : undefined);
 		hooks.limitGroup = (group, frames) => {
 			// A closed group takes no more frames, but one already past the cap still
 			// contradicts the end.
@@ -266,6 +278,13 @@ export class Producer {
 		if (this.#limit !== undefined && this.#state.total.peek() >= this.#limit) {
 			throw new Error(`frame ${this.#state.total.peek()} is past the track's end`);
 		}
+		// Frame 0 must fit between its neighbors in every track it joined, checked before
+		// any of them records it, so a refusal leaves the group and its tracks as they were.
+		let commits: (() => void)[] | undefined;
+		if (this.#admits && frame.timestamp !== undefined && this.#state.total.peek() === 0) {
+			const start = frame.timestamp;
+			commits = this.#admits.map((admit) => admit(start));
+		}
 		if (
 			this.#state.total.peek() >= MAX_GROUP_FRAMES ||
 			this.#state.cacheBytes + frame.payload.byteLength > MAX_GROUP_CACHE_BYTES
@@ -285,6 +304,10 @@ export class Producer {
 
 		this.#activity = now();
 		appendFrame(this.#state, frame);
+		if (commits) {
+			this.#admits = undefined;
+			for (const commit of commits) commit();
+		}
 
 		if (this.#mirrors) {
 			for (const mirror of this.#mirrors) {
@@ -431,8 +454,11 @@ export class Consumer {
 
 	static {
 		makeConsumer = (state) => new Consumer(state);
-		hooks.groupTimestamp = (group) => group.#state.timestamp;
-		hooks.groupLatest = (group) => group.#state.latest;
+		hooks.groupTimestamp = (group) => (group.#state.unverified ? undefined : group.#state.timestamp);
+		hooks.groupLatest = (group) => (group.#state.unverified ? undefined : group.#state.latest);
+		hooks.unverifyGroup = (group) => {
+			group.#state.unverified = true;
+		};
 		hooks.groupArrived = (group) => group.#state.arrived;
 		hooks.arriveGroup = (group, at) => {
 			group.#state.arrived = at;

js/net/src/ietf/subscriber.ts

@@ -663,6 +663,7 @@ export class Subscriber {
 		// Keep the request pending until SUBSCRIBE_OK supplies immutable track metadata.
 		// Group streams already wait on the alias, so early data stays behind this response.
 		const producer = hooks.pendingTrackProducer(request);
+		hooks.receiveTrack(producer);
 		const subscription: Subscription = { track: producer, tail: new Tail() };
 
 		// Open the stream and wait for SUBSCRIBE_OK under a timeout. State

js/net/src/internal.ts

@@ -174,6 +174,11 @@ export interface TrackSequence {
 	next: number;
 	/** Open producers writing through this namespace. */
 	holders: number;
+	/**
+	 * The highest-sequence checked group start any producer of the name wrote, which bounds a
+	 * replacement's groups the way the eviction watermark bounds a track's own.
+	 */
+	bound?: { sequence: number; start: Timestamp };
 }
 
 /**
@@ -239,6 +244,12 @@ export const hooks: {
 	 * any other second end. Mirrors the Rust `finish_at_pending`; assigned by `track.ts`.
 	 */
 	endTrack: (producer: Producer, end: Location) => void;
+	// pending: /quest/m1/public-model/ingest.md
+	/**
+	 * Mark a track as a session's copy of a remote one: its groups are never checked for a
+	 * rewind, since a received timestamp may not be media time.
+	 */
+	receiveTrack: (producer: Producer) => void;
 	/** Mint a track {@link Request}; assigned by `track.ts`. */
 	makeRequest: (options: TrackRequestOptions) => Request;
 	// pending: /quest/m1/public-model/ingest.md
@@ -282,6 +293,16 @@ export const hooks: {
 	 * Throws `ProtocolViolation` if a frame at or past the cap already exists.
 	 */
 	limitGroup: (group: GroupProducer, frames: number) => void;
+	/**
+	 * Check a group's first frame against a track it joined before the frame lands. `admit`
+	 * throws `TimestampMismatch` on a rewind, or returns the commit to run once every track
+	 * the group joined accepted it.
+	 */
+	admitGroupStart: (group: GroupProducer, admit: (start: Timestamp) => () => void) => void;
+	/** Return a group producer's first timestamp, if it already holds a frame. */
+	groupStart: (group: GroupProducer) => Timestamp | undefined;
+	/** Keep a mirror of a group accepted unverified below its track's eviction watermark out of media-time expiry. */
+	unverifyGroup: (group: GroupConsumer) => void;
 	/** Return a group's first timestamp, retained even after its first frame is read. */
 	groupTimestamp: (group: GroupConsumer) => Timestamp | undefined;
 	groupLatest: (group: GroupConsumer) => Timestamp | undefined;
@@ -315,6 +336,9 @@ export const hooks: {
 	/** Name the epoch of the route an origin resolved a broadcast handle through. */
 	stampEpoch: (target: BroadcastConsumer, epoch: Epoch.Valid | undefined) => void;
 } = {
+	receiveTrack: () => {
+		throw new Error("track.ts not loaded");
+	},
 	endTrack: () => {
 		throw new Error("track.ts not loaded");
 	},
@@ -351,6 +375,15 @@ export const hooks: {
 	limitGroup: () => {
 		throw new Error("group.ts not loaded");
 	},
+	admitGroupStart: () => {
+		throw new Error("group.ts not loaded");
+	},
+	groupStart: () => {
+		throw new Error("group.ts not loaded");
+	},
+	unverifyGroup: () => {
+		throw new Error("group.ts not loaded");
+	},
 	arriveGroup: () => {
 		throw new Error("group.ts not loaded");
 	},

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

@@ -205,7 +205,7 @@ async function subscribeEnd(sequences: number[], version: Version = Version.DRAF
 	// track ending rather than resolving a subscribe against an already-closed one.
 	for (const sequence of sequences) {
 		const group = new GroupProducer(sequence);
-		group.writeFrame(textFrame("hello"));
+		group.writeFrame(textFrame("hello", Timestamp.fromMillis(sequence)));
 		group.close();
 		track.writeGroup(group);
 
@@ -641,9 +641,10 @@ async function servedSubscription(
 		serving,
 		parked: gate.parked,
 		release: gate.release,
+		// Stamped by sequence, so a group served out of order still starts in order.
 		serve(sequence: number) {
 			const group = new GroupProducer(sequence);
-			for (const frame of frames) group.writeFrame(textFrame(frame));
+			for (const frame of frames) group.writeFrame(textFrame(frame, Timestamp.fromMillis(sequence)));
 			group.close();
 			track.writeGroup(group);
 		},