diff --git a/doc/concept/moq-lite.md b/doc/concept/moq-lite.md index be643b0c2f..6383d9da23 100644 --- a/doc/concept/moq-lite.md +++ b/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 diff --git a/drafts/draft-lcurley-moq-lite.md b/drafts/draft-lcurley-moq-lite.md index 222a7173fa..3feb811f58 100644 --- a/drafts/draft-lcurley-moq-lite.md +++ b/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,7 +1524,7 @@ The FRAME message is a payload within a group. ~~~ FRAME Message { - Timestamp Delta (i) + [Timestamp Delta (i)] Message Length (i) Payload (b) } @@ -1531,6 +1532,7 @@ FRAME Message { **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`. diff --git a/js/net/src/lite/datagram.test.ts b/js/net/src/lite/datagram.test.ts index 80abce58db..2e2893fe2b 100644 --- a/js/net/src/lite/datagram.test.ts +++ b/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"); + } +}); diff --git a/js/net/src/lite/datagram.ts b/js/net/src/lite/datagram.ts index 6fbb0df9d0..be4c983558 100644 --- a/js/net/src/lite/datagram.ts +++ b/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 { + /** + * 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 { 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); } diff --git a/js/net/src/lite/publisher.ts b/js/net/src/lite/publisher.ts index d3cb53bc6e..b24cae1236 100644 --- a/js/net/src/lite/publisher.ts +++ b/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 { 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; diff --git a/js/net/src/lite/subscriber.ts b/js/net/src/lite/subscriber.ts index d8e7368030..1313aec26e 100644 --- a/js/net/src/lite/subscriber.ts +++ b/js/net/src/lite/subscriber.ts @@ -820,7 +820,7 @@ export class Subscriber { // register it again or send its SUBSCRIBE. state.cancel.signal.throwIfAborted(); producer = request.accept(this.#toModelInfo(info)); - timescale.set(info.timescale); + timescale.set(info.timescale ?? 0); } else { // Older drafts negotiate nothing per-track: verbatim frames with no timeline. producer = request.accept(); @@ -918,7 +918,7 @@ export class Subscriber { // Map the wire TRACK_INFO onto the model track.Info a producer/consumer holds. #toModelInfo(info: TrackInfo): track.Info { return { - timescale: Time.Timescale(info.timescale), + timescale: info.timescale, // Publisher Max Age rides on the wire, so the local media-time budget // matches what the upstream advertises (relays re-serve with the same bound). maxAge: info.maxAge === undefined ? undefined : Time.Milli(info.maxAge), @@ -1032,7 +1032,7 @@ export class Subscriber { throw err; } - void this.#runFetchResponse(accepted.stream, group, Time.Timescale(accepted.info.timescale)); + void this.#runFetchResponse(accepted.stream, group, accepted.info.timescale ?? 0); } catch (err: unknown) { group.close(error(err)); throw err; @@ -1077,9 +1077,10 @@ export class Subscriber { ); } - // Read the FETCH response (bare zigzag-delta-timestamped frames) into the group, then - // FIN. A stream-level failure aborts the group so its reader observes the gap. - async #runFetchResponse(stream: Stream, group: netGroup.Producer, timescale: Time.Timescale): Promise { + // Read the FETCH response (bare frames, zigzag-delta-timestamped unless `timescale` is 0) + // into the group, then FIN. A stream-level failure aborts the group so its reader observes + // the gap. + async #runFetchResponse(stream: Stream, group: netGroup.Producer, timescale: number): Promise { try { const decode = frameDecoder(timescale); @@ -1397,17 +1398,23 @@ export class Subscriber { // Decode one datagram body and hand it to the matching subscription's producer. Drops the // datagram (best-effort) if the subscription is unknown/closed or its timescale isn't resolved. async #routeDatagram(payload: Uint8Array): Promise { - const dg = await DatagramMessage.decode(payload, this.version); + // An unknown or unresolved subscription's datagram is dropped below, however it parses. + const dg = await DatagramMessage.decode( + payload, + this.version, + (id) => !!this.#subscribes.get(id)?.timescale.peek(), + ); const entry = this.#subscribes.get(dg.subscribe); if (!entry) return; // Unknown or already-closed subscription. - // Datagrams are lite-05+, which always negotiates a timescale; if it hasn't resolved - // yet (the datagram raced ahead of TRACK_INFO), drop rather than guess. + // If the timescale hasn't resolved yet (the datagram raced ahead of TRACK_INFO), drop + // rather than guess. Zero means the track carries no timestamps. const scale = entry.timescale.peek(); - if (!scale) return; + if (scale === undefined) return; - const timestamp = new Time.Timestamp(dg.timestamp, Time.Timescale(scale)); + const timestamp = + dg.timestamp === undefined ? undefined : new Time.Timestamp(dg.timestamp, Time.Timescale(scale)); // A datagram's sequence is never owed a stream, so it never holds the tail open. entry.tail.account(dg.sequence, dg.sequence + 1); entry.track.insertDatagram(dg.sequence, timestamp, dg.payload); diff --git a/js/net/src/lite/tail.test.ts b/js/net/src/lite/tail.test.ts index 823b56bb36..09c9bf12b5 100644 --- a/js/net/src/lite/tail.test.ts +++ b/js/net/src/lite/tail.test.ts @@ -5,7 +5,7 @@ import { randomHop } from "../hop.ts"; import { createMockTransportPair } from "../mock.ts"; import * as Path from "../path.ts"; import { Reader, Stream } from "../stream.ts"; -import { Milli } from "../time.ts"; +import { Milli, Timescale } from "../time.ts"; import type { Groups } from "../track.ts"; import { Group as GroupMessage } from "./group.ts"; import { StreamId } from "./stream.ts"; @@ -61,7 +61,7 @@ async function subscribed(version: Version, maxDelay = GRACE, groups?: Groups) { if (!info) throw new Error("the subscriber never asked for TRACK_INFO"); expect(await info.reader.u53()).toBe(StreamId.Track); await TrackMessage.decode(info.reader, version); - await new TrackInfo({ maxAge: 60_000 }).encode(info.writer, version); + await new TrackInfo({ maxAge: 60_000, timescale: Timescale.MILLI }).encode(info.writer, version); info.close(); const sub = await Stream.accept(pair.server, version); diff --git a/js/net/src/lite/track.test.ts b/js/net/src/lite/track.test.ts index 58b7cc12c2..081eb86125 100644 --- a/js/net/src/lite/track.test.ts +++ b/js/net/src/lite/track.test.ts @@ -43,18 +43,36 @@ test("TrackInfo round-trips on draft-05", async () => { const got = await TrackInfo.decode(reader, Version.DRAFT_05); expect(got.priority).toBe(7); expect(got.maxAge).toBe(2000); - expect(got.timescale).toBe(90000); + expect(got.timescale).toBe(Timescale(90000)); }); test("TrackInfo for Rust's default Info matches cross-language wire bytes", async () => { - // Neither language has a default timescale; an untimed track declares milliseconds on lite-05. - const model = infoDefaults({ timescale: Timescale.MILLI }); + // The default track is untimed, which lite-05 declares as milliseconds. + const model = infoDefaults(); const info = new TrackInfo({ ...model, timescale: model.timescale ?? undefined }); expect(await bytes((w) => info.encode(w, Version.DRAFT_05), Version.DRAFT_05)).toEqual( new Uint8Array([0x0c, 0x7f, 0x00, 0xc0, 0x1f, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0x43, 0xe8]), ); }); +test("TrackInfo carries an untimed track as zero on lite-07 and milliseconds before", async () => { + const info = new TrackInfo({ priority: 1 }); + const lite07 = await bytes((w) => info.encode(w, Version.DRAFT_07), Version.DRAFT_07); + expect(lite07.at(-1)).toBe(0); + const got = await TrackInfo.decode(new Reader(undefined, lite07, Version.DRAFT_07), Version.DRAFT_07); + expect(got.timescale).toBeUndefined(); + + for (const version of [Version.DRAFT_05, Version.DRAFT_06]) { + const encoded = await bytes((w) => info.encode(w, version), version); + const legacy = await TrackInfo.decode(new Reader(undefined, encoded, version), version); + expect(legacy.timescale).toBe(Timescale.MILLI); + // Zero is malformed where no untimed encoding exists. A one-byte timescale makes room for it. + const zero = await bytes((w) => new TrackInfo({ priority: 1, timescale: 1 }).encode(w, version), version); + zero[zero.length - 1] = 0; + await expect(TrackInfo.decode(new Reader(undefined, zero, version), version)).rejects.toThrow(RangeError); + } +}); + test("Track request round-trips on draft-05", async () => { const msg = new Track(Path.from("room"), "video"); const reader = new Reader( @@ -86,7 +104,7 @@ test("TrackInfo round-trips valid boundary metadata", async () => { const got = await TrackInfo.decode(reader, Version.DRAFT_05); expect(got.priority).toBe(255); expect(got.maxAge).toBe(Number.MAX_SAFE_INTEGER - 1); - expect(got.timescale).toBe(Number.MAX_SAFE_INTEGER); + expect(got.timescale).toBe(Timescale(Number.MAX_SAFE_INTEGER)); }); test("TrackInfo refuses invalid fields at construction", () => { @@ -117,10 +135,10 @@ test("mutated TrackInfo fields emit no bytes on encode", async () => { info.priority = 256; }, (info: TrackInfo) => { - info.timescale = 0; + info.timescale = 0 as Timescale; }, (info: TrackInfo) => { - info.timescale = Number.MAX_SAFE_INTEGER + 1; + info.timescale = (Number.MAX_SAFE_INTEGER + 1) as Timescale; }, (info: TrackInfo) => { info.maxAge = Number.MAX_SAFE_INTEGER + 1; diff --git a/js/net/src/lite/track.ts b/js/net/src/lite/track.ts index 146c8cc3a6..a787c20394 100644 --- a/js/net/src/lite/track.ts +++ b/js/net/src/lite/track.ts @@ -4,7 +4,7 @@ import type { Reader, Writer } from "../stream.ts"; import { Timescale } from "../time.ts"; import { decodeEpoch, encodeEpoch } from "./epoch.ts"; import * as Message from "./message.ts"; -import { hasGroupOrder, Version } from "./version.ts"; +import { hasGroupOrder, Version, wireTimescale } from "./version.ts"; // The Track Stream (0x6) is draft-05+ only. function guardTrack(version: Version) { @@ -73,16 +73,16 @@ export class TrackInfo { */ maxAge?: number; /** - * Per-frame timestamp scale (units per second). Mandatory on Lite05: a real - * (non-zero) scale, and every frame on the wire is prefixed with a zigzag-delta - * timestamp at this scale. + * Per-frame timestamp scale (units per second), or undefined for an untimed track. + * Lite-07 encodes undefined as zero; lite-05/06 have no untimed encoding, so it goes + * out as milliseconds (see {@link wireTimescale}). */ - timescale: number; + timescale?: Timescale; constructor({ priority = 0, maxAge, - timescale = Timescale.MILLI, + timescale, }: { priority?: number; maxAge?: number; @@ -96,7 +96,7 @@ export class TrackInfo { } this.priority = priority; this.maxAge = maxAge; - this.timescale = Timescale(timescale); + this.timescale = timescale === undefined ? undefined : Timescale(timescale); } async #encode(w: Writer, version: Version) { @@ -106,7 +106,7 @@ export class TrackInfo { const legacy = version === Version.DRAFT_05 || version === Version.DRAFT_06; const age = this.maxAge === undefined ? undefined : BigInt(this.maxAge); await w.u62(legacy ? (age ?? BigInt(Number.MAX_SAFE_INTEGER)) : age === undefined ? 0n : age + 1n); - await w.u53(this.timescale); + await w.u53(wireTimescale(version, this.timescale) ?? 0); } static async #decode(r: Reader, version: Version): Promise { @@ -124,7 +124,9 @@ export class TrackInfo { if (age !== undefined && age > BigInt(Number.MAX_SAFE_INTEGER)) throw new RangeError("maxAge exceeds safe milliseconds"); const maxAge = age === undefined ? undefined : Number(age); - const timescale = await r.u53(); + // Lite-05/06 have no untimed encoding, so zero is malformed there. + const scale = await r.u53(); + const timescale = scale === 0 && !legacy ? undefined : Timescale(scale); return new TrackInfo({ priority, maxAge, timescale }); } @@ -138,7 +140,7 @@ export class TrackInfo { if (this.maxAge !== undefined && (!Number.isSafeInteger(this.maxAge) || this.maxAge < 0)) { throw new RangeError(`maxAge must be a safe non-negative integer: ${this.maxAge}`); } - Timescale(this.timescale); + if (this.timescale !== undefined) Timescale(this.timescale); return Message.encode(w, (w) => this.#encode(w, version)); } diff --git a/js/net/src/lite/version.ts b/js/net/src/lite/version.ts index 91c3df2169..4c94a1075a 100644 --- a/js/net/src/lite/version.ts +++ b/js/net/src/lite/version.ts @@ -1,3 +1,5 @@ +import { Timescale } from "../time.ts"; + export const Version = { DRAFT_01: 0xff0dad01, DRAFT_02: 0xff0dad02, @@ -408,3 +410,23 @@ export function hasLargest(version: Version): boolean { return true; } } + +/** + * The timescale the wire carries for a track with `timescale`, or `undefined` when its frames + * and datagrams carry no timestamp. Lite-05 and lite-06 always carry one, so an untimed track + * declares milliseconds there and its payloads carry their send times instead. + */ +export function wireTimescale(version: Version, timescale: Timescale | null | undefined): Timescale | undefined { + switch (version) { + case Version.DRAFT_01: + case Version.DRAFT_02: + case Version.DRAFT_03: + case Version.DRAFT_04: + return undefined; + case Version.DRAFT_05: + case Version.DRAFT_06: + return timescale ?? Timescale.MILLI; + default: + return timescale ?? undefined; + } +} diff --git a/js/net/src/untimed.test.ts b/js/net/src/untimed.test.ts index ca918dcb09..d9e1ec49f8 100644 --- a/js/net/src/untimed.test.ts +++ b/js/net/src/untimed.test.ts @@ -30,7 +30,7 @@ const WIRES: Wire[] = [ { name: "lite-04", protocol: Lite.ALPN_04, declares: false, stamps: false }, { name: "lite-05", protocol: Lite.ALPN_05, declares: true, stamps: true }, { name: "lite-06", protocol: Lite.ALPN_06, declares: true, stamps: true }, - { name: "lite-07-wip", protocol: Lite.ALPN_07_WIP, declares: true, stamps: true }, + { name: "lite-07-wip", protocol: Lite.ALPN_07_WIP, declares: true, stamps: false }, { name: "ietf-14", protocol: "", version: Ietf.Version.DRAFT_14, declares: false, stamps: false }, { name: "ietf-16", protocol: Ietf.ALPN.DRAFT_16, declares: false, stamps: false }, { name: "ietf-17", protocol: Ietf.ALPN.DRAFT_17, declares: true, stamps: false }, @@ -187,3 +187,37 @@ for (const wire of WIRES) { } }); } + +for (const wire of WIRES.filter((wire) => wire.name === "lite-06" || wire.name === "lite-07-wip")) { + test(`${wire.name}: an untimed datagram arrives untimed, or with a send time where the wire must stamp`, async () => { + const pair = createMockTransportPair(wire.protocol); + const origin = new OriginProducer(); + const [client, server] = await Promise.all([ + connect({ url, transport: pair.client }), + accept({ transport: pair.server, url, publish: origin.consume() }), + ]); + + const broadcast = origin.createBroadcast(Path.from("test")); + broadcast.announce(); + const producer = broadcast.createTrack("video", {}); + const subscriber = wireOf(client).consume(Path.from("test")).track("video").subscribe(); + + // Datagrams aren't cached, so a group read back first proves the subscription is live. + const group = producer.appendGroup(); + group.writeFrame({ payload: new Uint8Array([0]) }); + group.close(); + expect(await subscriber.recvGroup()).toBeDefined(); + + producer.appendDatagram(undefined, new Uint8Array([7])); + const datagram = await subscriber.recvDatagram(); + + expect(datagram?.payload).toEqual(new Uint8Array([7])); + if (wire.stamps) expect(datagram?.timestamp).toBeDefined(); + else expect(datagram?.timestamp).toBeUndefined(); + + broadcast.close(); + subscriber.close(); + client.abort(); + server.abort(); + }); +} diff --git a/quest/m1/README.md b/quest/m1/README.md index 0074331c06..c5ece0d5fb 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -83,7 +83,6 @@ above them. - [SUBSCRIBE_DROP](/quest/m1/subscribe-drop.md) - every stream group in a lite subscription arrives or is dropped by name, and lite-07 drops its stream count for it - [Subscribe ranges](/quest/m1/subscribe-ranges/README.md) - a lite-07 SUBSCRIBE asks for past and live ranges with absolute or relative ends, in either order, and replaces FETCH; merged ranges never starve a subscriber; relays fill misses by range, including over moq-transport - [Live largest location](/quest/m1/largest.md) - `largest()` replaces `latest()` in Rust and JS as a live stream of the track's largest location, known before the first group from the publisher's report -- [Untimed lite-07](/quest/m1/lite-untimed.md) - lite-07 carries an untimed track in both languages; lite-05/06 write send time - [Live TS clock validation](/quest/m1/ts-live-clock.md) - resolve the local PCR-rate failure with reproducible timing evidence and CI controls - [Slow source fails](/quest/m1/slow-source.md) - an export whose source runs further off real time than the output may follow fails promptly naming the rate, instead of dropping frames for hours - [Same-instance TS resume](/quest/m1/ts-same-instance-resume.md) - prove valid continuation and fix the unexpected final exporter failure without weakening identity diff --git a/quest/m1/lite-untimed.md b/quest/m1/lite-untimed.md deleted file mode 100644 index 7d2c7f94d4..0000000000 --- a/quest/m1/lite-untimed.md +++ /dev/null @@ -1,49 +0,0 @@ -# [S] lite-07 carries an untimed track - -## Goal - -On moq-lite-07, an untimed track's frames and datagrams cross the wire as -untimed in both Rust and JS, so absence survives a lite relay hop. lite-05 and lite-06 -keep writing the encoder's send time, documented as a downgrade for old -peers. Covers `rs/moq-net`, `js/net` and `drafts/draft-lcurley-moq-lite.md`. - -## Plan - -Decided (2026-10-05, maintainer): timedness is per track (the untimed -model, [#4822](https://github.com/moq-dev/moq/pull/4822)), and an untimed track sends no -TIMESCALE and no Timestamp. On lite-07 that likely means an optional -Timescale in TRACK_INFO and no Timestamp fields on an untimed track. The -2026-10-01 shift-by-one (0 means an absent timestamp) is dropped: it served an -absent timestamp inside a timed track, which no longer exists (settled -2026-10-06: a mismatched frame is refused). - -Decided (2026-10-08, with [#4826](https://github.com/moq-dev/moq/pull/4826)): -CMAF decoders anchor a fragment to its frame timestamp and fall back to `tfdt` -only on an untimed frame. Until this lands, an untimed CMAF track (a -draft-14 publisher such as MOQtail) relayed to a lite-05+ subscriber anchors -to the relay's send times, and lite-05/06 peers keep doing so. Shipped anyway: -the invented timestamp is this quest's to remove. - -Decided (2026-10-02): one PR for both languages. Both model quests have -landed (#4822). Shipping one language first would break Rust-JS interop on -lite-07-wip in between. - -Rust has no `Timescale` default: the lite encoder and the datagram decoder -name `Timescale::MILLI` for an untimed track's send times, which this quest -replaces on lite-07. - -Draft work: - -- The draft describes only the current wire. Add the lite-07 encoding, the - lite-05/06 downgrade, and a lite-07 changelog entry. -- Fix the text that assumes every frame is timed: "each frame carries a - presentation timestamp", the datagram's "any varint value (including 0) is - valid", and expiration's "reach" (the first frame timestamp of the next - group) for an untimed group. -- Check how a lite track that has no timescale says so on lite-07. - -Test: an untimed frame and an untimed datagram round-trip Rust-to-JS and -JS-to-Rust on lite-07. lite-06 still receives a timestamp. Run -`just drafts check` and `just test interop --all`. - -Public API: none. Wire: lite-07-wip only, which is unpublished. diff --git a/quest/m1/lite07-finalize.md b/quest/m1/lite07-finalize.md index f0490a1cdd..4f0c63abab 100644 --- a/quest/m1/lite07-finalize.md +++ b/quest/m1/lite07-finalize.md @@ -79,7 +79,6 @@ is published; older versions are unchanged. - [SUBSCRIBE_DROP](/quest/m1/subscribe-drop.md) - lite-07 restores SUBSCRIBE_DROP in place of `Stream Count` - [FETCH max-delay](/quest/m1/fetch-max-delay.md) - historical requests have a reader budget, carried by FETCH while present and by ranges after its removal - [Subscribe ranges](/quest/m1/subscribe-ranges/README.md) - SUBSCRIBE carries ranges and an order and lite FETCH is gone, in Rust and JS -- [Untimed lite-07](/quest/m1/lite-untimed.md) - an untimed track crosses the wire untimed - [Announcement takeover](/quest/m1/announce-takeover.md) - ANNOUNCE_TAKEOVER preserves a proven instance under a new independent route ## Related diff --git a/quest/m1/publish-timestamp.md b/quest/m1/publish-timestamp.md index aec7b48dc4..1f7de4dae4 100644 --- a/quest/m1/publish-timestamp.md +++ b/quest/m1/publish-timestamp.md @@ -16,7 +16,7 @@ timestamp is optional end to end, on the wire and in every API, and absent means untimed. A library never substitutes now, wall clock, or arrival time. How absence travels through the model and over the wire is the untimed model ([#4822](https://github.com/moq-dev/moq/pull/4822)) and -[lite-07 encodes an absent timestamp](/quest/m1/lite-untimed.md); this quest +lite-07's untimed encoding; this quest removes every place that fills one in. Today `Timed.at: None` means "stamp when written": moq-json and moq-flate diff --git a/rs/moq-net/src/fuzz.rs b/rs/moq-net/src/fuzz.rs index 46e31e30df..9f40914cd2 100644 --- a/rs/moq-net/src/fuzz.rs +++ b/rs/moq-net/src/fuzz.rs @@ -106,18 +106,22 @@ where /// [`roundtrip`] for a datagram body, which runs to the end of its buffer. fn datagram(data: &[u8], version: lite::Version) -> bool { - let Ok(decoded) = lite::Datagram::decode(Bytes::copy_from_slice(data), version) else { - return false; - }; - - let Ok(first) = decoded.encode_bytes(version) else { - return true; - }; + // The body doesn't say whether it carries a timestamp, so try both readings. + let mut any = false; + for timed in [true, false] { + let Ok(decoded) = lite::Datagram::decode(Bytes::copy_from_slice(data), version, |_| timed) else { + continue; + }; + any = true; - let echo = lite::Datagram::decode(first, version).expect("could not decode our own encoding"); - assert_eq!(echo, decoded, "datagram did not survive a round trip"); + let Ok(first) = decoded.encode_bytes(version) else { + continue; + }; - true + let echo = lite::Datagram::decode(first, version, |_| timed).expect("could not decode our own encoding"); + assert_eq!(echo, decoded, "datagram did not survive a round trip"); + } + any } /// Decode one moq-lite wire object from arbitrary bytes. @@ -356,7 +360,7 @@ impl LiteSample { Self::Datagram => lite::Datagram { subscribe: 3, sequence: 1_234, - timestamp: 1_234_567_890, + timestamp: Some(1_234_567_890), payload: bytes::Bytes::new(), } .encode(w, version) @@ -389,7 +393,7 @@ impl LiteSample { } Self::Group => lite::Group::decode(r, version).map(|_| 1).unwrap(), Self::Subscribe => lite::Subscribe::decode(r, version).map(|_| 1).unwrap(), - Self::Datagram => lite::Datagram::decode(bytes::Bytes::copy_from_slice(data), version) + Self::Datagram => lite::Datagram::decode(bytes::Bytes::copy_from_slice(data), version, |_| true) .map(|_| 1) .unwrap(), Self::Setup => lite::Setup::decode(r, version).map(|_| 1).unwrap(), @@ -552,7 +556,7 @@ impl Default for Messages { lite_info: lite::TrackInfo { priority: 1, max_age: Some(Duration::from_secs(10)), - timescale: crate::Timescale::MICRO, + timescale: Some(crate::Timescale::MICRO), }, lite_group: lite::Group { subscribe: 7, diff --git a/rs/moq-net/src/lite/datagram.rs b/rs/moq-net/src/lite/datagram.rs index 90055d2fcc..e573d9e99b 100644 --- a/rs/moq-net/src/lite/datagram.rs +++ b/rs/moq-net/src/lite/datagram.rs @@ -1,9 +1,10 @@ //! Wire-level QUIC datagram body for moq-lite-05 (ยง6.4). //! //! One datagram carries a single-frame group routed over an existing subscription. The body is -//! `subscribe (i) | sequence (i) | timestamp (i) | payload (b)`; the payload runs to the datagram -//! boundary, so unlike a [`super::Message`] there is no inner length prefix. The model counterpart -//! is [`crate::Datagram`]. +//! `subscribe (i) | sequence (i) | [timestamp (i)] | payload (b)`, with the timestamp present only +//! when the subscription's track carries one (see [`Version::wire_timescale`]). The payload runs to +//! the datagram boundary, so unlike a [`super::Message`] there is no inner length prefix. The model +//! counterpart is [`crate::Datagram`]. use bytes::Bytes; @@ -18,8 +19,9 @@ pub struct Datagram { pub subscribe: u64, /// Group sequence number (shared with the track's group namespace). pub sequence: u64, - /// Absolute presentation timestamp, in the track's negotiated timescale. - pub timestamp: u64, + /// Absolute presentation timestamp in the track's wire timescale, or `None` on a track + /// that carries no timestamps. + pub timestamp: Option, /// The frame payload, delimited by the datagram boundary. pub payload: Bytes, } @@ -30,9 +32,16 @@ impl Encode for Datagram { return Err(EncodeError::Version); } + // A version that always carries a timescale has no untimed body. + if self.timestamp.is_none() && version.wire_timescale(None).is_some() { + return Err(EncodeError::Version); + } + w.varint(self.subscribe)?; w.varint(self.sequence)?; - w.varint(self.timestamp)?; + if let Some(timestamp) = self.timestamp { + w.varint(timestamp)?; + } // Payload runs to the datagram boundary: written raw, no length prefix. w.slice(&self.payload); @@ -42,7 +51,10 @@ impl Encode for Datagram { impl Datagram { /// Decode a whole datagram body. The payload shares `buf` rather than copying it. - pub fn decode(buf: Bytes, version: Version) -> Result { + /// + /// `timed` reports whether the track of the given subscribe ID carries a timestamp, + /// which the body itself does not say. + pub fn decode(buf: Bytes, version: Version, timed: impl FnOnce(u64) -> bool) -> Result { if !version.has_datagrams() { return Err(DecodeError::Version); } @@ -50,7 +62,7 @@ impl Datagram { let mut r = Decoder::new(&buf, version.into()); let subscribe = r.varint()?; let sequence = r.varint()?; - let timestamp = r.varint()?; + let timestamp = if timed(subscribe) { Some(r.varint()?) } else { None }; // Everything remaining is the payload (the datagram boundary delimits it). let payload = buf.slice(buf.len() - r.remaining()..); @@ -73,11 +85,11 @@ mod test { let original = Datagram { subscribe: 7, sequence: 42, - timestamp: 1_000, + timestamp: Some(1_000), payload: Bytes::from_static(b"hello"), }; let buf = original.encode_bytes(Version::Lite05).unwrap(); - let decoded = Datagram::decode(buf, Version::Lite05).unwrap(); + let decoded = Datagram::decode(buf, Version::Lite05, |_| true).unwrap(); assert_eq!(decoded, original, "payload has no trailing length prefix"); } @@ -86,11 +98,11 @@ mod test { let original = Datagram { subscribe: 0, sequence: 0, - timestamp: 0, + timestamp: Some(0), payload: Bytes::new(), }; let buf = original.encode_bytes(Version::Lite05).unwrap(); - let decoded = Datagram::decode(buf, Version::Lite05).unwrap(); + let decoded = Datagram::decode(buf, Version::Lite05, |_| true).unwrap(); assert_eq!(decoded, original); } @@ -101,7 +113,7 @@ mod test { let dg = Datagram { subscribe: 1, sequence: 2, - timestamp: 3, + timestamp: Some(3), payload: Bytes::from_static(b"world"), }; let buf = dg.encode_bytes(Version::Lite05).unwrap(); @@ -110,18 +122,53 @@ mod test { assert_eq!(&buf[3..], b"world"); } + #[test] + fn untimed_omits_the_timestamp() { + let dg = Datagram { + subscribe: 1, + sequence: 2, + timestamp: None, + payload: Bytes::from_static(b"world"), + }; + let buf = dg.encode_bytes(Version::Lite07).unwrap(); + assert_eq!(buf.len(), 7); + assert_eq!(&buf[2..], b"world"); + let decoded = Datagram::decode(buf, Version::Lite07, |id| { + assert_eq!(id, 1); + false + }) + .unwrap(); + assert_eq!(decoded, dg); + } + + #[test] + fn untimed_needs_lite07() { + let dg = Datagram { + subscribe: 1, + sequence: 2, + timestamp: None, + payload: Bytes::from_static(b"world"), + }; + for version in [Version::Lite05, Version::Lite06] { + assert!( + matches!(dg.encode_bytes(version), Err(EncodeError::Version)), + "{version}" + ); + } + } + #[test] fn rejects_old_versions() { let dg = Datagram { subscribe: 1, sequence: 2, - timestamp: 3, + timestamp: Some(3), payload: Bytes::from_static(b"x"), }; assert!(matches!(dg.encode_bytes(Version::Lite04), Err(EncodeError::Version))); assert!(matches!( - Datagram::decode(Bytes::from_static(b"\x01\x02\x03x"), Version::Lite04), + Datagram::decode(Bytes::from_static(b"\x01\x02\x03x"), Version::Lite04, |_| true), Err(DecodeError::Version) )); } diff --git a/rs/moq-net/src/lite/publisher.rs b/rs/moq-net/src/lite/publisher.rs index 0d8feb0d3a..55fbe18a07 100644 --- a/rs/moq-net/src/lite/publisher.rs +++ b/rs/moq-net/src/lite/publisher.rs @@ -79,8 +79,8 @@ struct Shared { // Control streams still serving the peer data, which a draining close waits for. owed: AtomicUsize, interests: Interests, - // The send time an untimed track's frames carry, since no lite version can mark them - // untimed yet (see `wire_timestamp`). + // The send time an untimed track's frames carry on Lite05/06, which can't mark them + // untimed (see `wire_timestamp`). runtime: crate::time::Clock, estimate: kio::ProducerWeak, } @@ -1881,13 +1881,11 @@ impl Request for TrackInfoServe { ) -> Poll, Error>> { let info = ready!(self.querying.poll_ok(waiter))?; - // TRACK_INFO only flows on Lite05+ (the encode errors otherwise), where the - // timescale is mandatory. An untimed track declares milliseconds, the scale its - // frames' send times go out at (see `wire_timestamp`). + // TRACK_INFO only flows on Lite05+ (the encode errors otherwise). writer.buffer(&lite::TrackInfo { priority: info.priority, max_age: info.max_age, - timescale: info.timescale.unwrap_or(crate::Timescale::MILLI), + timescale: info.timescale, })?; Poll::Ready(Ok(ControlFlow::Break(()))) } @@ -1989,16 +1987,10 @@ impl Request for SubscribeServe { unreachable!() }; - // Per-frame timestamps require a wire format that carries them. Lite05+ - // prefixes every frame with a zigzag-delta timestamp at the track's - // timescale; older drafts have no wire field, so `None` here means - // "don't emit the prefix" (the frames still carry timestamps in the - // model, just not on this wire). - let timescale = if shared.version.has_track_stream() { - Some(track.info().timescale.unwrap_or(crate::Timescale::MILLI)) - } else { - None - }; + // `None` omits the per-frame timestamp prefix: an untimed track on + // Lite07+, or any track before Lite05 (whose frames still carry + // timestamps in the model, just not on this wire). + let timescale = shared.version.wire_timescale(track.info().timescale); // Lite05+ accepts implicitly: no SUBSCRIBE_OK, the immutable properties // live in TRACK_INFO, and the resolved range arrives as @@ -2159,13 +2151,8 @@ impl Request for FetchServe { // `group::Fetch::frame_start`). group.end_at(msg.end_frame.map_or(Bound::Unbounded, Bound::Included)); - // FETCH is gated to lite-05+, which learned the track timescale via - // TRACK_INFO. - let timescale = if shared.version.has_track_stream() { - Some(group.timescale().unwrap_or(crate::Timescale::MILLI)) - } else { - None - }; + // Matches the timescale TRACK_INFO declared for the track. + let timescale = shared.version.wire_timescale(group.timescale()); *self = Self::Serve { group, @@ -2917,9 +2904,9 @@ fn buffer_frame_info( /// A frame or datagram timestamp as its raw value at the wire `timescale`. /// -/// No lite version encodes an absent timestamp yet, so a payload on an untimed track -/// carries its send time on `runtime` instead, at the millisecond scale its TRACK_INFO -/// declares. A timed payload is already at the track's timescale. +/// Lite05/06 cannot encode an untimed track, so its payloads carry their send time on +/// `runtime` instead, at the millisecond scale its TRACK_INFO declares. A timed payload +/// is already at the track's timescale. fn wire_timestamp( timestamp: Option, timescale: crate::Timescale, @@ -3143,10 +3130,10 @@ struct Subscription { /// Last steer observed by this clone, so a change only fires once. steer_seen: Steer, version: Version, - /// Negotiated timestamp scale for this track. `Some(_)` on lite-05+ after - /// TRACK_INFO; used to validate per-frame timestamps before encoding. + /// The track's wire timescale, or `None` when its frames and datagrams carry no + /// timestamp (see [`Version::wire_timescale`]). timescale: Option, - /// The clock an untimed track's send times are read from. + /// The clock an untimed track's Lite05/06 send times are read from. runtime: crate::time::Clock, /// The group streams this subscription opened, shared by every group it serves. opens: Arc, @@ -3174,12 +3161,11 @@ impl Subscription { /// transport's datagram limit or the send fails (congestion / no capacity right now). /// Returns whether it was handed to the transport. fn serve_datagram(&mut self, datagram: crate::Datagram) -> bool { - // Datagrams are lite-05+, which always declares a timescale in TRACK_INFO. - let Ok(timestamp) = wire_timestamp( - datagram.timestamp, - self.timescale.unwrap_or(crate::Timescale::MILLI), - &self.runtime, - ) else { + let Ok(timestamp) = self + .timescale + .map(|timescale| wire_timestamp(datagram.timestamp, timescale, &self.runtime)) + .transpose() + else { return false; }; let body = lite::Datagram { @@ -3188,7 +3174,8 @@ impl Subscription { timestamp, payload: datagram.payload, }; - // has_datagrams is checked before this runs, so encoding never hits the version guard. + // has_datagrams is checked before this runs, and the timescale is set wherever the + // version needs a timestamp, so encoding never hits a version guard. let Ok(body) = body.encode_bytes(self.version) else { return false; }; diff --git a/rs/moq-net/src/lite/subscriber.rs b/rs/moq-net/src/lite/subscriber.rs index d08b62df13..e953f5333e 100644 --- a/rs/moq-net/src/lite/subscriber.rs +++ b/rs/moq-net/src/lite/subscriber.rs @@ -500,22 +500,26 @@ impl Subscriber { /// Decode one datagram body and hand it to the matching subscription's producer. fn route_datagram(&self, payload: bytes::Bytes) -> Result<(), Error> { - let dg = lite::Datagram::decode(payload, self.version)?; - // Write through the map rather than cloning the entry out: a `TrackEntry` clone // is a handful of atomic bumps on every datagram, and a producer held past its // removal would keep the track (its cached groups, its stats subscription) alive. // The group path already writes to a producer under this lock. let mut subscribes = self.subscribes.lock(); + // An unknown subscription's datagram is dropped below, however it parses. + let dg = lite::Datagram::decode(payload, self.version, |id| { + subscribes.get(&id).is_some_and(|entry| entry.timescale.is_some()) + })?; let Some(entry) = subscribes.get_mut(&dg.subscribe) else { // Unknown or already-closed subscription: drop the datagram. return Ok(()); }; - // Datagrams are lite-05+, which always negotiates a timescale; fall back to milliseconds defensively. - let scale = entry.timescale.unwrap_or(crate::Timescale::MILLI); - let timestamp = - Timestamp::new(dg.timestamp, scale).map_err(|_| Error::BoundsExceeded(crate::coding::BoundsExceeded))?; + let timestamp = dg + .timestamp + .zip(entry.timescale) + .map(|(value, scale)| Timestamp::new(value, scale)) + .transpose() + .map_err(|_| Error::BoundsExceeded(crate::coding::BoundsExceeded))?; // A datagram is never owed a stream, so it never holds the subscription's end open. if let Ok(mut tail) = entry.tail.write() { @@ -2170,7 +2174,7 @@ mod tests { lite::Datagram { subscribe: 7, sequence, - timestamp: sequence, + timestamp: Some(sequence), payload: bytes::Bytes::from_static(b"x"), } .encode_bytes(VERSION) @@ -5523,11 +5527,8 @@ impl kio::Task for FetchServeRun { // Make the group available (resolving the downstream fetch) and fill // it. The track::Info only takes effect if the track isn't accepted yet // (a fetch with no live subscription); otherwise the group inherits the - // accepted timescale. Relay-served FETCH is lite-05+, so `timescale` is - // `Some`; fall back to milliseconds defensively rather than - // panicking. - let group_info = - track::Info::default().with_timescale(self.timescale.unwrap_or(crate::Timescale::MILLI)); + // accepted timescale. + let group_info = track::Info::default().with_timescale(self.timescale); // The joined fetches pick the group up from the cache only when next // polled, so their demand outlives the request until then. let joined = request.result.clone(); diff --git a/rs/moq-net/src/lite/track.rs b/rs/moq-net/src/lite/track.rs index 01f6b5c909..bb8d5d4c70 100644 --- a/rs/moq-net/src/lite/track.rs +++ b/rs/moq-net/src/lite/track.rs @@ -64,9 +64,10 @@ pub struct TrackInfo { /// Publisher Max Age: an upper bound on how long the publisher caches a /// non-latest group past the arrival of a newer one. Encoded as milliseconds. pub max_age: Option, - /// Per-frame timestamp scale (units per second). Mandatory on Lite05+: every track - /// is timed, so this is always a real scale on the wire (never zero). - pub timescale: Timescale, + /// Per-frame timestamp scale (units per second), or `None` for an untimed track. + /// Lite07 encodes `None` as zero; Lite05/06 have no untimed encoding, so `None` + /// goes out as milliseconds (see [`Version::wire_timescale`]). + pub timescale: Option, } impl Message for TrackInfo { @@ -82,7 +83,11 @@ impl Message for TrackInfo { Version::Lite05 | Version::Lite06 => (encoded < LEGACY_UNLIMITED).then(|| Duration::from_millis(encoded)), _ => encoded.checked_sub(1).map(Duration::from_millis), }; - let timescale = Timescale::new(r.varint()?).map_err(|_| DecodeError::InvalidValue)?; + let timescale = match r.varint()? { + // Lite05/06 have no untimed encoding, so zero is malformed there. + 0 if !matches!(version, Version::Lite05 | Version::Lite06) => None, + scale => Some(Timescale::new(scale).map_err(|_| DecodeError::InvalidValue)?), + }; Ok(Self { priority, @@ -105,7 +110,7 @@ impl Message for TrackInfo { (_, Some(age)) => u64::try_from(age.as_millis() + 1).map_err(|_| EncodeError::BoundsExceeded)?, }; w.varint(encoded)?; - w.varint(u64::from(self.timescale))?; + w.varint(version.wire_timescale(self.timescale).map_or(0, u64::from))?; Ok(()) } } @@ -118,7 +123,7 @@ mod test { TrackInfo { priority: 7, max_age: Some(Duration::from_millis(2000)), - timescale: Timescale::MICRO, + timescale: Some(Timescale::MICRO), } } @@ -196,7 +201,7 @@ mod test { let got = info_roundtrip(Version::Lite05, &info_sample()); assert_eq!(got.priority, 7); assert_eq!(got.max_age, Some(Duration::from_millis(2000))); - assert_eq!(got.timescale, Timescale::MICRO); + assert_eq!(got.timescale, Some(Timescale::MICRO)); } #[test] @@ -205,8 +210,8 @@ mod test { let info = TrackInfo { priority: info.priority, max_age: info.max_age, - // The default track is untimed, which declares milliseconds like the publisher. - timescale: info.timescale.unwrap_or(Timescale::MILLI), + // The default track is untimed, which lite-05 declares as milliseconds. + timescale: info.timescale, }; let mut buf = Vec::new(); info.encode(&mut Encoder::new(&mut buf, Version::Lite05.into()), Version::Lite05) @@ -220,6 +225,44 @@ mod test { ); } + #[test] + fn lite07_encodes_an_untimed_track_as_zero() { + let info = TrackInfo { + timescale: None, + ..info_sample() + }; + let buf = info.encode_bytes(Version::Lite07).unwrap(); + assert_eq!(buf.last(), Some(&0)); + assert_eq!(info_roundtrip(Version::Lite07, &info).timescale, None); + } + + #[test] + fn lite06_declares_an_untimed_track_as_milliseconds() { + for version in [Version::Lite05, Version::Lite06] { + let info = TrackInfo { + timescale: None, + ..info_sample() + }; + assert_eq!(info_roundtrip(version, &info).timescale, Some(Timescale::MILLI)); + } + } + + #[test] + fn lite06_rejects_a_zero_timescale() { + for version in [Version::Lite05, Version::Lite06] { + let mut raw = Vec::new(); + let w = &mut Encoder::new(&mut raw, version.into()); + w.u8(0); + super::super::subscribe::pad_group_order(w, version).unwrap(); + w.varint(0u64).unwrap(); + w.varint(0u64).unwrap(); + assert!(matches!( + TrackInfo::decode_msg(&mut Decoder::new(&raw, version.into()), version), + Err(DecodeError::InvalidValue) + )); + } + } + #[test] fn track_info_errors_before_lite05() { let mut buf = Vec::new(); @@ -235,7 +278,7 @@ mod test { let info = TrackInfo { priority: 255, max_age: None, - timescale: Timescale::new((1u64 << 62) - 1).unwrap(), + timescale: Some(Timescale::new((1u64 << 62) - 1).unwrap()), }; let got = info_roundtrip(Version::Lite05, &info); assert_eq!(got.priority, 255); @@ -248,7 +291,7 @@ mod test { let info = TrackInfo { priority: 0, max_age: Some(Duration::from_nanos(999_999)), - timescale: Timescale::MILLI, + timescale: Some(Timescale::MILLI), }; let got = info_roundtrip(Version::Lite05, &info); assert_eq!(got.max_age, Some(Duration::ZERO)); @@ -260,7 +303,7 @@ mod test { let info = TrackInfo { priority: 7, max_age: Some(Duration::from_millis(u64::MAX)), - timescale: Timescale::MILLI, + timescale: Some(Timescale::MILLI), }; // The partial bytes are the Writer's to drop; see `a_failed_encode_leaves_no_partial_bytes`. assert!(matches!( diff --git a/rs/moq-net/src/lite/version.rs b/rs/moq-net/src/lite/version.rs index c83d6a7015..85aeb0d7ef 100644 --- a/rs/moq-net/src/lite/version.rs +++ b/rs/moq-net/src/lite/version.rs @@ -297,6 +297,19 @@ impl Version { _ => true, } } + + /// The timescale the wire carries for a track with `timescale`, or `None` when its + /// frames and datagrams carry no timestamp. + /// + /// Lite05 and Lite06 always carry one, so an untimed track declares milliseconds there + /// and its payloads carry their send times instead. + pub(crate) fn wire_timescale(self, timescale: Option) -> Option { + match self { + Self::Lite01 | Self::Lite02 | Self::Lite03 | Self::Lite04 => None, + Self::Lite05 | Self::Lite06 => Some(timescale.unwrap_or(crate::Timescale::MILLI)), + _ => timescale, + } + } } impl fmt::Display for Version { diff --git a/rs/moq-net/src/model/frame.rs b/rs/moq-net/src/model/frame.rs index b0acdcf928..e125f52cee 100644 --- a/rs/moq-net/src/model/frame.rs +++ b/rs/moq-net/src/model/frame.rs @@ -31,8 +31,8 @@ pub struct Info { /// [`group::Producer::create_frame`] converts it into the parent track's /// timescale, so the scale you build it with doesn't have to match the track. /// No receiver fills one in, so an untimed track stays untimed across hops, except - /// over moq-lite 05 and later, which can't mark a track untimed yet and carries the - /// encoder's send time instead. + /// over moq-lite 05 and 06, which can't mark a track untimed and carry the encoder's + /// send time instead. pub timestamp: Option, } diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index 2e8e834c28..3429e5333b 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -5385,20 +5385,20 @@ mod test { let body = lite::Datagram { subscribe: 5, sequence: d.sequence, - timestamp: d.timestamp.unwrap().value(), + timestamp: Some(d.timestamp.unwrap().value()), payload: d.payload.clone(), } .encode_bytes(version) .unwrap(); // Subscriber decodes the body and writes it downstream, preserving the sequence. - let wire = lite::Datagram::decode(body, version).unwrap(); + let wire = lite::Datagram::decode(body, version, |_| true).unwrap(); let mut downstream = track_producer("test", Info::timed()); let mut downstream_dg = downstream.subscribe(None); downstream .insert_datagram( wire.sequence, - Timestamp::new(wire.timestamp, Timescale::MILLI).unwrap(), + Timestamp::new(wire.timestamp.unwrap(), Timescale::MILLI).unwrap(), wire.payload, ) .unwrap(); diff --git a/rs/moq-net/src/test_interop.rs b/rs/moq-net/src/test_interop.rs index a3b9dbe47e..172bde40d0 100644 --- a/rs/moq-net/src/test_interop.rs +++ b/rs/moq-net/src/test_interop.rs @@ -80,9 +80,9 @@ fn lite_values() -> Vec { values } -/// One GROUP header and its frames, as a publisher writes them: a zigzag timestamp delta, -/// the payload size, then the payload. -fn lite_group(version: lite::Version) -> Vec { +/// One GROUP header and its frames, as a publisher writes them: a zigzag timestamp delta on +/// a timed track, the payload size, then the payload. +fn lite_group(version: lite::Version, frames: impl IntoIterator, usize)>) -> Vec { let mut buf = Vec::new(); let w = &mut Encoder::new(&mut buf, version.into()); let header = lite::Group { @@ -91,8 +91,10 @@ fn lite_group(version: lite::Version) -> Vec { frame_start: 200, }; header.encode(w, version).unwrap(); - for (delta, size) in [(0i64, 10usize), (33_333, 300), (-1_000, 20_000), (1 << 40, 1)] { - w.varint(varint::zigzag(delta)).unwrap(); + for (delta, size) in frames { + if let Some(delta) = delta { + w.varint(varint::zigzag(delta)).unwrap(); + } w.varint(size as u64).unwrap(); w.slice(&vec![0xAB; size]); } @@ -100,12 +102,16 @@ fn lite_group(version: lite::Version) -> Vec { } #[derive(serde::Deserialize)] +#[serde(rename_all = "camelCase")] struct Echo { values: Vec, varints: Vec, setup: Vec, datagram: Vec, + untimed_datagram: Vec, + track_info: Vec, group: Vec, + untimed_group: Vec, beyond: Vec, } @@ -135,13 +141,54 @@ fn lite_varint_interop() { let datagram = lite::Datagram { subscribe: (1 << 40) + 1, sequence: 300, - timestamp: (1 << 50) + 5, + timestamp: Some((1 << 50) + 5), payload: bytes::Bytes::from_static(b"x"), } .encode_bytes(version) .unwrap(); - let group = lite_group(version); + // Only lite-07 can carry an untimed track; lite-06 stamps it, so it has no untimed bytes. + let untimed = version.wire_timescale(None).is_none(); + + // The body doesn't say it is untimed; the subscription's track does. + let untimed_datagram = lite::Datagram { + subscribe: 9, + sequence: 1 << 30, + timestamp: None, + payload: bytes::Bytes::from_static(b"untimed"), + } + .encode_bytes(version); + let untimed_datagram = match untimed { + true => untimed_datagram.unwrap().to_vec(), + false => { + assert!(untimed_datagram.is_err(), "{version}: encoded an untimed datagram"); + Vec::new() + } + }; + + // An untimed track: zero on lite-07, milliseconds before it. + let track_info = lite::TrackInfo { + priority: 1, + max_age: None, + timescale: None, + } + .encode_bytes(version) + .unwrap(); + + let group = lite_group( + version, + [ + (Some(0), 10), + (Some(33_333), 300), + (Some(-1_000), 20_000), + (Some(1 << 40), 1), + ], + ); + // Small, since the whole input rides in one argument. + let untimed_group = match untimed { + true => lite_group(version, [(None, 10), (None, 300)]), + false => Vec::new(), + }; let input = serde_json::json!({ "version": crate::Version::from(version).alpn(), @@ -149,7 +196,10 @@ fn lite_varint_interop() { "varints": varints, "setup": setup.to_vec(), "datagram": datagram.to_vec(), + "untimedDatagram": untimed_datagram, + "trackInfo": track_info.to_vec(), "group": group, + "untimedGroup": untimed_group, }); let echo: Echo = serde_json::from_slice(&bun("lite-varint.ts", input)).expect("JS returned its encodings"); @@ -158,7 +208,10 @@ fn lite_varint_interop() { assert_eq!(echo.varints, varints, "{version}: JS encoded the values differently"); assert_eq!(echo.setup, setup, "{version}: SETUP"); assert_eq!(echo.datagram, datagram, "{version}: datagram"); + assert_eq!(echo.untimed_datagram, untimed_datagram, "{version}: untimed datagram"); + assert_eq!(echo.track_info, track_info, "{version}: TRACK_INFO"); assert_eq!(echo.group, group, "{version}: group"); + assert_eq!(echo.untimed_group, untimed_group, "{version}: untimed group"); match version { lite::Version::Lite07 => { diff --git a/rs/moq-net/tests/untimed.rs b/rs/moq-net/tests/untimed.rs index bbcbec5317..e30cd0b01d 100644 --- a/rs/moq-net/tests/untimed.rs +++ b/rs/moq-net/tests/untimed.rs @@ -1,9 +1,9 @@ //! An untimed track reaches every subscriber untimed, through a relay too, on every wire //! that can say so, and a timed track keeps its timestamps. //! -//! No receiver fills in arrival time. Until lite-07 encodes absence, a lite-05+ encoder -//! writes its send time instead, so those hops are the one place an untimed track gains a -//! timeline. +//! No receiver fills in arrival time. Lite-05 and lite-06 cannot encode absence, so their +//! encoder writes its send time instead, and those hops are the one place an untimed track +//! gains a timeline. mod support; @@ -25,7 +25,7 @@ const TIMEOUT: Duration = Duration::from_secs(10); enum Arrives { /// Both keep what the publisher wrote. Faithful, - /// The untimed track carries the encoder's send times (lite-05+ before lite-07). + /// The untimed track carries the encoder's send times (lite-05 and lite-06). SendTime, /// The wire declares no timescale, so even the timed track arrives untimed. Untimed, @@ -34,7 +34,7 @@ enum Arrives { fn expected(version: &str) -> Arrives { match version { "moq-lite-03" | "moq-lite-04" => Arrives::Untimed, - "moq-lite-05" | "moq-lite-06" | "moq-lite-07-wip" => Arrives::SendTime, + "moq-lite-05" | "moq-lite-06" => Arrives::SendTime, // Drafts 14 to 16 can't carry TIMESCALE. "moq-transport-14" | "moq-transport-16" => Arrives::Untimed, _ => Arrives::Faithful, @@ -151,9 +151,14 @@ async fn frames_arrive_as_published() { /// does not gain one downstream. #[moq_net_sim::test] async fn a_relay_forwards_an_untimed_track_untimed() { - let (frame, info) = frame_through(&["moq-transport-17", "moq-transport-20"], untimed()).await; - assert_eq!(frame, None, "the relay did not stamp it"); - assert_eq!(info.timescale, None, "the relay must not claim a timeline"); + for hops in [ + &["moq-transport-17", "moq-transport-20"][..], + &["moq-lite-07-wip", "moq-lite-07-wip"][..], + ] { + let (frame, info) = frame_through(hops, untimed()).await; + assert_eq!(frame, None, "{hops:?}: the relay did not stamp it"); + assert_eq!(info.timescale, None, "{hops:?}: the relay must not claim a timeline"); + } for hops in [ &["moq-lite-03", "moq-transport-17"][..], @@ -169,7 +174,7 @@ async fn a_relay_forwards_an_untimed_track_untimed() { /// moq-transport: only an unfloored subscriber jumps to the latest group. /// /// The subscription asks for history the way a resume does, with a budget covering it: -/// lite-05+ carries send times, so downstream the groups are timed and a zero budget would +/// lite-05 and lite-06 carry send times, so downstream the groups are timed and a zero budget would /// skip the older ones there, as it would on any timed track. moq-transport before draft /// 20 can only join a past group through a joining FETCH, which our publisher refuses for /// groups before the live one, so draft 20 stands in for it. @@ -206,8 +211,8 @@ async fn an_explicit_start_holds_on_an_untimed_track() { } } -/// An untimed track's datagram arrives untimed on moq-transport, and with a send time on -/// lite. +/// An untimed track's datagram arrives untimed, except with a send time on lite-05 and +/// lite-06. #[moq_net_sim::test] async fn datagrams_arrive_as_published() { for version in ["moq-lite-05", "moq-lite-07-wip", "moq-transport-16", "moq-transport-20"] { @@ -228,15 +233,15 @@ async fn datagrams_arrive_as_published() { .expect("datagram timeout") .unwrap() .expect("a datagram"); - let lite = version.starts_with("moq-lite"); - assert_eq!(datagram.timestamp.is_some(), lite, "{version}"); + let stamps = version == "moq-lite-05"; + assert_eq!(datagram.timestamp.is_some(), stamps, "{version}"); } } /// A standalone FETCH of an untimed track answers with untimed objects. #[moq_net_sim::test] async fn a_fetched_untimed_frame_arrives_untimed() { - for version in ["moq-lite-06", "moq-transport-19"] { + for version in ["moq-lite-06", "moq-lite-07-wip", "moq-transport-19"] { let chain = Chain::new(&[version]).await; let broadcast = chain.nodes[0].create_broadcast("room").unwrap(); let mut dynamic = broadcast.dynamic(); @@ -261,7 +266,7 @@ async fn a_fetched_untimed_frame_arrives_untimed() { let mut fetched = waiting.await.unwrap(); let frame = fetched.read_frame().await.unwrap().unwrap(); - let lite = version.starts_with("moq-lite"); - assert_eq!(frame.timestamp.is_some(), lite, "{version}"); + let stamps = version == "moq-lite-06"; + assert_eq!(frame.timestamp.is_some(), stamps, "{version}"); } } diff --git a/test/interop/lite-varint.ts b/test/interop/lite-varint.ts index df3797fbf0..58faec72a4 100644 --- a/test/interop/lite-varint.ts +++ b/test/interop/lite-varint.ts @@ -3,6 +3,7 @@ import assert from "node:assert/strict"; import { Datagram } from "../../js/net/src/lite/datagram.ts"; import { frameDecoder, Group } from "../../js/net/src/lite/group.ts"; import { Setup } from "../../js/net/src/lite/setup.ts"; +import { TrackInfo } from "../../js/net/src/lite/track.ts"; import { Version } from "../../js/net/src/lite/version.ts"; import { Reader, Writer } from "../../js/net/src/stream.ts"; @@ -12,7 +13,10 @@ const input: { varints: number[]; setup: number[]; datagram: number[]; + untimedDatagram: number[]; + trackInfo: number[]; group: number[]; + untimedGroup: number[]; } = JSON.parse(process.argv[2]); const versions: Record = { @@ -60,9 +64,25 @@ if (version === Version.DRAFT_07) { const setup = await Setup.decode(reader(input.setup), version); const setupBytes = await write((w) => setup.encode(w, version)); -const datagram = await Datagram.decode(Uint8Array.from(input.datagram), version); +const datagram = await Datagram.decode(Uint8Array.from(input.datagram), version, () => true); const datagramBytes = [...datagram.encode(version)]; +// Only lite-07 can carry an untimed track; lite-06 stamps it, so it has no untimed bytes. +const untimed = version === Version.DRAFT_07; + +let untimedDatagramBytes: number[] = []; +if (untimed) { + const datagram = await Datagram.decode(Uint8Array.from(input.untimedDatagram), version, () => false); + assert.equal(datagram.timestamp, undefined); + untimedDatagramBytes = [...datagram.encode(version)]; +} else { + assert.throws(() => new Datagram(9n, 1, undefined, new Uint8Array()).encode(version)); +} + +const trackInfo = await TrackInfo.decode(reader(input.trackInfo), version); +assert.equal(trackInfo.timescale, version === Version.DRAFT_07 ? undefined : 1_000); +const trackInfoBytes = await write((w) => trackInfo.encode(w, version)); + // A GROUP header, then frames until the stream ends. const g = reader(input.group); const group = await Group.decode(g, version); @@ -86,13 +106,37 @@ const groupBytes = await write(async (w) => { } }); +// The same frames on an untimed track: no timestamp delta. +let untimedGroupBytes: number[] = []; +if (untimed) { + const u = reader(input.untimedGroup); + const untimedGroup = await Group.decode(u, version); + const untimedFrames: Uint8Array[] = []; + for (;;) { + const frame = await u.decodeMaybe(frameDecoder(0)); + if (!frame) break; + assert.equal(frame.timestamp, undefined); + untimedFrames.push(frame.payload); + } + untimedGroupBytes = await write(async (w) => { + await untimedGroup.encode(w, version); + for (const payload of untimedFrames) { + await w.u53(payload.byteLength); + await w.write(payload); + } + }); +} + console.log( JSON.stringify({ values: values.map((v) => v.toString()), varints, setup: setupBytes, datagram: datagramBytes, + untimedDatagram: untimedDatagramBytes, + trackInfo: trackInfoBytes, group: groupBytes, + untimedGroup: untimedGroupBytes, beyond, }), );