diff --git a/cpp/moq/test/probe.cpp b/cpp/moq/test/probe.cpp index d2c07c81c6d..fe3fd8aa1df 100644 --- a/cpp/moq/test/probe.cpp +++ b/cpp/moq/test/probe.cpp @@ -194,7 +194,8 @@ int main() { CHECK(!reading.valid()); CHECK(frame.has_value()); CHECK(frame->payload == bytes("hello")); - CHECK(frame->timestamp_us == 1000); + // The default moq-lite-06 session drops received timestamps, so the track arrives untimed. + CHECK(!frame->timestamp_us.has_value()); // A continuation receives the next frame on the executor. { diff --git a/doc/concept/moq-lite.md b/doc/concept/moq-lite.md index baa7088ea0b..cbe0349ea5a 100644 --- a/doc/concept/moq-lite.md +++ b/doc/concept/moq-lite.md @@ -283,6 +283,14 @@ 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 publisher refuses a group that starts before a lower one, and so does a +moq-lite 07 receiver. Since moq-lite 05 and 06 send times look like media time, +a receiver drops every timestamp it gets over them and the track arrives +untimed. +Timestamps received over moq-transport, whose peers may stamp arrival times, +are not checked and are only passed on over moq-transport: a relay serves that +track to moq-lite untimed. + A publisher may compress a track, which pays off for repetitive data such as JSON and not for media. Each group's frames become one DEFLATE stream ([moq-flate](/draft/moq-flate)), and a datagram diff --git a/doc/setup/upgrade.md b/doc/setup/upgrade.md index c49ef5c4de6..2cd45e031d6 100644 --- a/doc/setup/upgrade.md +++ b/doc/setup/upgrade.md @@ -65,8 +65,10 @@ These land with the next breaking release. - **A track is timed or untimed, and nothing fills in a timestamp on receive.** Tracks from moq-lite before 05, and moq-transport tracks without `TIMESCALE` (every track on drafts 14 to 16), arrive untimed instead of - stamped with their arrival time. moq-lite 05 and later can't encode absence - yet and carry a send time instead; see + stamped with their arrival time. moq-lite 05 and 06 can't encode absence and + carry a send time instead, so their receivers drop every timestamp and those + tracks arrive untimed too: read media time from the payload. moq-transport + timestamps are served to moq-lite untimed; see [untimed tracks](/concept/moq-lite#subscriptions). A new subscriber with no start on an untimed track starts at the latest group. - **Max age and max delay count wall-clock time too.** A superseded group is diff --git a/drafts/draft-lcurley-moq-lite.md b/drafts/draft-lcurley-moq-lite.md index 7752bbb30aa..47b43b038a9 100644 --- a/drafts/draft-lcurley-moq-lite.md +++ b/drafts/draft-lcurley-moq-lite.md @@ -184,7 +184,9 @@ Each frame of a timed Track carries a presentation timestamp expressed in the Tr A Track without a `Timescale` is untimed: its frames and datagrams carry no timestamp. A publisher MUST NOT start a timed Group's frame 0 earlier than that of any lower-sequence Group it publishes. lite-05 and lite-06 cannot encode an untimed Track, so a publisher on those versions declares a `Timescale` of 1000 for it and stamps every frame and datagram with its send time in milliseconds. -A receiver cannot tell such send times from presentation timestamps, and they need not rise with sequence, so it does not check received Groups against this rule. +A receiver cannot tell such send times from presentation timestamps, and they need not rise with sequence, so it discards every timestamp received on those versions and treats the Track as untimed. +On later versions a receiver MAY check a received Group against this rule, refusing a violating Group's stream (STOP_SENDING) with TIMESTAMP_MISMATCH. +A relay that cannot vouch for a Track's timestamps, such as ones received over another protocol, MUST serve the Track untimed. ## Positions {#positions} A Position is a (Group Sequence, Frame Index) pair identifying one frame within a Track. @@ -365,7 +367,7 @@ Sent when resetting a stream (RESET_STREAM), or when refusing to receive one (ST | ------- | ------------- | ----------- | | 0x38 | FRAME_TOO_LARGE | A frame declared a payload larger than the receiver accepts. | | ------- | ------------- | ----------- | -| 0x39 | TIMESTAMP_MISMATCH | A frame's timestamp does not match its track's timescale. | +| 0x39 | TIMESTAMP_MISMATCH | A frame's timestamp does not fit its track: it is present on an untimed track, absent on a timed one, or starts a Group before a lower-sequence one. | | ------- | ------------- | ----------- | | 0x3A | NOT_FETCHABLE | The FETCH named a group delivered only as a datagram, which is never cached (see [Datagrams](#datagrams)). It has no moq-transport value; a bridge refuses the FETCH with DOES_NOT_EXIST. | | ------- | ------------- | ----------- | @@ -1761,7 +1763,7 @@ The `Message Length` describes the payload size on the wire. - Replaced ANNOUNCE_OK Active Count with an ordered, empty SYNCED event (0x4), marking the initial set without retaining routes that end before delivery. -- A publisher must not start a timed Group's frame 0 earlier than a lower-sequence Group's. Receivers do not check it, since a received timestamp may be a send time. +- A publisher must not start a timed Group's frame 0 earlier than a lower-sequence Group's, and a receiver may refuse a Group that does. A receiver discards the timestamps it receives over lite-05/06, which may be send times, and a relay serves a Track whose timestamps it cannot vouch for untimed. - Added the reliable ordered RTCDataChannel binding, with one Qmux Record per binary message and no datagrams. - Added `Group Order` to SUBSCRIBE and SUBSCRIBE_UPDATE: descending (0x0, the previous fixed order) or ascending (0x1). A relay requests descending upstream unless every subscription it serves is ascending. diff --git a/js/e2ee/src/transport.test.ts b/js/e2ee/src/transport.test.ts index acd755d131d..84588ae7242 100644 --- a/js/e2ee/src/transport.test.ts +++ b/js/e2ee/src/transport.test.ts @@ -48,10 +48,10 @@ async function active(origin: Origin.Producer, path: Path.Valid): Promise BigInt(Number.MAX_SAFE_INTEGER)) { throw new RangeError("largest group exceeds safe integers"); diff --git a/js/net/src/integration.test.ts b/js/net/src/integration.test.ts index 3da09eddb31..72a0af7d523 100644 --- a/js/net/src/integration.test.ts +++ b/js/net/src/integration.test.ts @@ -984,10 +984,10 @@ test("integration: a peer is served the cheaper route it was offered, not the lo origin.close(); }); -test("integration: lite draft-05 fetches a cached group", async () => { +test("integration: lite draft-07 fetches a cached group", async () => { const enc = new TextEncoder(); const dec = new TextDecoder(); - const pair = createMockTransportPair(Lite.ALPN_05); + const pair = createMockTransportPair(Lite.ALPN_07_WIP); const origin = new OriginProducer(); const [client, server] = await Promise.all([ @@ -1160,10 +1160,10 @@ test("integration: lite draft-05 coalesces concurrent fetches of one group", asy server.abort(); }); -test("integration: lite draft-05 fetches an in-progress group", async () => { +test("integration: lite draft-07 fetches an in-progress group", async () => { const enc = new TextEncoder(); const dec = new TextDecoder(); - const pair = createMockTransportPair(Lite.ALPN_05); + const pair = createMockTransportPair(Lite.ALPN_07_WIP); const origin = new OriginProducer(); const [client, server] = await Promise.all([ @@ -1202,8 +1202,8 @@ test("integration: lite draft-05 fetches an in-progress group", async () => { // The publisher caches TRACK_INFO so it only asks the application once per track. The cache // has to expire with the broadcast that answered it: a republish puts a different producer // on the path, and its immutable properties are its own. -test("integration: lite draft-05 track info follows a republished broadcast", async () => { - const pair = createMockTransportPair(Lite.ALPN_05); +test("integration: lite draft-07 track info follows a republished broadcast", async () => { + const pair = createMockTransportPair(Lite.ALPN_07_WIP); const origin = new OriginProducer(); const [client, server] = await Promise.all([ @@ -1240,9 +1240,9 @@ test("integration: lite draft-05 track info follows a republished broadcast", as // FETCH reads the same cache to decide the timescale it serves frames in, so a stale entry // quantizes the successor's timestamps to the predecessor's resolution. -test("integration: lite draft-05 fetch uses the republished track's timescale", async () => { +test("integration: lite draft-07 fetch uses the republished track's timescale", async () => { const enc = new TextEncoder(); - const pair = createMockTransportPair(Lite.ALPN_05); + const pair = createMockTransportPair(Lite.ALPN_07_WIP); const origin = new OriginProducer(); const [client, server] = await Promise.all([ diff --git a/js/net/src/internal.ts b/js/net/src/internal.ts index aaf94008bbb..2ed28d2d81f 100644 --- a/js/net/src/internal.ts +++ b/js/net/src/internal.ts @@ -417,12 +417,6 @@ export const hooks: { * as the Rust `request_start`. */ startTrack: (producer: Producer, served: { start?: number; requested?: number }) => void; - // pending: /quest/m1/public-model/ingest.md - /** - * Mark a track as a session's copy of a remote one: its groups are never checked for a - * rewind, since a received timestamp may not be media time. - */ - receiveTrack: (producer: Producer) => void; /** Mint a track {@link Request}; assigned by `track.ts`. */ makeRequest: (options: TrackRequestOptions) => Request; // pending: /quest/m1/public-model/ingest.md @@ -571,9 +565,6 @@ export const hooks: { /** Name the epoch of the route an origin resolved a broadcast handle through. */ stampEpoch: (target: BroadcastConsumer, epoch: Epoch.Valid | undefined) => void; } = { - receiveTrack: () => { - throw new Error("track.ts not loaded"); - }, endTrack: () => { throw new Error("track.ts not loaded"); }, diff --git a/js/net/src/lite/group.ts b/js/net/src/lite/group.ts index b7e57567db1..90b4bf4ecbe 100644 --- a/js/net/src/lite/group.ts +++ b/js/net/src/lite/group.ts @@ -76,7 +76,7 @@ function unzigzag(v: bigint): bigint { * FRAME format), decoded into a Timestamp at that scale. Scale 0 (pre-lite-05) carries no * timestamp, so frames arrive untimed. */ -export function frameDecoder(scale: number): (c: Cursor) => netGroup.Frame { +export function frameDecoder(scale: number, timed = true): (c: Cursor) => netGroup.Frame { if (scale === 0) { return (c) => ({ payload: c.read(c.u53()) }); } @@ -88,13 +88,23 @@ export function frameDecoder(scale: number): (c: Cursor) => netGroup.Frame { const payload = c.read(c.u53()); // After the last read, so a decode that ran short and gets retried adds the delta once. prevTs += delta; + // Read to stay in step, but dropped where it may be a send time. + if (!timed) return { payload }; return { payload, timestamp: new Time.Timestamp(Number(prevTs), timescale) }; }; } -/** Write a group stream's frames into `producer` until the stream ends or the producer closes. */ -export async function readFrames(stream: Reader, producer: netGroup.Producer, scale: number): Promise { - const decode = frameDecoder(scale); +/** + * Write a group stream's frames into `producer` until the stream ends or the producer closes. + * Each frame's timestamp is read at `scale` and kept only when `timed`. + */ +export async function readFrames( + stream: Reader, + producer: netGroup.Producer, + scale: number, + timed = true, +): Promise { + const decode = frameDecoder(scale, timed); for (;;) { // Every frame already buffered is written without an await, so the reader wakes once diff --git a/js/net/src/lite/publisher.test.ts b/js/net/src/lite/publisher.test.ts index 881f844fc57..7cfcf76eb51 100644 --- a/js/net/src/lite/publisher.test.ts +++ b/js/net/src/lite/publisher.test.ts @@ -534,6 +534,41 @@ test("lite draft-05: the fetch response ranks the publisher's own writes", async } }); +// An unchecked track goes out untimed, so lite-06 carries its send time, never the frame's own +// timestamp, which may be a moq-transport arrival time. +test("lite draft-06: an unchecked track's fetched frames carry their send time", async () => { + const pair = createMockTransportPair(ALPN_06); + const origin = new OriginProducer(); + const publisher = new Publisher(pair.server, Version.DRAFT_06, randomHop(), origin.consume()); + + const broadcast = publish(origin, Path.from("test")); + const track = broadcast.createTrack("video", { timescale: Timescale.MICRO, unchecked: true }); + + const group = new GroupProducer(0); + group.writeFrame(textFrame("hello", Timestamp.fromMillis(36_000_000))); + group.close(); + track.writeGroup(group); + + const client = await Stream.open(pair.client, { version: Version.DRAFT_06 }); + const server = await Stream.accept(pair.server, Version.DRAFT_06); + if (!server) throw new Error("publisher never saw the fetch stream"); + + const msg = new Fetch({ broadcast: Path.from("test"), track: "video", priority: 0, group: 0 }); + try { + const before = Math.floor(Timestamp.now().value); + await publisher.runFetch(msg, server); + const after = Math.ceil(Timestamp.now().value); + + // The first frame's zigzag delta is against 0, in the milliseconds TRACK_INFO declares. + const sent = Number((await client.reader.u62()) >> 1n); + expect(sent).toBeGreaterThanOrEqual(before); + expect(sent).toBeLessThanOrEqual(after); + } finally { + publisher.close(); + client.close(); + } +}); + // How long a served-subscription test waits for the next group stream before calling the // publisher idle. Nothing waits this out: it only bounds the read when a group never comes. const IDLE_MS = 500; diff --git a/js/net/src/lite/publisher.ts b/js/net/src/lite/publisher.ts index 1fc5254c016..de68d6b03e4 100644 --- a/js/net/src/lite/publisher.ts +++ b/js/net/src/lite/publisher.ts @@ -17,6 +17,7 @@ import { Priority, sendOrder } from "../priority.ts"; import { type Reader, type Stream, Writer } from "../stream.ts"; import { type Timescale, Timestamp } from "../time.ts"; import type * as track from "../track.ts"; +import { liteTimescale } from "../track.ts"; import { untilAborted } from "../util/abort.ts"; import { Interests, LIMITS, type Limits, Slots } from "../util/limits.ts"; import { nextTask } from "../util/yield.ts"; @@ -64,12 +65,26 @@ function zigzag(delta: bigint): bigint { return delta >= 0n ? delta << 1n : (-delta << 1n) - 1n; } -/** - * 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)); +/** How a lite hop stamps a track's frames and datagrams. */ +interface Stamp { + /** The wire timescale TRACK_INFO declared (see `wireTimescale`). */ + timescale: Timescale; + /** + * Whether each frame's own timestamp goes out. Lite-05/06 cannot encode an untimed track, so + * one served untimed carries its send time instead, unchecked ones included. + */ + timed: boolean; +} + +/** How `version` stamps a track served at `timescale` (see `liteTimescale`), or undefined when it carries none. */ +function stamping(version: Version, timescale: Timescale | undefined): Stamp | undefined { + const wire = wireTimescale(version, timescale); + return wire === undefined ? undefined : { timescale: wire, timed: timescale !== undefined }; +} + +/** A frame or datagram's `timestamp` as its raw wire value, or its send time when the track goes out untimed. */ +function wireTime(timestamp: Timestamp | undefined, stamp: Stamp): number { + return Math.round(((stamp.timed ? timestamp : undefined) ?? Timestamp.now()).as(stamp.timescale)); } /** @@ -99,8 +114,8 @@ interface RunGroup { /** The group to serve. */ group: group.Consumer; - /** The track's wire timescale, applied to every frame timestamp, or undefined to send none. */ - timescale: Timescale | undefined; + /** How every frame is stamped, or undefined to send no timestamps. */ + stamp: Stamp | undefined; /** The subscription's ranking, which this stream joins for as long as it runs. */ priority: Priority; @@ -155,8 +170,8 @@ function frameRange(bounds: FrameBounds, sequence: number): { start: number; end type ServeGroup = { /** The Subscribe ID the GROUP message references. */ sub: bigint; - /** The track's wire timescale, which every frame timestamp is converted to, or undefined to send none. */ - timescale: Timescale | undefined; + /** How every frame is stamped, or undefined to send no timestamps. */ + stamp: Stamp | 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. */ @@ -854,7 +869,7 @@ export class Publisher { try { // Older drafts carry no timestamps. - let timescale: Timescale | undefined; + let stamp: Stamp | undefined; if (hasTrackStream(this.version)) { // Lite-05+ accepts implicitly: no SUBSCRIBE_OK (the immutable @@ -867,7 +882,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(this.version, info.timescale); + stamp = stamping(this.version, liteTimescale(info)); } else { // Older drafts acknowledge with SUBSCRIBE_OK and stream frames verbatim. const ok = new SubscribeOk({ @@ -884,7 +899,7 @@ export class Publisher { // Serve datagrams concurrently with groups whenever the transport carries them // (the writer exists iff so). No group fallback: otherwise they simply aren't sent. if (this.#datagramWriter) { - datagrams = this.#runDatagrams(msg.id, track, timescale, served); + datagrams = this.#runDatagrams(msg.id, track, stamp, served); } controls = new SubscriptionControls({ @@ -913,7 +928,7 @@ export class Publisher { const finished = await this.#runTrack(track, stream.writer, controls, { sub: msg.id, broadcast: msg.broadcast, - timescale, + stamp, served, bounds: { startGroup: msg.startGroup, @@ -1042,7 +1057,7 @@ export class Publisher { return; } await this.#runFetchGroup(group, stream.writer, { - timescale: wireTimescale(this.version, info.timescale), + stamp: stamping(this.version, info.timescale), start: msg.startFrame, end: msg.endFrame, }); @@ -1081,12 +1096,12 @@ export class Publisher { serving: { sub: bigint; broadcast: Path.Valid; - timescale: Timescale | undefined; + stamp: Stamp | undefined; served: Run[] | undefined; bounds: FrameBounds; }, ): Promise { - const { sub, broadcast, timescale, served, bounds } = serving; + const { sub, broadcast, stamp, served, bounds } = serving; // Lite-05+ resolves the range on the subscribe stream: SUBSCRIBE_START once the // first group is known, SUBSCRIBE_END when the track finishes. const emitRange = hasTrackStream(this.version); @@ -1357,7 +1372,7 @@ export class Publisher { const options: RunGroup = { sub, group, - timescale, + stamp, priority, unsubscribed, start: range.start, @@ -1473,8 +1488,9 @@ export class Publisher { // Publisher Max Age: the publisher's retention bound, advertised so // relays re-serve with the same window. maxAge: info.maxAge, - // `#serveGroup` emits each frame converted to it. - timescale: info.timescale ?? undefined, + // `#serveGroup` emits each frame converted to it. Unchecked timestamps never + // enter moq-lite, whose receivers check them. + timescale: liteTimescale(info), }); })(); @@ -1514,12 +1530,7 @@ export class Publisher { * * @internal */ - async #runDatagrams( - sub: bigint, - track: track.Subscriber, - timescale: Timescale | undefined, - served: Run[] | undefined, - ) { + async #runDatagrams(sub: bigint, track: track.Subscriber, stamp: Stamp | undefined, served: Run[] | undefined) { const writer = this.#datagramWriter; if (!writer) return; // Only reached with a writer (see the #datagramWriter gate). const maxSize = DatagramStream.maxDatagramSize(this.#quic); @@ -1530,7 +1541,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 = timescale === undefined ? undefined : wireTime(datagram.timestamp, timescale); + const ts = stamp === undefined ? undefined : wireTime(datagram.timestamp, stamp); 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. @@ -1555,7 +1566,7 @@ export class Publisher { async #runFetchGroup( group: group.Consumer, stream: Writer, - { timescale, start: startFrame, end: endFrame }: ServeFetch, + { stamp, start: startFrame, end: endFrame }: ServeFetch, ) { // The response carries no header, so the receiver numbers the first frame it gets // as `startFrame`. Skipping the head here is the only thing keeping those numbers @@ -1571,8 +1582,8 @@ export class Publisher { const frame = await race([group.readFrame(), stream.closed]); if (!frame) break; - if (timescale !== undefined) { - const ts = BigInt(wireTime(frame.timestamp, timescale)); + if (stamp !== undefined) { + const ts = BigInt(wireTime(frame.timestamp, stamp)); await stream.u62(zigzag(ts - prevTs)); prevTs = ts; } @@ -1637,7 +1648,7 @@ export class Publisher { * @internal */ async #serveGroup(stream: Writer, options: RunGroup) { - const { sub, group, timescale, priority, start: startFrame, end: endFrame } = options; + const { sub, group, stamp, priority, start: startFrame, end: endFrame } = options; // This model holds whole groups, so frame `startFrame` is always reachable unless // the group ends first. Declaring it up front keeps the stream self-describing. const msg = new GroupMessage({ subscribe: sub, sequence: group.sequence, frameStart: startFrame }); @@ -1677,9 +1688,9 @@ export class Publisher { if (read.sequence < startFrame) continue; if (endFrame !== undefined && read.sequence > endFrame) break; - if (timescale !== undefined) { + if (stamp !== undefined) { // Convert each frame to the track's wire timescale. - const ts = BigInt(wireTime(read.frame.timestamp, timescale)); + const ts = BigInt(wireTime(read.frame.timestamp, stamp)); await hooks.guardGroup(group, () => stream.u62(zigzag(ts - prevTs))); prevTs = ts; } diff --git a/js/net/src/lite/subscriber.test.ts b/js/net/src/lite/subscriber.test.ts index c98388f23ea..fc8efae24e2 100644 --- a/js/net/src/lite/subscriber.test.ts +++ b/js/net/src/lite/subscriber.test.ts @@ -4,7 +4,16 @@ import { Signal } from "@moq/signals"; import type { Grant } from "../auth.ts"; import type { Probe as ProbeStats } from "../connection/stats.ts"; import * as Epoch from "../epoch.ts"; -import { error, fromTransport, ProtocolViolation, reason, SessionCode, StreamCode, StreamError } from "../error.ts"; +import { + error, + fromTransport, + ProtocolViolation, + reason, + SessionCode, + StreamCode, + StreamError, + TimestampMismatch, +} from "../error.ts"; import { HopSchema, isAnonymous, MAX_HOPS, Route, UNKNOWN_HOP } from "../hop.ts"; import * as Path from "../path.ts"; import { Reader, Writer } from "../stream.ts"; @@ -1718,44 +1727,67 @@ async function groupStream(timestamp: number, version: Version = Version.DRAFT_0 ); } -// A received timestamp may not be media time (lite-05/06 stamp an untimed track's frames with -// their send time), so a received group that rewinds or lands out of order is accepted at every -// version. Only a local publisher's writes refuse a rewind. -test.each([Version.DRAFT_05, Version.DRAFT_06, Version.DRAFT_07])( - "version %p accepts a received rewind", - async (version) => { - const { quic, streams, opened } = fakeSession(); - const subscriber = new Subscriber(quic, version, HopSchema.parse(1n)); - const broadcast = subscriber.consume(Path.from("room")); - // Old enough to keep every group, however far behind the newest start. - const track = broadcast.track("video").subscribe({ maxDelay: Milli(60_000) }); - try { - await opened(0); - await answerTrackInfo(streams[0], version, 1000); - await (await opened(1)).reading; - await subscriber.runGroup(new Group({ subscribe: 0n, sequence: 4 }), await groupStream(1000, version)); - // Starts before the group below it. - await subscriber.runGroup(new Group({ subscribe: 0n, sequence: 5 }), await groupStream(500, version)); - // Lands below group 5 but starts after it. - await subscriber.runGroup(new Group({ subscribe: 0n, sequence: 3 }), await groupStream(2000, version)); - const got = []; - for (let i = 0; i < 3; i++) { - const group = await track.recvGroup(); - got.push([group?.sequence, (await group?.readFrame())?.payload.length]); - } - expect(got).toEqual([ - [3, 1], - [4, 1], - [5, 1], - ]); - expect(track.closed.peek()).toBeUndefined(); - } finally { - track.close(); - broadcast.close(); - subscriber.close(); +// Lite-05/06 stamp an untimed track's frames with their send time, which a receiver cannot tell +// from media time, so it drops every timestamp there: a group that rewinds or lands out of order is +// accepted untimed. +test.each([Version.DRAFT_05, Version.DRAFT_06])("version %p drops received timestamps", async (version) => { + const { quic, streams, opened } = fakeSession(); + const subscriber = new Subscriber(quic, version, HopSchema.parse(1n)); + const broadcast = subscriber.consume(Path.from("room")); + // Old enough to keep every group, however far behind the newest start. + const track = broadcast.track("video").subscribe({ maxDelay: Milli(60_000) }); + try { + await opened(0); + await answerTrackInfo(streams[0], version, 1000); + await (await opened(1)).reading; + await subscriber.runGroup(new Group({ subscribe: 0n, sequence: 4 }), await groupStream(1000, version)); + // Starts before the group below it. + await subscriber.runGroup(new Group({ subscribe: 0n, sequence: 5 }), await groupStream(500, version)); + // Lands below group 5 but starts after it. + await subscriber.runGroup(new Group({ subscribe: 0n, sequence: 3 }), await groupStream(2000, version)); + const got = []; + for (let i = 0; i < 3; i++) { + const group = await track.recvGroup(); + const frame = await group?.readFrame(); + got.push([group?.sequence, frame?.payload.length, frame?.timestamp]); } - }, -); + expect(got).toEqual([ + [3, 1, undefined], + [4, 1, undefined], + [5, 1, undefined], + ]); + expect((await track.info()).timescale).toBeUndefined(); + expect(track.closed.peek()).toBeUndefined(); + } finally { + track.close(); + broadcast.close(); + subscriber.close(); + } +}); + +// Lite-07 timestamps are media time, so a received group is checked like a local write. +test("a lite-07 rewind is refused", async () => { + const version = Version.DRAFT_07; + const { quic, streams, opened } = fakeSession(); + const subscriber = new Subscriber(quic, version, HopSchema.parse(1n)); + const broadcast = subscriber.consume(Path.from("room")); + const track = broadcast.track("video").subscribe({ maxDelay: Milli(60_000) }); + try { + await opened(0); + await answerTrackInfo(streams[0], version, 1000); + await (await opened(1)).reading; + await subscriber.runGroup(new Group({ subscribe: 0n, sequence: 4 }), await groupStream(1000, version)); + // Starts before the group below it, so its stream is stopped with the refusal. + const rewound = await groupStream(500, version); + const stop = spyOn(rewound, "stop"); + await subscriber.runGroup(new Group({ subscribe: 0n, sequence: 5 }), rewound); + expect(stop.mock.calls[0]?.[0]).toBeInstanceOf(TimestampMismatch); + } finally { + track.close(); + broadcast.close(); + subscriber.close(); + } +}); test("a publisher FIN without SUBSCRIBE_END still violates an open subscription", async () => { const { quic, streams, opened, closes } = fakeSession(); diff --git a/js/net/src/lite/subscriber.ts b/js/net/src/lite/subscriber.ts index 7d4c3ab8196..b780692a3d6 100644 --- a/js/net/src/lite/subscriber.ts +++ b/js/net/src/lite/subscriber.ts @@ -55,6 +55,7 @@ import { hasAnnounceOk, hasDatagrams, hasFetchMaxDelay, + hasMediaTime, hasOrder, hasProbeRtt, hasRefresh, @@ -863,7 +864,6 @@ export class Subscriber { timescale.set(0); drainOk = true; } - hooks.receiveTrack(producer); // Lite-06+ resolves where the feed starts only with SUBSCRIBE_START, so a relay serving // this copy holds its first group until the answer arrives. if (resolvesStart(this.version)) hooks.startTrack(producer, { requested: msg.startGroup }); @@ -965,7 +965,8 @@ export class Subscriber { // Map the wire TRACK_INFO onto the model track.Info a producer/consumer holds. #toModelInfo(info: TrackInfo): track.Info { return { - timescale: info.timescale, + // Lite-05/06 timestamps may be send times, so the copy is untimed and the ingest drops them. + timescale: hasMediaTime(this.version) ? info.timescale : undefined, // 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 : Milli(info.maxAge), @@ -1164,7 +1165,7 @@ export class Subscriber { // the gap. async #runFetchResponse(stream: Stream, group: netGroup.Producer, timescale: number): Promise { try { - const decode = frameDecoder(timescale); + const decode = frameDecoder(timescale, hasMediaTime(this.version)); // Serve until the stream FINs, the group closes, or every reader leaves. A group can // stay open indefinitely (a catalog or JSON stream), so an abandoned fetch is stopped by @@ -1456,7 +1457,7 @@ export class Subscriber { scale = timescale.peek(); } - await readFrames(stream, producer, scale); + await readFrames(stream, producer, scale, hasMediaTime(this.version)); producer.close(); stream.stop(StreamCode.Cancel); @@ -1532,8 +1533,11 @@ export class Subscriber { const scale = entry.timescale.peek(); if (scale === undefined) return; + // Dropped where it may be a send time, like a frame's. const timestamp = - dg.timestamp === undefined ? undefined : new Time.Timestamp(dg.timestamp, Time.Timescale(scale)); + dg.timestamp === undefined || !hasMediaTime(this.version) + ? 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/track_stream.test.ts b/js/net/src/lite/track_stream.test.ts index 3b8f9ca6047..77579216486 100644 --- a/js/net/src/lite/track_stream.test.ts +++ b/js/net/src/lite/track_stream.test.ts @@ -382,4 +382,36 @@ describe.each([Version.DRAFT_05, Version.DRAFT_06, Version.DRAFT_07])("%s", (ver broadcast.close(); origin.close(); }); + + // Timestamps received over moq-transport may be arrival times, so they never enter moq-lite, + // whose receivers check every timestamp: the track is served untimed (milliseconds of send + // time before lite-07). + test("an unchecked track is served untimed", async () => { + const pair = createMockTransportPair(ALPN_05); + const origin = new OriginProducer(); + const publisher = new Publisher(pair.server, version, randomHop(), origin.consume()); + const broadcast = origin.createBroadcast(Path.from("room")); + broadcast.announce(); + const serving = (async () => { + for (;;) { + const request = await wireOf(broadcast).requested(); + if (!request) return; + request.accept({ timescale: Timescale.MICRO, unchecked: true }); + } + })(); + + const client = await Stream.open(pair.client, { version }); + const server = await Stream.accept(pair.server, version); + if (!server) throw new Error("the publisher never accepted the TRACK stream"); + void publisher.runTrackInfo(new TrackMessage(Path.from("room"), "video"), server); + + const info = await TrackInfo.decode(client.reader, version); + expect(info.timescale).toBe(version === Version.DRAFT_07 ? undefined : Timescale.MILLI); + + client.writer.close(); + publisher.close(); + broadcast.close(); + await serving; + origin.close(); + }); }); diff --git a/js/net/src/lite/version.ts b/js/net/src/lite/version.ts index fd159544c6c..45907a01f4e 100644 --- a/js/net/src/lite/version.ts +++ b/js/net/src/lite/version.ts @@ -423,6 +423,24 @@ export function hasRefresh(version: Version): boolean { } } +/** Whether received frame timestamps are the publisher's media time. Lite-05/06 stamp an untimed + * track's payloads with their send time, which a receiver cannot tell from it, so it drops every + * timestamp received on them and the track arrives untimed. */ +export function hasMediaTime(version: Version): boolean { + // Explicitly list older versions so future versions keep the lite-07+ behavior. + switch (version) { + case Version.DRAFT_01: + case Version.DRAFT_02: + case Version.DRAFT_03: + case Version.DRAFT_04: + case Version.DRAFT_05: + case Version.DRAFT_06: + return false; + default: + return true; + } +} + /// The WebTransport subprotocol identifier for moq-lite. /// Version negotiation still happens via SETUP when this is used. export const ALPN = "moql"; diff --git a/js/net/src/track.ts b/js/net/src/track.ts index e01dc0d198d..c2608641cf4 100644 --- a/js/net/src/track.ts +++ b/js/net/src/track.ts @@ -129,6 +129,15 @@ export interface Info { * snapshot or keyframe. `0` asks on any join after a delta. An integer in `0..2^32`. */ refresh?: number; + /** + * Whether this timed track's timestamps arrived over moq-transport, whose peers may stamp + * arrival times, so no group is checked for a rewind. A moq-lite hop serves the track + * untimed, since its receivers check every timestamp. Mirrors the Rust + * `track::Info::unchecked`. + * + * @internal + */ + unchecked?: boolean; } // Normalize a duration for the wire, which carries it as an unsigned varint. @@ -155,13 +164,23 @@ function priorityByte(value: number): number { return value; } +// private: passthrough +/** The timescale a moq-lite hop serves a track at: none for an {@link Info.unchecked} one. @internal */ +export function liteTimescale(info: Info): Timescale | undefined { + return info.unchecked ? undefined : (info.timescale ?? undefined); +} + /** Fill in any unset {@link Info} fields with their defaults. */ export function infoDefaults(info: Partial = {}): Info { + const timescale = info.timescale == null ? undefined : Timescale(info.timescale); + // Only a timed track has timestamps to doubt. + if (info.unchecked && timescale === undefined) throw new RangeError("an untimed track can't be unchecked"); return { - timescale: info.timescale == null ? undefined : Timescale(info.timescale), + timescale, maxAge: info.maxAge === undefined ? undefined : wireMillis("maxAge", info.maxAge), priority: priorityByte(info.priority ?? DEFAULT_PRIORITY), refresh: info.refresh == null ? undefined : refreshRatio(info.refresh), + ...(info.unchecked ? { unchecked: true } : {}), }; } @@ -703,9 +722,6 @@ export class Producer { // group above it is checked against it as well as its cached neighbors; one at or below // it may have lost its lower bound. One entry, so nothing grows with the cache. #evicted?: Start; - // A session's copy of a remote track, whose groups are never checked for a rewind: a - // received timestamp may not be media time (a lite-05/06 untimed track carries send times). - #remote = false; // When the cache was last scanned. See #prune. #pruned = Number.NEGATIVE_INFINITY; // When the track closed, which restarts the former newest group's idle window. See #prune. @@ -762,9 +778,6 @@ export class Producer { producer.#owe(); for (const state of [producer.#state, ...producer.#sinks]) state.startChanged.update((rev) => rev + 1); }; - hooks.receiveTrack = (producer) => { - producer.#remote = true; - }; // Anything written before binding still counts, so the namespace never goes backwards. bindProducerSequence = (producer, sequence) => { const own = producer.#sequence; @@ -1268,8 +1281,8 @@ export class Producer { this.#drop(group.sequence, group.sequence + 1, closed); }); const sequence = group.sequence; - if (this.#remote) { - // Only a local publisher's writes are checked. + if (this.#unchecked()) { + // Its timestamps may rewind, so they are not checked. } else if (start) { this.#record(entry, start, late); } else { @@ -1402,12 +1415,17 @@ export class Producer { } // A group written before it joined is checked here, before it is cached or fanned out. - const start = this.#remote ? undefined : hooks.groupStart(group); + const start = this.#unchecked() ? undefined : hooks.groupStart(group); const late = start !== undefined && this.#admit(group.sequence, start); this.#publish(group, start, late); } + // Whether this track's timestamps may rewind (see {@link Info.unchecked}). + #unchecked(): boolean { + return this.#state.info.peek()?.unchecked === true; + } + // Fan a datagram out to every live subscriber, dropping the oldest once the ring is full. // Late subscribers do NOT replay old datagrams (best-effort, unlike the group cache). #publishDatagram(datagram: Datagram): void { diff --git a/js/net/src/untimed.test.ts b/js/net/src/untimed.test.ts index 146a85c52c6..921a0284739 100644 --- a/js/net/src/untimed.test.ts +++ b/js/net/src/untimed.test.ts @@ -18,23 +18,21 @@ interface Wire { name: string; protocol: string; version?: number; - /** Whether the wire can declare a track's timeline. */ + /** Whether a timed track arrives timed. Lite-05/06 receivers drop timestamps, which may be send times. */ declares: boolean; - /** Whether the wire carries a timestamp on every frame, a send time for an untimed one. */ - stamps: boolean; } const WIRES: Wire[] = [ - { name: "lite-01", protocol: "", version: Lite.Version.DRAFT_01, declares: false, stamps: false }, - { name: "lite-03", protocol: Lite.ALPN_03, declares: false, stamps: false }, - { 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: 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 }, - { name: "ietf-20", protocol: Ietf.ALPN.DRAFT_20, declares: true, stamps: false }, + { name: "lite-01", protocol: "", version: Lite.Version.DRAFT_01, declares: false }, + { name: "lite-03", protocol: Lite.ALPN_03, declares: false }, + { name: "lite-04", protocol: Lite.ALPN_04, declares: false }, + { name: "lite-05", protocol: Lite.ALPN_05, declares: false }, + { name: "lite-06", protocol: Lite.ALPN_06, declares: false }, + { name: "lite-07-wip", protocol: Lite.ALPN_07_WIP, declares: true }, + { name: "ietf-14", protocol: "", version: Ietf.Version.DRAFT_14, declares: false }, + { name: "ietf-16", protocol: Ietf.ALPN.DRAFT_16, declares: false }, + { name: "ietf-17", protocol: Ietf.ALPN.DRAFT_17, declares: true }, + { name: "ietf-20", protocol: Ietf.ALPN.DRAFT_20, declares: true }, ]; /** Serve one group of `timestamps` on a track with `timescale`, and read it back over `wire`. */ @@ -156,21 +154,14 @@ test("a forwarded untimed track stays untimed", async () => { }); for (const wire of WIRES) { - test(`${wire.name}: an untimed track arrives untimed, or with send times where the wire must stamp`, async () => { + test(`${wire.name}: an untimed track arrives untimed`, async () => { const { info, frames } = await roundTrip(wire, undefined, [undefined, undefined]); expect(frames.map((frame) => frame.payload[0])).toEqual([0, 1]); - - if (wire.stamps) { - // Lite-05 and lite-06 can't say a track is untimed, so it goes out timed at milliseconds. - expect(info.timescale).toBe(Timescale.MILLI); - for (const frame of frames) expect(frame.timestamp).toBeDefined(); - } else { - expect(info.timescale).toBeUndefined(); - for (const frame of frames) expect(frame.timestamp).toBeUndefined(); - } + expect(info.timescale).toBeUndefined(); + for (const frame of frames) expect(frame.timestamp).toBeUndefined(); }); - test(`${wire.name}: a timed track keeps its timestamps only where the wire declares units`, async () => { + test(`${wire.name}: a timed track keeps its timestamps only where the wire delivers them`, async () => { const { info, frames } = await roundTrip(wire, Timescale.MICRO, [ Timestamp.fromMicros(1_000), Timestamp.fromMicros(1_234), @@ -181,7 +172,7 @@ for (const wire of WIRES) { expect(info.timescale).toBe(Timescale.MICRO); expect(frames.map((frame) => frame.timestamp?.asMicros())).toEqual([1_000, 1_234]); } else { - // No units on the wire, and no arrival time made up in their place. + // No trusted units on the wire, and no arrival time made up in their place. expect(info.timescale).toBeUndefined(); for (const frame of frames) expect(frame.timestamp).toBeUndefined(); } @@ -189,7 +180,7 @@ 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 () => { + test(`${wire.name}: an untimed datagram arrives untimed`, async () => { const pair = createMockTransportPair(wire.protocol); const origin = new OriginProducer(); const [client, server] = await Promise.all([ @@ -212,8 +203,7 @@ for (const wire of WIRES.filter((wire) => wire.name === "lite-06" || wire.name = const datagram = await subscriber.recvDatagram(); expect(datagram?.payload).toEqual(new Uint8Array([7])); - if (wire.stamps) expect(datagram?.timestamp).toBeDefined(); - else expect(datagram?.timestamp).toBeUndefined(); + expect(datagram?.timestamp).toBeUndefined(); broadcast.close(); subscriber.close(); diff --git a/quest/m1/README.md b/quest/m1/README.md index 5e5e87c05b5..124f3d26cad 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -112,7 +112,6 @@ lite-07 stream type here. - [Resumed datagrams](/quest/m1/resume-datagrams.md) - a resumed track skips only datagrams it already delivered, not every lower sequence - [Recovery verdicts](/quest/m1/recovery-verdicts.md) - group recovery refetches only on errors a new copy can fix, so a repeating error ends the group instead of looping - [Rust resume holds a declared end](/quest/m1/resume-rust-end.md) - a resumed Rust track keeps a declared end across copies, dedupes flapping copies, and judges retention inside the reader's window -- [Received timestamps are media time](/quest/m1/lite-time-provenance.md) - a received timed track's timestamps are media time, so the rewind check can run across hops - [WebGPU on Safari](/quest/m1/webgpu-safari.md) - the WebGPU renderer is verified on Safari 26 for macOS and iOS - [Close tail drain](/quest/m1/close-tail-drain.md) - after a peer close, subscribers read the group streams their transport already delivered instead of aborting them, so a finished tail arrives on every version - [Silent-peer close](/quest/m1/quic-silent-peer-drain.md) - moq-quic closes a connection that never heard its peer without the 3 s drain diff --git a/quest/m1/ietf-extension/serve-timed.md b/quest/m1/ietf-extension/serve-timed.md index c4669fc90b9..28e6ea043c8 100644 --- a/quest/m1/ietf-extension/serve-timed.md +++ b/quest/m1/ietf-extension/serve-timed.md @@ -24,7 +24,3 @@ the timestamps came from moq-transport and are unchecked. Mirror it in js/net, where `Info.unchecked` is `@internal`. If the timed encoding on that hop needs a wire change, update `drafts/draft-lcurley-moq-lite.md` (or the extension's draft) in the same PR. - -## Required - -- [Received timestamps are media time](/quest/m1/lite-time-provenance.md) - adds `unchecked` and the untimed serving this changes \ No newline at end of file diff --git a/quest/m1/lite-time-provenance.md b/quest/m1/lite-time-provenance.md deleted file mode 100644 index 586562c9b38..00000000000 --- a/quest/m1/lite-time-provenance.md +++ /dev/null @@ -1,38 +0,0 @@ -# [M] Received timestamps are media time - -## Goal - -A receiver can trust a received timed track's timestamps as media time, so -the rewind check that today runs only on a local publisher's writes can run -across hops too. - -## Plan - -Split from [#5803](https://github.com/moq-dev/moq/pull/5803), which refuses a -rewind only on a local publisher's writes: a relay hop that checked received -groups closed tracks whose timestamps were never media time. - -Where non-media timestamps come from today: - -- Released clients stamp wall-clock write time on data tracks: the moq-mux - catalog, moq-json, moq-stats, moq-room chat, js/json, js/binary, and - js/net's `writeString`, `writeJson` and `writeBool`. -- Released relays stamp arrival time on input that carries no timestamps - (IETF, and lite-04 or older) and forward it as media time. -- Unreleased main stamps send time on untimed tracks over lite-05/06, which - cannot encode an untimed track - ([#4822](https://github.com/moq-dev/moq/pull/4822), - [#4968](https://github.com/moq-dev/moq/pull/4968)). -- Media tracks carry real presentation times. - -Options weighed: - -1. Fix our unreleased publishers and relays so only media time reaches the - wire, and stop stamping data tracks. Recommended. -2. Reserve an untimed marker value on lite-05/06. -3. Treat a 1000 timescale as untimed. -4. Distrust every lite-05/06 timestamp. -5. A lite-07 TRACK_INFO marker for timestamp provenance. - -Released peers still send wall-clock stamps, so whichever option lands must -say which received tracks the check may trust. diff --git a/quest/m1/public-model/ingest.md b/quest/m1/public-model/ingest.md index 3c328a23b16..2f7305125d3 100644 --- a/quest/m1/public-model/ingest.md +++ b/quest/m1/public-model/ingest.md @@ -170,11 +170,11 @@ Rust items (`rs/moq-net/src/model/`), as they stood at b4eab0db1: `track::TrackState`, `track::FetchOutcome`, `finish_track_at`, `poll_landed`): moved to [the deferred quest](/quest/m1/public-model/deferred.md), which waits on Model ranges. -- `track::Producer::received` (lite and IETF subscribers; JS - `hooks.receiveTrack`) marks a session's copy so the local-write rewind - check skips it (#5803). It goes public with the other ingest calls, unless - [time provenance](/quest/m1/lite-time-provenance.md) makes received - timestamps checkable first. +- `track::Info::unchecked` (JS `Info.unchecked`, `@internal`) marks a + track whose timestamps arrived over moq-transport, so no group is checked + and moq-lite serves it untimed. A custom session's copies stay checked; + expose it with the other ingest calls only if one must ingest arrival + times. - Passthrough, stay private and marked: `group::Producer::wire`, `set_wire`, `mark_gap`, `set_end_object`, `group::Wire`, the `object` and `datagram` modules, `track::Producer::insert_wire_datagram`, diff --git a/quest/m1/ts-reliability/arrival-stamps.md b/quest/m1/ts-reliability/arrival-stamps.md index 59b6e1f8f79..271c47d2300 100644 --- a/quest/m1/ts-reliability/arrival-stamps.md +++ b/quest/m1/ts-reliability/arrival-stamps.md @@ -10,7 +10,7 @@ path no longer lets held frames out past their deadline: the ignored arm `#[ignore]`. Non-goals: wall-clock or cross-hop provenance -([received timestamps are media time](/quest/m1/lite-time-provenance.md)), +([received timestamps are media time](https://github.com/moq-dev/moq/pull/6150)), and any wire change. ## Plan diff --git a/rs/moq-cli/src/publish.rs b/rs/moq-cli/src/publish.rs index 66651054d18..f7f54952fbe 100644 --- a/rs/moq-cli/src/publish.rs +++ b/rs/moq-cli/src/publish.rs @@ -1185,7 +1185,7 @@ pub(crate) mod tests { let _env = crate::test_env::EnvGuard::clear(&["MOQ_CONNECT", "MOQ_HOP"]); let _ = moq_tokio::crypto::install_default(); - for version in ["moq-lite-06", "moq-transport-14"] { + for version in ["moq-lite-06", "moq-lite-07-wip", "moq-transport-14"] { let fixture = moq_relay::test_relay().await.unwrap(); let ready = fixture.relay.ready(); tokio::spawn(fixture.relay.run()); @@ -1339,8 +1339,9 @@ pub(crate) mod tests { // Joining once every group is out, with a budget half the jump, reads every group // after it past the marker, and none of the groups the jump left more than // the budget behind. An IETF subscription starts at the newest group, so only lite - // has the groups behind it to read. - if version.starts_with("moq-lite") { + // has the groups behind it to read, and a lite-06 subscriber drops the timestamps + // that carry the jump. + if version == "moq-lite-07-wip" { let late = read(start(moq_net::track::Subscription::default().with_max_delay(Duration::from_secs(10))).await) .await diff --git a/rs/moq-net/MIGRATE.md b/rs/moq-net/MIGRATE.md index c0be3ad9d98..c2e5a6930ec 100644 --- a/rs/moq-net/MIGRATE.md +++ b/rs/moq-net/MIGRATE.md @@ -119,7 +119,9 @@ let rate = bandwidth.as_ref().and_then(|b| b.peek()); ### A timed track refuses a group that rewinds (#5803) A local write whose first frame starts before a lower-sequence group's, or after a higher one's, fails with `TimestampMismatch`; `@moq/net` throws it the same way. -Received groups are accepted as is, since their timestamps may be send times. +A group received over moq-lite-07 is checked the same way. +A track received over moq-lite-05/06 arrives untimed, since its timestamps may be send times: read times from the payload, such as the hang container's, instead of `frame::Frame::timestamp`. +One received over moq-transport keeps its timestamps unchecked, and a relay serves it to moq-lite untimed. A publisher whose timeline restarts, such as an HLS `EXT-X-DISCONTINUITY` that resets decode times, publishes a new broadcast instead. ### `broadcast::Producer::finish` is back, and lingers (#5795) diff --git a/rs/moq-net/src/ietf/subscriber.rs b/rs/moq-net/src/ietf/subscriber.rs index ce40f19cd30..d9f52d58821 100644 --- a/rs/moq-net/src/ietf/subscriber.rs +++ b/rs/moq-net/src/ietf/subscriber.rs @@ -480,6 +480,8 @@ fn accepted_info(properties: &ietf::Properties, version: Version) -> track::Info version, values: properties.unknown.clone(), })); + // A moq-transport peer may stamp arrival times, and nothing on the wire says it didn't. + info.unchecked = info.timescale.is_some(); info } @@ -3031,7 +3033,6 @@ where (request.accept(info), dynamic) } }; - track.received(); // A live join starts at the publisher's edge; an absolute one where it asked. With // no Largest the join skips nothing, so every group is owed: a lower one's stream // may still arrive after a higher one's. @@ -3893,7 +3894,6 @@ where tracing::debug!(broadcast = %self.origin.absolute(broadcast_path), track = %track_name, "track status known"); let dynamic = request.dynamic(); let mut track = request.accept(accepted_info(&ok.properties, self.version)); - track.received(); // Nothing feeds this copy live: what it caches is fetched, never the live edge. track.set_idle(); Some(Idle { @@ -12184,12 +12184,14 @@ mod stitch_tests { ); // The subscriber accepts a timed track at microseconds, matching `run_subscribe`. - let track = track::Producer::new( - std::sync::Arc::new(crate::broadcast::Info::default()), - "video", - track::Info::default().with_timescale(timescale.map(|_| Timescale::MICRO)), + let info = accepted_info( + &ietf::Properties { + timescale: timescale.map(|_| Timescale::MICRO), + ..Default::default() + }, + VERSION, ); - track.received(); + let track = track::Producer::new(std::sync::Arc::new(crate::broadcast::Info::default()), "video", info); let fill = kio::Producer::new(fill); { @@ -12353,30 +12355,19 @@ mod stitch_tests { settled.await; } - /// Object numbering is relay metadata; plain readers still see consecutive frames. - async fn relay_passthrough( - script: Vec, - ingress: Version, - egress: Version, - fetch: Option<(u64, u64)>, - ) -> (Vec, crate::lite::test_transport::Log) { - relay_passthrough_timed(script, ingress, egress, fetch, None).await - } - - /// Relay `script` through a track the ingress times in `timescale`, which the cache - /// normalizes to microseconds like [`accepted_info`]. + /// Relay `script` through an untimed track. /// /// Also relays it through a second relay behind a moq-lite hop with the `ietf` - /// extension, which must not change a byte. - async fn relay_passthrough_timed( + /// extension, which must not change a byte. Object numbering is relay metadata; plain + /// readers still see consecutive frames. + async fn relay_passthrough( script: Vec, ingress: Version, egress: Version, fetch: Option<(u64, u64)>, - timescale: Option, ) -> (Vec, crate::lite::test_transport::Log) { - let direct = relay_passthrough_via(script.clone(), ingress, egress, fetch, timescale, false).await; - let hopped = relay_passthrough_via(script, ingress, egress, fetch, timescale, true).await; + let direct = relay_passthrough_via(script.clone(), ingress, egress, fetch, None, false).await; + let hopped = relay_passthrough_via(script, ingress, egress, fetch, None, true).await; let fetch_ok = |bytes: &[u8]| { let mut decoder = Decoder::new(bytes, egress.into()); (decoder.varint().ok()? == ietf::FetchOk::ID).then_some(())?; @@ -12418,7 +12409,9 @@ mod stitch_tests { direct } - /// [`relay_passthrough_timed`] through one relay, or two joined by a moq-lite `hop`. + /// Relay `script` through a track the ingress times in `timescale`, which the cache + /// normalizes to microseconds like [`accepted_info`], via one relay or two joined by a + /// moq-lite `hop`. async fn relay_passthrough_via( script: Vec, ingress: Version, @@ -12605,22 +12598,30 @@ mod stitch_tests { /// A relayed Timestamp leaves in the units the publisher declared: the cache holds the /// track on the publisher's grid, so a copy of the same instance over another route - /// matches it. + /// matches it. A moq-lite hop serves the unchecked track untimed, so the Timestamp + /// stops there and the other property survives. #[moq_clock::test] async fn passthrough_keeps_the_publishers_timestamp_units() { let version = Version::Draft22; // Timestamp 42 (ms), then an unknown property that must survive beside it. let properties = [0x10, 42, 0x70, 7]; let script = object_stream(version, true, &[(0, &properties, None)]); - let (bytes, log) = relay_passthrough_timed(script, version, version, None, Some(Timescale::MILLI)).await; - assert!(log.resets().is_empty()); - let mut decoder = Decoder::new(&bytes, version.into()); - ietf::GroupHeader::decode(&mut decoder, version).unwrap(); - assert_eq!(decoder.varint().unwrap(), 0); - let relayed = bytes::Bytes::copy_from_slice(decoder.bytes().unwrap()); - let decoded = super::super::object::read(&relayed, version, Some(Timescale::MILLI)).unwrap(); - assert_eq!(decoded.timestamp, Some(Timestamp::new(42, Timescale::MILLI).unwrap())); - assert_eq!(super::super::object::listed(&decoded.properties, version).len(), 1); + for (hop, timestamp) in [(false, Timestamp::new(42, Timescale::MILLI).ok()), (true, None)] { + let (bytes, log) = + relay_passthrough_via(script.clone(), version, version, None, Some(Timescale::MILLI), hop).await; + assert!(log.resets().is_empty(), "hop: {hop}"); + let mut decoder = Decoder::new(&bytes, version.into()); + ietf::GroupHeader::decode(&mut decoder, version).unwrap(); + assert_eq!(decoder.varint().unwrap(), 0); + let relayed = bytes::Bytes::copy_from_slice(decoder.bytes().unwrap()); + let decoded = super::super::object::read(&relayed, version, timestamp.map(|_| Timescale::MILLI)).unwrap(); + assert_eq!(decoded.timestamp, timestamp, "hop: {hop}"); + assert_eq!( + super::super::object::listed(&decoded.properties, version).len(), + 1, + "hop: {hop}" + ); + } } fn object_stream(version: Version, first: bool, objects: &[(u64, &[u8], Option)]) -> Vec { @@ -14259,8 +14260,8 @@ mod stitch_tests { assert!(h.session.log.closes().is_empty()); } - /// A received group whose first object rewinds past the group before it is accepted: - /// only a local publisher's writes refuse a rewind. + /// A received group whose first object rewinds past the group before it is accepted: a + /// moq-transport peer may stamp arrival times, so its timed tracks are unchecked. #[moq_clock::test] async fn a_received_rewind_is_accepted() { use futures::FutureExt; diff --git a/rs/moq-net/src/lite/publisher.rs b/rs/moq-net/src/lite/publisher.rs index 5f318539282..4bae56c7831 100644 --- a/rs/moq-net/src/lite/publisher.rs +++ b/rs/moq-net/src/lite/publisher.rs @@ -1957,14 +1957,7 @@ impl Request for TrackInfoServe { let info = ready!(self.querying.poll_ok(waiter))?; // TRACK_INFO only flows on Lite05+ (the encode errors otherwise). - writer.buffer(&lite::TrackInfo { - refresh: info.refresh, - priority: info.priority, - max_age: info.max_age, - timescale: info.timescale, - // An older peer is served the frames inflated. - compress: info.compress && shared.version.has_compression(), - })?; + writer.buffer(&lite::TrackInfo::serve(&info, shared.version))?; if ietf { writer.buffer(&lite::ietf::TrackProperties(info.properties.clone()))?; } @@ -2021,14 +2014,7 @@ impl Request for TracksServe { let event = match ready!(self.tracks.poll_next(waiter))? { Some(crate::broadcast::TracksEvent::Added { name, info }) => lite::TracksEvent::Added { name: name.to_string().into(), - info: lite::TrackInfo { - refresh: info.refresh, - priority: info.priority, - max_age: info.max_age, - timescale: info.timescale, - // An older peer is served the frames inflated. - compress: info.compress && shared.version.has_compression(), - }, + info: lite::TrackInfo::serve(&info, shared.version), }, Some(crate::broadcast::TracksEvent::Removed { name }) => lite::TracksEvent::Removed { name: name.to_string().into(), @@ -2141,7 +2127,7 @@ impl Request for SubscribeServe { // `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); + let stamp = Stamp::new(shared.version, track.info().lite_timescale()); // Lite05+ accepts implicitly: no SUBSCRIBE_OK, the immutable properties // live in TRACK_INFO, and the resolved range arrives as @@ -2178,7 +2164,7 @@ impl Request for SubscribeServe { steer: steer_tx.consume(), steer_seen: steer, version: shared.version, - timescale, + stamp, runtime: shared.runtime.clone(), outcomes: Default::default(), ietf, @@ -2238,7 +2224,7 @@ enum FetchServe { /// (the subscriber decodes against the same baseline). Serve { group: group::Consumer, - timescale: Option, + stamp: Option, prev_ts: u64, frame: Option, chunk: Option, @@ -2315,17 +2301,18 @@ impl Request for FetchServe { group.end_at(msg.end_frame.map_or(Bound::Unbounded, Bound::Included)); // Matches the timescale TRACK_INFO declared for the track. - let timescale = shared.version.wire_timescale(group.timescale()); + let stamp = Stamp::new(shared.version, group.lite_timescale()); // And its compression, which a lite-07 peer receives as stored. if shared.version.has_compression() { group = group.compressed(); } let capped = msg.end_frame.is_some(); + let timescale = stamp.map(|stamp| stamp.timescale); let records = lite::ietf::Records::start(writer, ietf, &group, capped, timescale)?; *self = Self::Serve { group, - timescale, + stamp, prev_ts: 0, frame: None, chunk: None, @@ -2336,7 +2323,7 @@ impl Request for FetchServe { } Self::Serve { group, - timescale, + stamp, prev_ts, frame, chunk, @@ -2368,7 +2355,7 @@ impl Request for FetchServe { batched.frame.timestamp, batched.inflated, batched.frame.payload.len() as u64, - *timescale, + *stamp, prev_ts, &shared.runtime, )?; @@ -2403,7 +2390,7 @@ impl Request for FetchServe { next.timestamp, next.inflated, next.size, - *timescale, + *stamp, prev_ts, &shared.runtime, )?; @@ -3361,12 +3348,12 @@ fn buffer_frame_info( timestamp: Option, inflated: Option, size: u64, - timescale: Option, + stamp: Option, prev_ts: &mut u64, runtime: &crate::time::Deadlines, ) -> Result<(), Error> { - if let Some(timescale) = timescale { - buffer_zigzag_delta(writer, wire_timestamp(timestamp, timescale, runtime)?, prev_ts)?; + if let Some(stamp) = stamp { + buffer_zigzag_delta(writer, stamp.wire(timestamp, runtime)?, prev_ts)?; } // Only a compressed group read in its stored form has one, and only on lite-07. if let Some(inflated) = inflated { @@ -3376,22 +3363,35 @@ fn buffer_frame_info( Ok(()) } -/// A frame or datagram timestamp as its raw value at the wire `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, +/// How a lite hop stamps a track's frames and datagrams. +#[derive(Clone, Copy, Debug)] +struct Stamp { + /// The wire timescale TRACK_INFO declared (see [`Version::wire_timescale`]). timescale: crate::Timescale, - runtime: &crate::time::Deadlines, -) -> Result { - match timestamp { - Some(timestamp) => Ok(timestamp.value()), - None => crate::Timestamp::from(runtime.now()) - .convert(timescale) - .map(|now| now.value()) - .map_err(|_| Error::BoundsExceeded(crate::coding::BoundsExceeded)), + /// Whether each frame's own timestamp goes out. Lite05/06 cannot encode an untimed + /// track, so one served untimed carries its send time instead, unchecked ones included. + timed: bool, +} + +impl Stamp { + /// How `version` stamps a track served at `timescale` (see + /// [`track::Info::lite_timescale`]), or `None` when it carries no timestamps. + fn new(version: Version, timescale: Option) -> Option { + Some(Self { + timescale: version.wire_timescale(timescale)?, + timed: timescale.is_some(), + }) + } + + /// A frame or datagram's `timestamp` as its raw wire value, or its send time on + /// `runtime` when the track goes out untimed. + fn wire(self, timestamp: Option, runtime: &crate::time::Deadlines) -> Result { + timestamp + .filter(|_| self.timed) + .unwrap_or_else(|| crate::Timestamp::from(runtime.now())) + .convert(self.timescale) + .map(|time| time.value()) + .map_err(|_| Error::BoundsExceeded(crate::coding::BoundsExceeded)) } } @@ -3615,9 +3615,8 @@ struct Subscription { /// Last steer observed by this clone, so a change only fires once. steer_seen: Steer, version: Version, - /// The track's wire timescale, or `None` when its frames and datagrams carry no - /// timestamp (see [`Version::wire_timescale`]). - timescale: Option, + /// How frames and datagrams are stamped, or `None` when they carry no timestamp. + stamp: Option, /// The clock an untimed track's Lite05/06 send times are read from. runtime: crate::time::Deadlines, /// Outcomes shared by the subscription and every group it serves. @@ -3653,9 +3652,9 @@ impl Subscription { return false; } let Ok(timestamp) = self - .timescale + .stamp .filter(|_| !status) - .map(|timescale| wire_timestamp(datagram.timestamp, timescale, &self.runtime)) + .map(|stamp| stamp.wire(datagram.timestamp, &self.runtime)) .transpose() else { return false; @@ -3668,7 +3667,7 @@ impl Subscription { if wire .properties .sealed_time - .is_some_and(|sealed| self.timescale != Some(sealed.scale())) => + .is_some_and(|sealed| self.stamp.map(|stamp| stamp.timescale) != Some(sealed.scale())) => { return false; } @@ -3685,7 +3684,7 @@ impl Subscription { payload: payload.clone(), ietf, }; - // has_datagrams is checked before this runs, and the timescale is set wherever the + // has_datagrams is checked before this runs, and the stamp 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; @@ -4377,7 +4376,13 @@ impl GroupServe { }; let header = writer.buffer(&lite::DataType::Group).and_then(|()| writer.buffer(&msg)); let records = header.and_then(|()| { - lite::ietf::Records::start(&mut writer, self.ctx.ietf, &self.group, self.capped, self.ctx.timescale) + lite::ietf::Records::start( + &mut writer, + self.ctx.ietf, + &self.group, + self.capped, + self.ctx.stamp.map(|stamp| stamp.timescale), + ) }); let records = match records { Ok(records) => records, @@ -4514,7 +4519,7 @@ impl GroupServe { batched.frame.timestamp, batched.inflated, batched.frame.payload.len() as u64, - self.ctx.timescale, + self.ctx.stamp, &mut self.prev_ts, &self.ctx.runtime, ) @@ -4560,7 +4565,7 @@ impl GroupServe { next.timestamp, next.inflated, next.size, - self.ctx.timescale, + self.ctx.stamp, &mut self.prev_ts, &self.ctx.runtime, ) @@ -4784,6 +4789,26 @@ mod serve_group_test { } } + /// A track served untimed to lite-05/06 carries its send time, never its frames' own + /// timestamps: those of an unchecked track may be moq-transport arrival times. + #[moq_clock::test] + async fn an_unchecked_track_goes_out_with_send_times() { + let runtime = crate::time::sim(); + let stamped = Timestamp::from_millis(36_000_000).unwrap(); + let sent = Timestamp::from(runtime.now()).value(); + let checked = track::Info::timed(); + let unchecked = track::Info { + unchecked: true, + ..track::Info::timed() + }; + + let stamp = Stamp::new(Version::Lite06, checked.lite_timescale()).unwrap(); + assert_eq!(stamp.wire(Some(stamped), &runtime).unwrap(), stamped.value()); + let stamp = Stamp::new(Version::Lite06, unchecked.lite_timescale()).unwrap(); + assert_eq!(stamp.wire(Some(stamped), &runtime).unwrap(), sent); + assert!(Stamp::new(Version::Lite07, unchecked.lite_timescale()).is_none()); + } + #[moq_clock::test] async fn resets_with_the_abort_code() { let log = Log::default(); @@ -4799,7 +4824,7 @@ mod serve_group_test { steer: steer.consume(), steer_seen: Steer::default(), version: Version::Lite06, - timescale: Some(crate::Timescale::MILLI), + stamp: Stamp::new(Version::Lite06, Some(crate::Timescale::MILLI)), runtime: crate::time::sim(), outcomes: Default::default(), ietf: false, @@ -4845,7 +4870,7 @@ mod serve_group_test { steer: steer.consume(), steer_seen: Steer::default(), version: Version::Lite06, - timescale: Some(crate::Timescale::MILLI), + stamp: Stamp::new(Version::Lite06, Some(crate::Timescale::MILLI)), runtime: crate::time::sim(), outcomes: Default::default(), ietf: false, @@ -4895,7 +4920,7 @@ mod serve_group_test { steer: steer.consume(), steer_seen: Steer::default(), version: Version::Lite06, - timescale: Some(crate::Timescale::MILLI), + stamp: Stamp::new(Version::Lite06, Some(crate::Timescale::MILLI)), runtime: crate::time::sim(), outcomes: Default::default(), ietf: false, @@ -4945,7 +4970,7 @@ mod serve_group_test { steer: steer.consume(), steer_seen: Steer::default(), version: Version::Lite06, - timescale: Some(crate::Timescale::MILLI), + stamp: Stamp::new(Version::Lite06, Some(crate::Timescale::MILLI)), runtime: crate::time::sim(), outcomes: Default::default(), ietf: false, @@ -5013,7 +5038,7 @@ mod serve_group_test { steer: steer.consume(), steer_seen: Steer::default(), version: Version::Lite06, - timescale: Some(crate::Timescale::MILLI), + stamp: Stamp::new(Version::Lite06, Some(crate::Timescale::MILLI)), runtime: crate::time::sim(), outcomes: Default::default(), ietf: false, @@ -5186,7 +5211,7 @@ mod serve_group_test { steer: steer.consume(), steer_seen: Steer::default(), version: Version::Lite06, - timescale: Some(crate::Timescale::MILLI), + stamp: Stamp::new(Version::Lite06, Some(crate::Timescale::MILLI)), runtime: crate::time::sim(), outcomes: Default::default(), ietf: false, @@ -5261,7 +5286,7 @@ mod serve_group_test { steer: steer.consume(), steer_seen: Steer::default(), version: Version::Lite06, - timescale: Some(crate::Timescale::MILLI), + stamp: Stamp::new(Version::Lite06, Some(crate::Timescale::MILLI)), runtime: crate::time::sim(), outcomes: Default::default(), ietf: false, @@ -5310,7 +5335,7 @@ mod serve_group_test { steer: steer.consume(), steer_seen: Steer::default(), version: Version::Lite06, - timescale: Some(crate::Timescale::MILLI), + stamp: Stamp::new(Version::Lite06, Some(crate::Timescale::MILLI)), runtime: crate::time::sim(), outcomes: Default::default(), ietf: false, @@ -5358,7 +5383,7 @@ mod serve_group_test { steer: steer.consume(), steer_seen: Steer::default(), version: Version::Lite06, - timescale: Some(crate::Timescale::MILLI), + stamp: Stamp::new(Version::Lite06, Some(crate::Timescale::MILLI)), runtime: crate::time::sim(), outcomes: Default::default(), ietf: false, @@ -5418,7 +5443,7 @@ mod serve_group_test { steer: steer.consume(), steer_seen: Steer::default(), version, - timescale: Some(crate::Timescale::MILLI), + stamp: Stamp::new(Version::Lite06, Some(crate::Timescale::MILLI)), runtime: crate::time::sim(), outcomes: Default::default(), ietf: false, @@ -6550,7 +6575,7 @@ mod serve_group_test { steer: steer.consume(), steer_seen: Steer::default(), version: Version::Lite07, - timescale: Some(crate::Timescale::MILLI), + stamp: Stamp::new(Version::Lite07, Some(crate::Timescale::MILLI)), runtime: crate::time::sim(), outcomes: outcomes.clone(), ietf: false, diff --git a/rs/moq-net/src/lite/subscriber.rs b/rs/moq-net/src/lite/subscriber.rs index 35b07f5e909..04b3b7906af 100644 --- a/rs/moq-net/src/lite/subscriber.rs +++ b/rs/moq-net/src/lite/subscriber.rs @@ -524,7 +524,9 @@ impl Subscriber { .zip(entry.framing.timescale) .map(|(value, scale)| Timestamp::new(value, scale)) .transpose() - .map_err(|_| Error::BoundsExceeded(crate::coding::BoundsExceeded))?; + .map_err(|_| Error::BoundsExceeded(crate::coding::BoundsExceeded))? + // Dropped where it may be a send time, like a frame's. + .filter(|_| self.version.has_media_time()); // A datagram is never owed a stream, so it never holds the subscription's end open. if let Ok(mut tail) = entry.tail.write() { @@ -1038,6 +1040,9 @@ struct FrameIngest { /// `Some` once the Group State of the `ietf` extension arrived: every frame follows an /// object record. records: Option, + /// Whether the session's timestamps are media time. Lite-05/06 ones may be send times, + /// so they are read to stay in step and dropped. + media_time: bool, } /// A group's object records on ingest, under the `ietf` extension. @@ -1099,6 +1104,7 @@ impl FrameIngest { budget: subscriber.frames.clone(), frame_start, records: None, + media_time: subscriber.version.has_media_time(), } } @@ -1220,8 +1226,9 @@ impl FrameIngest { self.prev_ts = next; let timestamp = Timestamp::new(next, scale) .map_err(|_| Error::BoundsExceeded(crate::coding::BoundsExceeded))?; + // Read to stay in step, but dropped where it may be a send time. self.phase = IngestPhase::Inflated { - timestamp: Some(timestamp), + timestamp: self.media_time.then_some(timestamp), }; } IngestPhase::Inflated { timestamp } => { @@ -1819,13 +1826,7 @@ impl TracksRecv { return Poll::Ready(Err(Error::ProtocolViolation)); } let slot = source.subscriber.announces.acquire()?; - let info = track::Info::default() - .with_timescale(info.timescale) - .with_max_age(info.max_age) - .with_priority(info.priority) - .with_refresh(info.refresh) - .with_compress(info.compress); - source.lister.add(name.clone(), info)?; + source.lister.add(name.clone(), info.into())?; self.listed.insert(name, slot); } lite::TracksEvent::Removed { name } => { @@ -2616,10 +2617,11 @@ mod tests { going_away: Default::default(), auth: crate::auth::Handle::new(false), }); + // The track is untimed where the subscriber drops received timestamps. let track = track::Producer::new( std::sync::Arc::new(crate::broadcast::Info::default()), "video", - crate::track::Info::timed(), + crate::track::Info::default().with_timescale(version.has_media_time().then_some(Timescale::MILLI)), ); let mut consumer = track.consume().subscribe(None).await.unwrap(); subscriber.subscribes.lock().insert( @@ -2913,8 +2915,8 @@ mod tests { ); } - /// Receive group 5 starting at 500ms over `version`, on a received timed track whose - /// group 4 starts at 1000ms, returning the stream's result and the track. + /// Receive group 5 stamped 500ms over `version` into the copy a subscriber makes there, + /// whose group 4 started at 1000ms, returning the stream's result and the track. async fn receive_rewind(version: Version) -> (Result<(), Error>, track::Producer) { use crate::transport::poll::Session as _; @@ -2944,15 +2946,16 @@ mod tests { auth: crate::auth::Handle::new(false), }); - let track = track::Producer::new( - std::sync::Arc::new(crate::broadcast::Info::default()), - "video", - crate::track::Info::timed(), - ); - track.received(); + // The copy is untimed where the wire's timestamps may be send times. + let timed = version.has_media_time(); + let info = match timed { + true => crate::track::Info::timed(), + false => crate::track::Info::default(), + }; + let track = track::Producer::new(std::sync::Arc::new(crate::broadcast::Info::default()), "video", info); let mut before = track.create_group(4u64.into()).unwrap(); before - .write_frame(Timestamp::from_millis(1000).unwrap(), &b"before"[..]) + .write_frame(timed.then(|| Timestamp::from_millis(1000).unwrap()), &b"before"[..]) .unwrap(); before.finish().unwrap(); subscriber.subscribes.lock().insert( @@ -2974,18 +2977,31 @@ mod tests { (res, track) } - /// A received timestamp may not be media time (lite-05/06 untimed tracks carry send - /// times), so a received group that rewinds is accepted at every version and the track - /// stays open. Only a local publisher's writes refuse a rewind. + /// Lite-05/06 timestamps may be send times, so a received frame is read past its + /// timestamp and lands untimed: nothing rewinds, and the track stays open. #[moq_clock::test] - async fn a_received_rewind_is_accepted_at_every_version() { - for version in [Version::Lite05, Version::Lite06, Version::Lite07] { + async fn lite06_timestamps_are_dropped() { + for version in [Version::Lite05, Version::Lite06] { let (res, track) = receive_rewind(version).await; res.unwrap_or_else(|err| panic!("{version}: {err:?}")); assert!(track.closed().now_or_never().is_none(), "{version}"); + let mut consumer = track.subscribe(None); + let mut group = consumer.recv_group().now_or_never().unwrap().unwrap().unwrap(); + assert_eq!(group.sequence, 5, "{version}"); + let frame = group.read_frame().now_or_never().unwrap().unwrap().unwrap(); + assert_eq!(frame.payload.as_ref(), b"rewound", "{version}"); + assert_eq!(frame.timestamp, None, "{version}"); } } + /// Lite-07 timestamps are media time, so a received group is checked like a local + /// write and one that rewinds is refused. + #[moq_clock::test] + async fn a_lite07_rewind_is_refused() { + let (res, _track) = receive_rewind(Version::Lite07).await; + assert!(matches!(res, Err(Error::TimestampMismatch)), "{res:?}"); + } + /// An idle copy that resubscribes gets its newest cached group again. That stream only /// duplicates the cache, but lite-07's SUBSCRIBE_END still counts it, so the tail must /// too, or the subscription waits out the grace before it ends. @@ -3389,8 +3405,9 @@ mod tests { let broadcast = crate::broadcast::Info::new().produce(); // The broadcast keeps only a weak handle, so the map below owns the only strong // `track::Producer`: dropping it is what ends the track. + // Untimed, as the subscriber drops lite-05 timestamps. let producer = broadcast - .create_track("datagrams", crate::track::Info::timed()) + .create_track("datagrams", crate::track::Info::default()) .unwrap(); let mut received = producer.subscribe(None); subscriber.subscribes.lock().insert( @@ -6008,7 +6025,7 @@ impl TrackServe { id, TrackEntry { producer: producer.clone(), - framing: producer.info().into(), + framing: lite::Framing::new(producer.info(), self.subscriber.version), tail: tail.clone(), }, ); @@ -6338,7 +6355,12 @@ impl kio::Task for TrackServeRun { let res = ready!(info.poll_fetch(&self.serve, waiter)); let request = request.take().expect("request pending"); match res { - Ok((info, held)) => { + Ok((mut info, held)) => { + // Lite05/06 timestamps may be send times, so the copy is untimed + // and the ingest reads past them (see `lite::Framing::new`). + if !self.serve.subscriber.version.has_media_time() { + info.timescale = None; + } let mut serve_loop = ServeLoop::new(&self.serve, request, info); serve_loop.held = Some(held); self.state = TrackRunState::Serve(serve_loop); @@ -6507,12 +6529,7 @@ impl TrackInfoFetch { // window matches what the upstream advertises (relays re-serve with // the same bound). `broadcast` is left at its default here; // `track::Request::accept` stamps the track's real broadcast. - let model = track::Info::default() - .with_timescale(info.timescale) - .with_max_age(info.max_age) - .with_priority(info.priority) - .with_refresh(info.refresh) - .with_compress(info.compress); + let model = track::Info::from(info); if !ietf { return Poll::Ready(Ok((model, HeldTrack(stream)))); } @@ -6779,7 +6796,6 @@ impl ServeLoop { false => request, }; let serving = request.accept(info); - serving.received(); Self { largest: None, demand: serving.demand(), @@ -7388,7 +7404,7 @@ impl kio::Task for FetchServeRun { joined, ingest: Box::new(FrameIngest::new( &self.serve.subscriber, - (&self.info).into(), + lite::Framing::new(&self.info, self.serve.subscriber.version), ietf, frame_start, )), diff --git a/rs/moq-net/src/lite/track.rs b/rs/moq-net/src/lite/track.rs index ab302455770..658f248991c 100644 --- a/rs/moq-net/src/lite/track.rs +++ b/rs/moq-net/src/lite/track.rs @@ -76,6 +76,32 @@ pub struct TrackInfo { pub compress: bool, } +impl TrackInfo { + /// The TRACK_INFO a publisher serves for `info` on `version`. + pub(crate) fn serve(info: &crate::track::Info, version: Version) -> Self { + Self { + refresh: info.refresh, + priority: info.priority, + max_age: info.max_age, + // Unchecked timestamps never enter moq-lite, whose receivers check them. + timescale: info.lite_timescale(), + // An older peer is served the frames inflated. + compress: info.compress && version.has_compression(), + } + } +} + +impl From for crate::track::Info { + fn from(info: TrackInfo) -> Self { + Self::default() + .with_timescale(info.timescale) + .with_max_age(info.max_age) + .with_priority(info.priority) + .with_refresh(info.refresh) + .with_compress(info.compress) + } +} + /// TRACK_INFO `Compression`: no codec. const COMPRESSION_NONE: u64 = 0; /// TRACK_INFO `Compression`: per-group DEFLATE (draft-lcurley-moq-flate). @@ -85,7 +111,7 @@ const COMPRESSION_DEFLATE: u64 = 1; /// `ietf` extension, TRACK_PROPERTIES. #[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] pub(crate) struct Framing { - /// The units of every frame Timestamp, or `None` on an untimed track. + /// The units of every frame Timestamp on the wire, or `None` when frames carry none. pub timescale: Option, /// The draft the track's moq-transport publisher spoke, which serialized its objects' /// properties, or `None` for a moq-lite one, which carries no moq-transport state. @@ -94,10 +120,12 @@ pub(crate) struct Framing { pub compressed: bool, } -impl From<&crate::track::Info> for Framing { - fn from(info: &crate::track::Info) -> Self { +impl Framing { + /// How `version` frames a track accepted as `info`. Lite05/06 stamp every frame, so a + /// track received untimed there still has timestamps to read past. + pub fn new(info: &crate::track::Info, version: Version) -> Self { Self { - timescale: info.timescale, + timescale: version.wire_timescale(info.timescale), ietf: info.ietf(), compressed: info.compress, } @@ -288,15 +316,8 @@ mod test { #[test] fn track_info_defaults_match_cross_language_wire_bytes() { - let info = crate::track::Info::default(); - let info = TrackInfo { - refresh: None, - priority: info.priority, - max_age: info.max_age, - // The default track is untimed, which lite-05 declares as milliseconds. - timescale: info.timescale, - compress: false, - }; + // The default track is untimed, which lite-05 declares as milliseconds. + let info = TrackInfo::serve(&crate::track::Info::default(), Version::Lite05); let mut buf = Vec::new(); info.encode(&mut Encoder::new(&mut buf, Version::Lite05.into()), Version::Lite05) .unwrap(); diff --git a/rs/moq-net/src/lite/version.rs b/rs/moq-net/src/lite/version.rs index bd2816958ee..d75f56d689a 100644 --- a/rs/moq-net/src/lite/version.rs +++ b/rs/moq-net/src/lite/version.rs @@ -345,6 +345,18 @@ impl Version { } } + /// Whether received frame timestamps are the publisher's media time. Lite05/06 stamp an + /// untimed track's payloads with their send time, which a receiver cannot tell from it, + /// so it drops every timestamp received on them and the track arrives untimed. + #[allow(clippy::match_like_matches_macro)] + pub(crate) fn has_media_time(self) -> bool { + // Match form so future versions default forward (AGENTS.md convention). + match self { + Self::Lite01 | Self::Lite02 | Self::Lite03 | Self::Lite04 | Self::Lite05 | Self::Lite06 => false, + _ => true, + } + } + /// Whether discovery streams end their initial set with SYNCED rather than a count. #[allow(clippy::match_like_matches_macro)] pub fn has_synced(self) -> bool { diff --git a/rs/moq-net/src/model/group.rs b/rs/moq-net/src/model/group.rs index fbc0b286ba5..7c1bd5b3b7b 100644 --- a/rs/moq-net/src/model/group.rs +++ b/rs/moq-net/src/model/group.rs @@ -2754,6 +2754,14 @@ impl Consumer { self.track.timescale } + /// The timescale a moq-lite hop serves the parent track at (see + /// [`track::Info::lite_timescale`]). + #[cfg_attr(not(feature = "async"), allow(dead_code))] + // private: passthrough + pub(crate) fn lite_timescale(&self) -> Option { + self.track.lite_timescale() + } + /// The draft the parent track's moq-transport publisher spoke, which serialized every /// object's properties, or `None` for a moq-lite one. #[cfg(feature = "async")] diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index 6edbe786e2f..e045d991bfe 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -167,6 +167,14 @@ pub struct Info { #[cfg_attr(not(feature = "async"), allow(dead_code))] // private: passthrough pub(crate) properties: Option>, + /// Whether this timed track's timestamps arrived over moq-transport, whose peers may + /// stamp arrival times, so no group is checked for a rewind. + /// + /// They stay on moq-transport: a moq-lite hop serves the track untimed, since its + /// receivers check every timestamp. Never set on an untimed track. + #[cfg_attr(not(feature = "async"), allow(dead_code))] + // pending: /quest/m1/public-model/ingest.md + pub(crate) unchecked: bool, } impl Default for Info { @@ -178,6 +186,7 @@ impl Default for Info { compress: false, refresh: None, properties: None, + unchecked: false, } } } @@ -217,6 +226,14 @@ impl Info { self } + /// The timescale a moq-lite hop serves this track at: none for an + /// [unchecked](Self::unchecked) one. + #[cfg_attr(not(feature = "async"), allow(dead_code))] + // private: passthrough + pub(crate) fn lite_timescale(&self) -> Option { + self.timescale.filter(|_| !self.unchecked) + } + /// The draft the track's moq-transport publisher spoke, `None` for a moq-lite one. // private: passthrough pub(crate) fn ietf(&self) -> Option { @@ -434,9 +451,6 @@ pub(crate) struct TrackState { // [`Subscriber::poll_start`]) wait on it; everything else treats the floor as usual. start_pending: bool, - // A session's copy of a remote track. Its groups skip the rewind check: a received - // timestamp may not be media time (a lite-05/06 untimed track carries send times). - received: bool, // Sequences the upstream publisher gave up, with the error each was given up with: an // incoming SUBSCRIBE_DROP or a producer cut. Kept apart from the cache so reclaiming a // slot cannot erase a disposition a relay must forward, and as ranges so a wide DROP @@ -831,9 +845,9 @@ impl TrackState { /// so the watermark caps it instead. /// /// Usually one group lookup each way; only unstarted or partial neighbors are skipped. - /// `None` on a received track, which only a local publisher's writes are checked on. + /// `None` on an [unchecked](Info::unchecked) track, whose timestamps may rewind. pub(super) fn neighbors(&self, sequence: u64) -> Option { - if self.received { + if self.info.as_ref().is_some_and(|info| info.unchecked) { return None; } let bounds = self.bounds.lock().unwrap_or_else(PoisonError::into_inner); @@ -2963,16 +2977,6 @@ impl Producer { Ok(()) } - /// Mark this track as a session's copy of a remote one: its groups are never checked - /// for a rewind, which only a local publisher's writes refuse. - #[cfg_attr(not(feature = "async"), allow(dead_code))] - // pending: /quest/m1/public-model/ingest.md - pub(crate) fn received(&self) { - if let Ok(mut state) = self.modify() { - state.received = true; - } - } - /// Record sequences the upstream publisher explicitly gave up with `err`, ending the /// groups still arriving in them with it. #[cfg_attr(not(feature = "async"), allow(dead_code))] diff --git a/rs/moq-net/tests/integration/datagram.rs b/rs/moq-net/tests/integration/datagram.rs index ccce0f587a0..2b373c9c6f6 100644 --- a/rs/moq-net/tests/integration/datagram.rs +++ b/rs/moq-net/tests/integration/datagram.rs @@ -48,7 +48,8 @@ async fn connect_datagram_track() -> Fixture { let producer = broadcast.create_track("datagrams", timed()).unwrap(); broadcast.announce(Default::default()).unwrap(); - let mut options = MockConnectOptions::new("moq-lite-05".parse::().unwrap()); + // Lite-07, since a lite-05/06 subscriber drops every timestamp it receives. + let mut options = MockConnectOptions::new("moq-lite-07-wip".parse::().unwrap()); options.server_publish = Some(publisher.consume()); options.client_subscribe = Some(consumer_origin.clone()); let pair = connect_mock(options).await; diff --git a/rs/moq-net/tests/integration/received_rewind.rs b/rs/moq-net/tests/integration/received_rewind.rs index 5cc999f7143..913d91a3de4 100644 --- a/rs/moq-net/tests/integration/received_rewind.rs +++ b/rs/moq-net/tests/integration/received_rewind.rs @@ -1,11 +1,11 @@ #![cfg(feature = "async")] -//! A relay re-serves what it received without checking it for a rewind. +//! A relay drops the timestamps it receives over lite-05/06. //! //! Lite-06 stamps an untimed track's frames with their send time and declares -//! milliseconds, so the relay's copy reads as timed and re-advertises that scale. A group -//! sent after a higher one carries the later stamp, which the next hop must not refuse: -//! only a local publisher's writes are checked. +//! milliseconds, so a group sent after a higher one carries the later stamp. The relay +//! cannot tell that from media time, so its copy is untimed and every hop past it sees an +//! untimed track that nothing can refuse. use crate::support; @@ -28,21 +28,32 @@ fn produce_origin(hop: u64) -> moq_net::origin::Producer { } fn write_group(track: &moq_net::track::Producer, sequence: u64) { + // A timed track's group starts at its sequence in seconds, so none rewinds. + let timestamp = track + .info() + .timescale + .map(|_| moq_net::Timestamp::from_millis(sequence * 1000).unwrap()); let mut group = track.create_group(moq_net::group::Info { sequence }).unwrap(); - group.write_frame(None, &b"x"[..]).unwrap(); + group.write_frame(timestamp, &b"x"[..]).unwrap(); group.finish().unwrap(); } -/// Groups 1 then 0 of an untimed track, into a relay over lite-06 and out over `downstream`. -/// Returns the scale the subscriber learned and the groups it read. -async fn relay(downstream: &str) -> (Option, Vec) { +/// Groups 1 then 0 of a track at `timescale`, into a relay over `upstream` and out over +/// `downstream`. Returns the scale the subscriber learned and the groups it read. +async fn relay( + upstream: &str, + downstream: &str, + timescale: Option, +) -> (Option, Vec) { let publisher = produce_origin(1); let broadcast = publisher.create_broadcast("bcast").unwrap(); - let track = broadcast.create_track("data", Info::default()).unwrap(); + let track = broadcast + .create_track("data", Info::default().with_timescale(timescale)) + .unwrap(); broadcast.announce(Default::default()).unwrap(); let relay_origin = produce_origin(2); - let mut options = MockConnectOptions::new("moq-lite-06".parse::().unwrap()); + let mut options = MockConnectOptions::new(upstream.parse::().unwrap()); options.server_publish = Some(publisher.consume()); options.client_subscribe = Some(relay_origin.clone()); let upstream = connect_mock(options).await; @@ -114,9 +125,27 @@ async fn relay(downstream: &str) -> (Option, Vec) { async fn a_relay_forwards_descending_send_times_without_closing_the_track() { let mut failures = Vec::new(); for downstream in ["moq-lite-06", "moq-lite-07-wip", "moq-transport-22"] { - let got = relay(downstream).await; - // The relay's copy reads as timed, so every hop past it sees a timescale. - if got.0.is_none() || got.1 != [1, 0] { + let got = relay("moq-lite-06", downstream, None).await; + if got.0.is_some() || got.1 != [1, 0] { + failures.push(format!("{downstream}: {got:?}")); + } + } + assert!(failures.is_empty(), "{failures:#?}"); +} + +/// A moq-transport peer may stamp arrival times, so a relay keeps a timed track's timestamps +/// unchecked and passes them on only over moq-transport: a moq-lite hop, which checks every +/// timestamp it receives, is served the track untimed. +#[moq_clock::test] +async fn moq_transport_timestamps_stay_on_moq_transport() { + let mut failures = Vec::new(); + for (downstream, expected) in [ + ("moq-lite-06", None), + ("moq-lite-07-wip", None), + ("moq-transport-22", Some(moq_net::Timescale::MILLI)), + ] { + let got = relay("moq-transport-22", downstream, Some(moq_net::Timescale::MILLI)).await; + if got.0 != expected || got.1 != [1, 0] { failures.push(format!("{downstream}: {got:?}")); } } @@ -125,7 +154,7 @@ async fn a_relay_forwards_descending_send_times_without_closing_the_track() { /// Fetch groups 1 then 0 of a finished untimed track over `version`, with no live /// subscription. Each answer carries its send time, so group 0 lands after group 1; the -/// copy the fetches fill is still received, so neither is refused. +/// copy the fetches fill drops them, so neither is refused. async fn fetch_only(version: &str) -> Vec> { let publisher = produce_origin(1); let broadcast = publisher.create_broadcast("bcast").unwrap(); diff --git a/rs/moq-net/tests/integration/untimed.rs b/rs/moq-net/tests/integration/untimed.rs index fce1facdfa8..6338e5adb6d 100644 --- a/rs/moq-net/tests/integration/untimed.rs +++ b/rs/moq-net/tests/integration/untimed.rs @@ -4,8 +4,8 @@ //! that can say so, and a timed track keeps its timestamps. //! //! 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. +//! encoder writes its send time instead, and their receivers drop every timestamp, which +//! leaves a timed track untimed too. use crate::support; @@ -27,16 +27,14 @@ 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 and lite-06). - SendTime, - /// The wire declares no timescale, so even the timed track arrives untimed. + /// The wire carries no trustworthy timescale, so even the timed track arrives untimed. Untimed, } fn expected(version: &str) -> Arrives { match version { - "moq-lite-03" | "moq-lite-04" => Arrives::Untimed, - "moq-lite-05" | "moq-lite-06" => Arrives::SendTime, + // Lite-05 and lite-06 stamp untimed tracks with send times, so their receivers drop them. + "moq-lite-03" | "moq-lite-04" | "moq-lite-05" | "moq-lite-06" => Arrives::Untimed, // Drafts 14 to 16 can't carry TIMESCALE. "moq-transport-14" | "moq-transport-16" => Arrives::Untimed, _ => Arrives::Faithful, @@ -136,10 +134,6 @@ async fn frames_arrive_as_published() { assert_eq!(untimed, None, "{version}: absence survives"); assert_eq!(untimed_info.timescale, None, "{version}: no timeline is declared"); } - Arrives::SendTime => { - assert_eq!(timed, Some(millis(7)), "{version}: the timestamp survives"); - assert!(untimed.is_some(), "{version}: a lite encoder writes its send time"); - } Arrives::Untimed => { assert_eq!(timed, None, "{version}: no units, so no timestamp"); assert_eq!(timed_info.timescale, None, "{version}: no timeline is declared"); @@ -175,9 +169,8 @@ async fn a_relay_forwards_an_untimed_track_untimed() { /// An explicit start on an untimed track is honored the same way on lite and /// 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 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 +/// The subscription asks for history the way a resume does, with a budget covering it. +/// 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. #[moq_clock::test] @@ -213,8 +206,8 @@ async fn an_explicit_start_holds_on_an_untimed_track() { } } -/// An untimed track's datagram arrives untimed, except with a send time on lite-05 and -/// lite-06. +/// An untimed track's datagram arrives untimed, though lite-05 and lite-06 carry a send +/// time that the receiver drops. #[moq_clock::test] async fn datagrams_arrive_as_published() { for version in ["moq-lite-05", "moq-lite-07-wip", "moq-transport-16", "moq-transport-20"] { @@ -235,8 +228,7 @@ async fn datagrams_arrive_as_published() { .expect("datagram timeout") .unwrap() .expect("a datagram"); - let stamps = version == "moq-lite-05"; - assert_eq!(datagram.timestamp.is_some(), stamps, "{version}"); + assert_eq!(datagram.timestamp, None, "{version}"); } } @@ -268,7 +260,6 @@ async fn a_fetched_untimed_frame_arrives_untimed() { let mut fetched = waiting.await.unwrap(); let frame = fetched.read_frame().await.unwrap().unwrap(); - let stamps = version == "moq-lite-06"; - assert_eq!(frame.timestamp.is_some(), stamps, "{version}"); + assert_eq!(frame.timestamp, None, "{version}"); } } diff --git a/rs/moq-net/tests/integration/wide_sequence.rs b/rs/moq-net/tests/integration/wide_sequence.rs index d1f626d950f..745329c806d 100644 --- a/rs/moq-net/tests/integration/wide_sequence.rs +++ b/rs/moq-net/tests/integration/wide_sequence.rs @@ -52,9 +52,9 @@ async fn largest_quic_sequence_reaches_the_subscriber() { assert_eq!(got.sequence, sequence, "{alpn} relayed={relayed}"); let frame = got.read_frame().await.unwrap().unwrap(); assert_eq!(frame.payload, b"ladder".as_ref()); - if alpn != "moq-lite-02" { - assert_eq!(frame.timestamp, Some(timestamp)); - } + // Lite-02 carries no timestamps, and a lite-06 subscriber drops them. + let timed = alpn == "moq-transport-19"; + assert_eq!(frame.timestamp, timed.then_some(timestamp), "{alpn} relayed={relayed}"); } } } diff --git a/rs/moq-relay/src/test_support.rs b/rs/moq-relay/src/test_support.rs index 7303f2355e4..419f5a6f9de 100644 --- a/rs/moq-relay/src/test_support.rs +++ b/rs/moq-relay/src/test_support.rs @@ -19,10 +19,15 @@ pub struct TestRelay { } /// Bind a public test relay on ephemeral loopback ports without TOML or port probes. +/// +/// It also accepts the work-in-progress moq-lite-07, for a client that opts in. pub async fn test_relay() -> anyhow::Result { let mut config = Config::default(); config.drain_timeout = std::time::Duration::ZERO; config.listen.bind = Some("127.0.0.1:0".parse()?); + config.listen.version = std::iter::once("moq-lite-07-wip".parse().map_err(anyhow::Error::msg)?) + .chain(moq_net::Versions::all().iter().copied()) + .collect(); config.listen.tls.generate = vec!["localhost".into()]; config.web.http.listen = Some("127.0.0.1:0".parse()?); config.auth.public = vec![moq_auth::Pattern::all()]; diff --git a/rs/moq-relay/tests/integration/connection.rs b/rs/moq-relay/tests/integration/connection.rs index 9dff056a5d4..45a0133c185 100644 --- a/rs/moq-relay/tests/integration/connection.rs +++ b/rs/moq-relay/tests/integration/connection.rs @@ -60,9 +60,9 @@ async fn largest_quic_sequence_reaches_the_subscriber() { assert_eq!(got.sequence, sequence, "{alpn}"); let frame = got.read_frame().await.unwrap().unwrap(); assert_eq!(frame.payload, b"ladder".as_ref()); - if alpn != "moq-lite-02" { - assert_eq!(frame.timestamp, Some(timestamp)); - } + // Lite-02 carries no timestamps, and a lite-06 subscriber drops them. + let timed = alpn == "moq-transport-19"; + assert_eq!(frame.timestamp, timed.then_some(timestamp), "{alpn}"); } } } diff --git a/rs/moq-tokio/tests/integration/broadcast.rs b/rs/moq-tokio/tests/integration/broadcast.rs index a9021e1cdd8..0411a770512 100644 --- a/rs/moq-tokio/tests/integration/broadcast.rs +++ b/rs/moq-tokio/tests/integration/broadcast.rs @@ -135,7 +135,9 @@ async fn broadcast_test(scheme: &str, client_version: Option<&str>, server_versi } /// Lite05 publisher↔subscriber round-trip exercising the per-frame timestamp -/// delta encoding, including negative deltas (B-frame ordering). +/// delta encoding, including negative deltas (B-frame ordering). The subscriber decodes +/// each delta to stay in step with the payloads, then drops it, since a lite-05 timestamp +/// may be a send time. async fn lite05_timestamp_roundtrip(scheme: &str) { use moq_tokio::moq_net::{Timescale, Timestamp}; @@ -235,12 +237,9 @@ async fn lite05_timestamp_roundtrip(scheme: &str) { .expect("next_frame failed") .expect("group closed prematurely"); - let ts = frame_sub.timestamp.expect("timed frame"); - assert_eq!(ts.scale(), Timescale::MICRO); - assert_eq!(ts.value(), expected_us); - - // Drain the payload so the stream advances to the next frame. - let _ = frame_sub.read_all().await; + assert_eq!(frame_sub.timestamp, None, "a lite-05 subscriber drops timestamps"); + let payload = frame_sub.read_all().await.expect("failed to read frame"); + assert_eq!(payload, bytes::Bytes::from(format!("frame@{expected_us}"))); } drop(connection); @@ -258,7 +257,7 @@ async fn broadcast_moq_lite_05_timestamps_webtransport() { /// Lite05 FETCH round-trip: retrieve a past group by sequence without holding a /// subscription, exercising the bare-FRAME fetch response and per-frame timestamp -/// decoding on the fetch stream. The track is also `compress`-hinted, so the +/// decoding on the fetch stream, whose timestamps the subscriber then drops. The track is also `compress`-hinted, so the /// fetched frames are Deflate-compressed (matching TRACK_INFO) and inflated by the /// subscriber, exercising fetch/TRACK_INFO codec consistency. async fn lite05_fetch_roundtrip(scheme: &str) { @@ -353,9 +352,7 @@ async fn lite05_fetch_roundtrip(scheme: &str) { .expect("next_frame failed") .expect("group closed prematurely"); - let ts = frame_sub.timestamp.expect("timed frame"); - assert_eq!(ts.scale(), Timescale::MICRO); - assert_eq!(ts.value(), expected_us); + assert_eq!(frame_sub.timestamp, None, "a lite-05 subscriber drops timestamps"); let payload = frame_sub.read_all().await.expect("failed to read frame"); assert_eq!(payload, bytes::Bytes::from(format!("frame@{expected_us}"))); @@ -643,13 +640,11 @@ async fn broadcast_moq_lite_05_fetch_during_subscribe_webtransport() { } /// Lite05 can't declare an untimed track, so a track that sets no timescale goes out -/// in milliseconds, its frames stamped with their send time. The subscriber receives -/// `Some(ts)` at that scale. +/// in milliseconds, its frames stamped with their send time. The subscriber drops them, +/// so the track arrives untimed. #[tracing_test::traced_test] #[tokio::test] async fn broadcast_moq_lite_05_untimed_track() { - use moq_tokio::moq_net::Timescale; - let pub_origin = moq_tokio::origin::spawn(); let broadcast = pub_origin.create_broadcast("test").expect("create broadcast"); broadcast.announce(Default::default()).expect("create broadcast"); @@ -724,12 +719,8 @@ async fn broadcast_moq_lite_05_untimed_track() { .expect("next_frame failed") .expect("group closed"); - let ts = frame_sub.timestamp.expect("timed frame"); - assert_eq!( - ts.scale(), - Timescale::MILLI, - "an untimed track's send times are milliseconds" - ); + assert_eq!(frame_sub.timestamp, None, "the send time is dropped"); + assert_eq!(track_sub.info().timescale, None, "the track arrives untimed"); drop(connection); server_handle diff --git a/test/wasm/baseline/consumer.ts b/test/wasm/baseline/consumer.ts index 0ff9b378e3d..1b30a668051 100644 --- a/test/wasm/baseline/consumer.ts +++ b/test/wasm/baseline/consumer.ts @@ -36,6 +36,8 @@ async function run(config: Sample) { let grew = false; let received = 0; const frames = config.publishers * config.subscribers * config.frames; + // Lite-02 carries no timestamps, and a lite-05/06 subscriber drops them. + const timed = !/^moq-lite-0[1-6]$/.test(config.version); const check = (bytes: Uint8Array, frame: number) => { first ??= performance.now() - start; verify(bytes, config.size, frame); @@ -90,12 +92,7 @@ async function run(config: Sample) { try { for (let frame = 0; frame < config.frames; frame++) { const got = await group.readFrame(); - if ( - !got || - (config.version === "moq-lite-02" - ? got.timestamp !== undefined - : got.timestamp?.asMillis() !== frame) - ) + if (!got || (timed ? got.timestamp?.asMillis() !== frame : got.timestamp !== undefined)) throw new Error(`incorrect JS frame timestamp ${frame}`); check(got.payload, frame); } @@ -178,9 +175,7 @@ async function run(config: Sample) { const got = await group.readFrame(); if ( !got || - (config.version === "moq-lite-02" - ? got.timestampUs !== undefined - : got.timestampUs !== BigInt(frame * 1000)) + (timed ? got.timestampUs !== BigInt(frame * 1000) : got.timestampUs !== undefined) ) throw new Error(`incorrect UniFFI frame timestamp ${frame}`); check(new Uint8Array(got.payload), frame); @@ -210,12 +205,11 @@ async function run(config: Sample) { }, metadata: { sequence: String(config.sequence), - timestamp: - config.version === "moq-lite-02" - ? "verified absent: this wire version carries none" - : config.backend === "wasm" - ? "unavailable: payload-only binding" - : "verified", + timestamp: !timed + ? "verified absent: this version delivers none" + : config.backend === "wasm" + ? "unavailable: payload-only binding" + : "verified", }, }; } finally { diff --git a/test/wasm/generator.md b/test/wasm/generator.md index 49b7523d757..09bf2338c47 100644 --- a/test/wasm/generator.md +++ b/test/wasm/generator.md @@ -179,7 +179,7 @@ three bindings. moq-wasm's column is the existing suite in [`src/main.ts`](src/m | reads a published broadcast, every byte, 128 KiB frames | yes | yes | yes | | refuses a missing track | yes | yes: rejects with `NotFound` (transport-19) or `Protocol` (lite) | yes: rejects with the slice's `Net { reason: "not found" }` | | publishes | no: consume only | yes | yes | -| frame timestamps | no: payload only | yes, exact, as `bigint` microseconds | yes | +| frame timestamps | no: payload only | yes, exact, as `bigint` microseconds on a version that delivers them (not lite-02 or lite-06) | yes | | u64 sequence above 2^53, end to end | not testable: no publisher can send it | yes, 2^53+1 on all three; 2^62-1 on lite-02 and transport-19 (see bugs) | same as moq-ffi | | cancellation through `AbortSignal` | no: `free()` rejects pending calls | rejects with `AbortError`; the next call still gets the next frame | UniFFI: `AbortError`; BoltFFI: `BoltFFICancelledError`; neither steals the next frame | | aborting releases the call's state | n/a | yes: a track's demand drops only once the aborted future is freed | not testable: the slice has no demand | diff --git a/test/wasm/uniffi/cases.ts b/test/wasm/uniffi/cases.ts index d6d0c5061ce..f50a476c159 100644 --- a/test/wasm/uniffi/cases.ts +++ b/test/wasm/uniffi/cases.ts @@ -267,9 +267,9 @@ const remote: Record Promise> = { const got = (await next) as MoqGroupConsumerLike | undefined; if (got?.sequence() !== sequence) throw new Error(`sequence ${got?.sequence()} != ${sequence}`); const frame = await got.readFrame(); - // moq-lite-02 carries no timestamp; every other version must keep it exact. - if (relay.version !== "moq-lite-02" && frame?.timestampUs !== sequence) - throw new Error(`timestamp ${frame?.timestampUs} != ${sequence}`); + // Lite-02 carries no timestamp and a lite-05/06 subscriber drops it; the rest keep it exact. + const expected = /^moq-lite-0[1-6]$/.test(relay.version) ? undefined : sequence; + if (frame?.timestampUs !== expected) throw new Error(`timestamp ${frame?.timestampUs} != ${expected}`); seen.push(String(sequence)); } return undefined;