feat(lite): lite-07 carries an untimed track untimed

doc/concept/moq-lite.md

@@ -216,8 +216,9 @@ doesn't match is refused. No receiver fills in an arrival time. An untimed
 group never falls past max delay, so a new subscriber starts at its latest
 group unless it names a start. Tracks from moq-lite before 05, or from
 moq-transport without a `TIMESCALE` (every track on drafts 14-16), arrive
-untimed. No moq-lite version can declare an untimed track, so on 05 and later
-its frames go out stamped with their send time.
+untimed. moq-lite 05 and 06 cannot declare an untimed track, so there its
+frames go out stamped with their send time; the work-in-progress 07 carries it
+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

drafts/draft-lcurley-moq-lite.md

@@ -172,9 +172,9 @@ Frames within a Group are numbered from 0 in the order they were produced.
 This index is not transmitted per frame; it is implied by position within the Group, anchored by the [GROUP](#group) message's `Frame Start`.
 A Group Stream normally starts at frame 0, but MAY start later when the publisher only holds (or was only asked for) part of the Group.
 
-Each frame carries a presentation timestamp expressed in the parent Track's `Timescale` (see [TRACK_INFO](#track-info)), used by the moq-lite layer for [expiration](#expiration) decisions.
-This version cannot mark a Track as having no presentation time, which a relay forwarding a Track received untimed over another protocol may need.
-Every frame and datagram of such a Track carries the publisher's send time instead, and its TRACK_INFO declares any `Timescale` for those send times.
+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.
+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.
 
 ## Positions {#positions}
 A Position is a (Group Sequence, Frame Index) pair identifying one frame within a Track.
@@ -638,6 +638,7 @@ It is not crucial to aggressively expire groups thanks to [prioritization](#prio
 It is RECOMMENDED that an application set conservative limits and only resort to expiration when data is absolutely no longer needed.
 
 A group is never expired until a later group (by sequence number) has presented a frame.
+A group of an untimed Track has no timestamps, so it is never expired.
 Once one has, the group's **timestamp age** is the difference between the *newest* frame timestamp of the latest group that has at least one frame, and this group's **reach**, defined below; the group is expired once that age meets the relevant `Max Age`.
 Timestamps are the only measure: they are consistent across relays and unaffected by buffering or jitter.
 Wall-clock reclamation of idle content is the retention cache's own policy (`Publisher Max Age`, or any implementation-defined bound), not part of this rule.
@@ -714,7 +715,7 @@ Each datagram body has the following encoding (note: there is no message length
 DATAGRAM Body {
   Subscribe ID (i)
   Group Sequence (i)
-  Timestamp (i)
+  [Timestamp (i)]
   Payload (b)
 }
 ~~~
@@ -728,7 +729,7 @@ The absolute sequence number of the group carried by this datagram.
 
 **Timestamp**:
 The absolute timestamp of the single frame in the group, expressed in the Track's `Timescale`.
-Any varint value (including 0) is valid.
+Present only on a timed Track, which the receiver learns from the subscription named by `Subscribe ID`.
 
 **Payload**:
 The frame payload, extending to the end of the datagram.
@@ -1235,8 +1236,8 @@ Older implementations interpret this as a finite duration; the sentinel fits the
 See the [Expiration](#expiration) section for more information.
 
 **Timescale**:
-The number of timestamp units per second for frame timestamps on this Track.
-It MUST be non-zero; a subscriber that receives 0 MUST reset the stream with a protocol violation.
+The number of timestamp units per second for frame timestamps on this Track, or `0` for an untimed Track, whose FRAME messages and datagrams carry no timestamp.
+lite-05 and lite-06 have no untimed encoding: there it MUST be non-zero, and a subscriber that receives 0 MUST reset the stream with a protocol violation.
 Common values include `1000` (milliseconds), `1000000` (microseconds), `48000` (audio sample rate), and `90000` (RTP video clock).
 
 ## SUBSCRIBE_OK {#subscribe-ok}
@@ -1523,14 +1524,15 @@ The FRAME message is a payload within a group.
 
 ~~~
 FRAME Message {
-  Timestamp Delta (i)
+  [Timestamp Delta (i)]
   Message Length (i)
   Payload (b)
 }
 ~~~
 
 **Timestamp Delta**:
 A signed delta from the previous frame's timestamp, in the Track's negotiated `Timescale`.
+Present only on a timed Track.
 Encoded as a zigzag-mapped variable-length integer:
 
 - Encode: `unsigned = (signed << 1) ^ (signed >> 63)` (arithmetic right shift).
@@ -1563,7 +1565,8 @@ The `Message Length` describes the payload size on the wire.
 - Added `Largest Group` and `Largest Frame` to SUBSCRIBE_OK: the publisher's largest position when it answers, which a subscriber takes as where the live feed is. A publisher MUST answer at once when the requested start is past it. Earlier versions carry no such position, so a subscriber takes the first frame instead.
 - Added `Epoch` to ANNOUNCE_START, SUBSCRIBE, TRACK, and FETCH: a UUIDv7 naming the publisher instance, or empty. A path and an Epoch name one Broadcast. A relay MAY move a subscription between routes with the same Epoch, continuing from the first frame the subscriber lacks instead of at a group boundary, and never between routes with different Epochs or none. The newest Epoch wins a path for new requests, while subscriptions to the older one stay until their route goes. Replaces the first-hop identity.
 - Added ANNOUNCE_RESTART (0x3): another publisher instance replaces a live advertisement, under a new Epoch or, without one, through another route, including a path beneath a prefix moving to another route. Older versions send ANNOUNCE_END then ANNOUNCE_START for it.
-- An untimed Track's frames and datagrams carry their send time. lite-05 and lite-06 publishers do the same.
+- An untimed Track declares a `Timescale` of 0 in TRACK_INFO, and its FRAME messages and datagrams omit the timestamp. lite-05 and lite-06 publishers declare 1000 for it and send each payload's send time instead.
+- A group of an untimed Track is never expired.
 - An open Track Stream is interest in the Track until the subscriber closes it, which a subscriber does once its Subscribe Stream has a response. lite-05 and lite-06 implementations do the same.
 
 - Assigned `moq-lite-07-wip` as this draft's protocol identifier until it is finalized as `moq-lite-07`.

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

@@ -8,7 +8,7 @@ const dec = new TextDecoder();
 test("datagram body round-trips", async () => {
 	for (const version of [Version.DRAFT_06, Version.DRAFT_07]) {
 		const dg = new Datagram(7n, 42, 1000, enc.encode("hello"));
-		const decoded = await Datagram.decode(dg.encode(version), version);
+		const decoded = await Datagram.decode(dg.encode(version), version, () => true);
 		expect(decoded.subscribe).toBe(7n);
 		expect(decoded.sequence).toBe(42);
 		expect(decoded.timestamp).toBe(1000);
@@ -33,6 +33,26 @@ test("datagram body varints follow the version's encoding", () => {
 
 test("datagram body round-trips an empty payload", async () => {
 	const dg = new Datagram(0n, 0, 0, new Uint8Array());
-	const decoded = await Datagram.decode(dg.encode(Version.DRAFT_07), Version.DRAFT_07);
+	const decoded = await Datagram.decode(dg.encode(Version.DRAFT_07), Version.DRAFT_07, () => true);
 	expect(decoded.payload.byteLength).toBe(0);
 });
+
+test("an untimed datagram body carries no timestamp", async () => {
+	const dg = new Datagram(1n, 2, undefined, enc.encode("world"));
+	const body = dg.encode(Version.DRAFT_07);
+	expect(body.byteLength).toBe(7);
+	const decoded = await Datagram.decode(body, Version.DRAFT_07, (subscribe) => {
+		expect(subscribe).toBe(1n);
+		return false;
+	});
+	expect(decoded.sequence).toBe(2);
+	expect(decoded.timestamp).toBeUndefined();
+	expect(dec.decode(decoded.payload)).toBe("world");
+});
+
+test("an untimed datagram body is refused before lite-07", () => {
+	const dg = new Datagram(1n, 2, undefined, enc.encode("world"));
+	for (const version of [Version.DRAFT_05, Version.DRAFT_06]) {
+		expect(() => dg.encode(version)).toThrow("has no untimed datagram");
+	}
+});

js/net/src/lite/datagram.ts

@@ -4,10 +4,11 @@
  * @module
  */
 import { encodeVarint, Reader } from "../stream.ts";
-import type { Version } from "./version.ts";
+import { type Version, wireTimescale } from "./version.ts";
 
 /**
- * A QUIC datagram body: `subscribe (i) | sequence (i) | timestamp (i) | payload (b)`.
+ * A QUIC datagram body: `subscribe (i) | sequence (i) | [timestamp (i)] | payload (b)`, with the
+ * timestamp present only when the subscription's track carries one (see `wireTimescale`).
  *
  * The payload runs to the datagram boundary, so unlike a size-prefixed lite message there is no
  * inner length prefix. Mirrors the Rust `lite::Datagram`; the model counterpart is
@@ -18,12 +19,12 @@ export class Datagram {
 	subscribe: bigint;
 	/** Group sequence number (shared with the track's group namespace). */
 	sequence: number;
-	/** Absolute presentation timestamp, in the track's negotiated timescale. */
-	timestamp: number;
+	/** Absolute presentation timestamp in the track's wire timescale, or undefined on a track that carries none. */
+	timestamp: number | undefined;
 	/** The frame payload, delimited by the datagram boundary. */
 	payload: Uint8Array;
 
-	constructor(subscribe: bigint, sequence: number, timestamp: number, payload: Uint8Array) {
+	constructor(subscribe: bigint, sequence: number, timestamp: number | undefined, payload: Uint8Array) {
 		this.subscribe = subscribe;
 		this.sequence = sequence;
 		this.timestamp = timestamp;
@@ -32,9 +33,14 @@ export class Datagram {
 
 	/** Encode the body to a single `Uint8Array` (no length prefix; the datagram boundary delimits it). */
 	encode(version: Version): Uint8Array {
+		// A version that always carries a timescale has no untimed body.
+		if (this.timestamp === undefined && wireTimescale(version, undefined) !== undefined) {
+			throw new Error(`version ${version} has no untimed datagram`);
+		}
+
 		const subscribe = encodeVarint(this.subscribe, version);
 		const sequence = encodeVarint(this.sequence, version);
-		const timestamp = encodeVarint(this.timestamp, version);
+		const timestamp = this.timestamp === undefined ? new Uint8Array() : encodeVarint(this.timestamp, version);
 
 		const out = new Uint8Array(
 			subscribe.byteLength + sequence.byteLength + timestamp.byteLength + this.payload.byteLength,
@@ -50,12 +56,15 @@ export class Datagram {
 		return out;
 	}
 
-	/** Decode a datagram body from the raw bytes of one QUIC datagram. */
-	static async decode(data: Uint8Array, version: Version): Promise<Datagram> {
+	/**
+	 * Decode a datagram body from the raw bytes of one QUIC datagram. `timed` reports whether the
+	 * track of the given subscribe ID carries a timestamp, which the body itself does not say.
+	 */
+	static async decode(data: Uint8Array, version: Version, timed: (subscribe: bigint) => boolean): Promise<Datagram> {
 		const r = new Reader(undefined, data, version);
 		const subscribe = await r.u62();
 		const sequence = await r.u53();
-		const timestamp = await r.u53();
+		const timestamp = timed(subscribe) ? await r.u53() : undefined;
 		const payload = await r.readAll();
 		return new Datagram(subscribe, sequence, timestamp, payload);
 	}

js/net/src/lite/publisher.ts

@@ -12,7 +12,7 @@ import type { Consumer as OriginConsumer } from "../origin.ts";
 import * as Path from "../path.ts";
 import { Priority, sendOrder } from "../priority.ts";
 import { type Reader, type Stream, Writer } from "../stream.ts";
-import { Milli, Timescale, Timestamp } from "../time.ts";
+import { Milli, type Timescale, Timestamp } from "../time.ts";
 import type * as track from "../track.ts";
 import { untilAborted } from "../util/abort.ts";
 import { Interests, LIMITS, type Limits, Slots } from "../util/limits.ts";
@@ -44,6 +44,7 @@ import {
 	resolvesStart,
 	Version,
 	waitsForSubscriberFin,
+	wireTimescale,
 } from "./version.ts";
 
 const PROBE_INTERVAL = 100; // ms
@@ -57,16 +58,8 @@ function zigzag(delta: bigint): bigint {
 }
 
 /**
- * The timescale TRACK_INFO declares for a track. Lite05+ requires one, so an untimed track
- * declares milliseconds, the scale its frames' send times go out at.
- */
-function wireTimescale(info: track.Info): Timescale {
-	return info.timescale ?? Timescale.MILLI;
-}
-
-/**
- * A frame or datagram timestamp as its raw value at the wire `timescale`. No lite version
- * encodes an absent timestamp yet, so an untimed payload carries its send time instead.
+ * A frame or datagram timestamp as its raw value at the wire `timescale`. Lite-05/06 cannot
+ * encode an untimed track, so its payloads carry their send time instead.
  */
 function wireTime(timestamp: Timestamp | undefined, timescale: Timescale): number {
 	return Math.round((timestamp ?? Timestamp.now()).as(timescale));
@@ -99,8 +92,8 @@ interface RunGroup {
 	/** The group to serve. */
 	group: group.Consumer;
 
-	/** The track's advertised timescale, applied to every frame timestamp. */
-	timescale: Timescale;
+	/** The track's wire timescale, applied to every frame timestamp, or undefined to send none. */
+	timescale: Timescale | undefined;
 
 	/** The subscription's ranking, which this stream joins for as long as it runs. */
 	priority: Priority;
@@ -169,8 +162,8 @@ function frameRange(bounds: FrameBounds, sequence: number): { start: number; end
 type ServeGroup = {
 	/** The Subscribe ID the GROUP message references. */
 	sub: bigint;
-	/** The track's advertised timescale, which every frame timestamp is converted to. */
-	timescale: Timescale;
+	/** The track's wire timescale, which every frame timestamp is converted to, or undefined to send none. */
+	timescale: Timescale | undefined;
 	/** First frame to send; anything below it was excluded by the subscription. */
 	start: number;
 	/** Last frame to send (inclusive), or undefined for the rest of the group. */
@@ -768,7 +761,8 @@ export class Publisher {
 		let controls: SubscriptionControls | undefined;
 
 		try {
-			let timescale: Timescale = Timescale.MILLI;
+			// Older drafts carry no timestamps.
+			let timescale: Timescale | undefined;
 
 			if (supportsTrackStream(this.version)) {
 				// Lite-05+ accepts implicitly: no SUBSCRIBE_OK (the immutable
@@ -781,7 +775,7 @@ export class Publisher {
 				// (accept never called, track closed) as an error here, which resets the
 				// stream.
 				const info = await untilAborted(track.info(), left.signal);
-				timescale = wireTimescale(info);
+				timescale = wireTimescale(this.version, info.timescale);
 			} else {
 				// Older drafts acknowledge with SUBSCRIBE_OK and stream frames verbatim.
 				const ok = new SubscribeOk({
@@ -935,7 +929,7 @@ export class Publisher {
 					return;
 				}
 				await this.#runFetchGroup(group, stream.writer, {
-					timescale: Timescale(info.timescale),
+					timescale: wireTimescale(this.version, info.timescale),
 					start: msg.startFrame,
 					end: msg.endFrame,
 				});
@@ -971,7 +965,7 @@ export class Publisher {
 		track: track.Subscriber,
 		stream: Writer,
 		controls: SubscriptionControls,
-		serving: { sub: bigint; broadcast: Path.Valid; timescale: Timescale; bounds: FrameBounds },
+		serving: { sub: bigint; broadcast: Path.Valid; timescale: Timescale | undefined; bounds: FrameBounds },
 	): Promise<boolean> {
 		const { sub, broadcast, timescale, bounds } = serving;
 		// Lite-05+ resolves the range on the subscribe stream: SUBSCRIBE_START once the
@@ -1292,9 +1286,8 @@ export class Publisher {
 					// Publisher Max Age: the publisher's retention bound, advertised so
 					// relays re-serve with the same window.
 					maxAge: info.maxAge,
-					// Lite05 mandates per-frame timestamps. Advertise the track's timescale;
 					// `#serveGroup` emits each frame converted to it.
-					timescale: wireTimescale(info),
+					timescale: info.timescale ?? undefined,
 				});
 			})();
 
@@ -1334,7 +1327,7 @@ export class Publisher {
 	 *
 	 * @internal
 	 */
-	async #runDatagrams(sub: bigint, track: track.Subscriber, timescale: Timescale) {
+	async #runDatagrams(sub: bigint, track: track.Subscriber, timescale: Timescale | undefined) {
 		const writer = this.#datagramWriter;
 		if (!writer) return; // Only reached with a writer (see the #datagramWriter gate).
 		const maxSize = DatagramStream.maxDatagramSize(this.#quic);
@@ -1345,7 +1338,7 @@ export class Publisher {
 				if (!datagram) return; // Track finished; #runTrack tears the subscription down.
 
 				// Convert the timestamp to the track's advertised timescale, matching #serveGroup.
-				const ts = wireTime(datagram.timestamp, timescale);
+				const ts = timescale === undefined ? undefined : wireTime(datagram.timestamp, timescale);
 				const body = new DatagramMessage(sub, datagram.sequence, ts, datagram.payload).encode(this.version);
 
 				// No group fallback: drop anything that doesn't fit a single datagram.
@@ -1364,7 +1357,7 @@ export class Publisher {
 	}
 
 	// Serialize a fetched group's frames onto the FETCH stream as bare records: each a
-	// zigzag-delta timestamp (at the track's advertised timescale) followed by size + bytes.
+	// zigzag-delta timestamp (at the track's wire timescale, if any) followed by size + bytes.
 	async #runFetchGroup(
 		group: group.Consumer,
 		stream: Writer,
@@ -1384,9 +1377,11 @@ export class Publisher {
 			const frame = await race([group.readFrame(), stream.closed]);
 			if (!frame) break;
 
-			const ts = BigInt(wireTime(frame.timestamp, timescale));
-			await stream.u62(zigzag(ts - prevTs));
-			prevTs = ts;
+			if (timescale !== undefined) {
+				const ts = BigInt(wireTime(frame.timestamp, timescale));
+				await stream.u62(zigzag(ts - prevTs));
+				prevTs = ts;
+			}
 
 			await stream.u53(frame.payload.byteLength);
 			await stream.write(frame.payload);
@@ -1445,9 +1440,6 @@ export class Publisher {
 				await msg.encode(stream, this.version);
 			});
 
-			// Lite05+ prefixes every frame with a zigzag-delta timestamp at the track's
-			// advertised timescale; older drafts omit it.
-			const timestamps = supportsTrackStream(this.version);
 			let prevTs = 0n;
 			// Whether the cursor ever reached the requested start, which decides how the
 			// end of the group is read below.
@@ -1472,8 +1464,8 @@ export class Publisher {
 					if (read.sequence < startFrame) continue;
 					if (endFrame !== undefined && read.sequence > endFrame) break;
 
-					if (timestamps) {
-						// Convert each frame to the track's advertised timescale.
+					if (timescale !== undefined) {
+						// Convert each frame to the track's wire timescale.
 						const ts = BigInt(wireTime(read.frame.timestamp, timescale));
 						await hooks.guardGroup(group, () => stream.u62(zigzag(ts - prevTs)));
 						prevTs = ts;