diff --git a/drafts/draft-lcurley-moq-lite.md b/drafts/draft-lcurley-moq-lite.md index a51b44f89f7..b7a699a6552 100644 --- a/drafts/draft-lcurley-moq-lite.md +++ b/drafts/draft-lcurley-moq-lite.md @@ -174,7 +174,9 @@ A Group Stream normally starts at frame 0, but MAY start later when the publishe Each frame of a timed Track carries a presentation timestamp expressed in the Track's `Timescale` (see [TRACK_INFO](#track-info)), used by the moq-lite layer for [expiration](#expiration) decisions. A Track without a `Timescale` is untimed: its frames and datagrams carry no timestamp. +A publisher MUST NOT start a timed Group's frame 0 earlier than that of any lower-sequence Group it publishes. lite-05 and lite-06 cannot encode an untimed Track, so a publisher on those versions declares a `Timescale` of 1000 for it and stamps every frame and datagram with its send time in milliseconds. +A receiver cannot tell such send times from presentation timestamps, and they need not rise with sequence, so it does not check received Groups against this rule. ## Positions {#positions} A Position is a (Group Sequence, Frame Index) pair identifying one frame within a Track. @@ -1582,6 +1584,7 @@ The `Message Length` describes the payload size on the wire. ## moq-lite-07 +- A publisher must not start a timed Group's frame 0 earlier than a lower-sequence Group's. Receivers do not check it, since a received timestamp may be a send time. - Added the reliable ordered RTCDataChannel binding, with one Qmux Record per binary message and no datagrams. - Added `Group Order` to SUBSCRIBE and SUBSCRIBE_UPDATE: descending (0x0, the previous fixed order) or ascending (0x1). A relay requests descending upstream unless every subscription it serves is ascending. diff --git a/js/hang/src/container/consumer.test.ts b/js/hang/src/container/consumer.test.ts index ef4eea4e2c9..17e62dc553b 100644 --- a/js/hang/src/container/consumer.test.ts +++ b/js/hang/src/container/consumer.test.ts @@ -526,10 +526,16 @@ function encodeLegacy(timestamp: Micro): Uint8Array { return data; } -function writeGroupWithLegacyFrames(track: Track.Producer, sequence: number, timestamps: Micro[]) { +// `stamp` is the transport timestamp, the container's own by default. +function writeGroupWithLegacyFrames( + track: Track.Producer, + sequence: number, + timestamps: Micro[], + stamp: (ts: Micro) => Time.Timestamp = (ts) => Time.Timestamp.fromMicros(ts), +) { const group = new Group.Producer(sequence); for (const ts of timestamps) { - group.writeFrame({ payload: encodeLegacy(ts), timestamp: Time.Timestamp.now() }); + group.writeFrame({ payload: encodeLegacy(ts), timestamp: stamp(ts) }); } group.close(); track.writeGroup(group); @@ -953,12 +959,16 @@ async function nextFrame(consumer: Consumer) { } } +// The track refuses a transport timestamp that rewinds, so these rewind only the container's +// own timestamps, under wall-clock transport stamps that rise in write order. +const wallClock = () => Time.Timestamp.now(); + test("Consumer aborts a group below the previous group start", async () => { const track = new Track.Producer("test"); const consumer = new Consumer(replay(track), { format: new LegacyFormat("data"), maxDelay: 30_000 as Milli }); - writeGroupWithLegacyFrames(track, 0, [10_000_000 as Micro]); - writeGroupWithLegacyFrames(track, 1, [0 as Micro, 100_000 as Micro]); + writeGroupWithLegacyFrames(track, 0, [10_000_000 as Micro], wallClock); + writeGroupWithLegacyFrames(track, 1, [0 as Micro, 100_000 as Micro], wallClock); await settle(); expect((await nextFrame(consumer))?.frame?.timestamp).toBe(10_000_000 as Micro); @@ -971,13 +981,13 @@ test("Consumer aborts a rewind carried by a later arrival", async () => { const track = new Track.Producer("test"); const consumer = new Consumer(replay(track), { format: new LegacyFormat("data"), maxDelay: 30_000 as Milli }); - writeGroupWithLegacyFrames(track, 0, [10_000_000 as Micro]); + writeGroupWithLegacyFrames(track, 0, [10_000_000 as Micro], wallClock); await settle(); expect((await nextFrame(consumer))?.frame?.timestamp).toBe(10_000_000 as Micro); const pending = nextFrame(consumer); await settle(); - writeGroupWithLegacyFrames(track, 1, [0 as Micro, 100_000 as Micro]); + writeGroupWithLegacyFrames(track, 1, [0 as Micro, 100_000 as Micro], wallClock); await expect(pending).rejects.toThrow("below the previous group start"); @@ -1047,10 +1057,10 @@ test("Consumer reads no gap when retiring a group holding only its endpoint", as const consumer = new Consumer(replay(track), { format: new LegacyFormat("video"), maxDelay: 150 as Milli }); const group = new Group.Producer(0); - group.writeFrame({ payload: encodeLegacy(0 as Micro), timestamp: Time.Timestamp.now() }); + group.writeFrame({ payload: encodeLegacy(0 as Micro), timestamp: Time.Timestamp.fromMicros(0) }); group.writeFrame({ payload: encodeLegacyFrame(33_000 as Micro, new Uint8Array()), - timestamp: Time.Timestamp.now(), + timestamp: Time.Timestamp.fromMicros(33_000), }); group.close(); track.writeGroup(group); @@ -1073,7 +1083,7 @@ test("Consumer reads a gap when retiring a truncated group", async () => { const consumer = new Consumer(replay(track), { format: new LegacyFormat("audio"), maxDelay: 150 as Milli }); const group = new Group.Producer(0); - group.writeFrame({ payload: encodeLegacy(0 as Micro), timestamp: Time.Timestamp.now() }); + group.writeFrame({ payload: encodeLegacy(0 as Micro), timestamp: Time.Timestamp.fromMicros(0) }); track.writeGroup(group); await settle(); expect(frameOf(await consumer.next()).timestamp).toBe(0 as Micro); @@ -1096,7 +1106,7 @@ test("Consumer reads a gap when retiring an open group", async () => { const consumer = new Consumer(replay(track), { format: new LegacyFormat("audio"), maxDelay: 150 as Milli }); const group = new Group.Producer(0); - group.writeFrame({ payload: encodeLegacy(0 as Micro), timestamp: Time.Timestamp.now() }); + group.writeFrame({ payload: encodeLegacy(0 as Micro), timestamp: Time.Timestamp.fromMicros(0) }); track.writeGroup(group); await settle(); expect(frameOf(await consumer.next()).timestamp).toBe(0 as Micro); @@ -1723,7 +1733,7 @@ test("Consumer reads a shed marker group as a gap", async () => { track.writeGroup(group0); group0.writeFrame({ payload: encodeLegacy(0 as Micro), - timestamp: Time.Timestamp.now(), + timestamp: Time.Timestamp.fromMicros(0), }); await settle(); expect((await nextFrame(consumer))?.frame?.timestamp).toBe(0 as Micro); diff --git a/js/net/src/broadcast.test.ts b/js/net/src/broadcast.test.ts index d8c42680736..773fb35b94f 100644 --- a/js/net/src/broadcast.test.ts +++ b/js/net/src/broadcast.test.ts @@ -1,7 +1,7 @@ import { expect, setSystemTime, test } from "bun:test"; import { Milli } from "@moq/clock"; import { Consumer as BroadcastConsumer, Producer as BroadcastProducer } from "./broadcast.ts"; -import { GroupTooLarge, NotFound } from "./error.ts"; +import { GroupTooLarge, NotFound, TimestampMismatch } from "./error.ts"; import { Producer as GroupProducer, MAX_GROUP_FRAMES } from "./group.ts"; import { TrackSequences } from "./internal.ts"; import { textFrame } from "./mock.ts"; @@ -55,6 +55,29 @@ test("consumer dedupes repeat subscriptions onto one upstream request", async () expect((await pendingRequest(consumer))?.name).toBe("video"); }); +// A replacement continues the name's timeline, so its groups can't start before the highest one an +// earlier producer checked, while a new broadcast starts over. +test("a replacement producer refuses a rewind past its predecessor", () => { + const broadcast = new BroadcastProducer(); + const first = broadcast.createTrack("media", { timescale: Timescale.MILLI }); + first.appendGroup().writeFrame(textFrame("a", Timestamp.fromMillis(1000))); + first.close(); + + const second = broadcast.createTrack("media", { timescale: Timescale.MILLI }); + const group = new GroupProducer(1); + second.writeGroup(group); + expect(() => group.writeFrame(textFrame("b", Timestamp.fromMillis(500)))).toThrow(TimestampMismatch); + group.writeFrame(textFrame("b", Timestamp.fromMillis(1000))); + + const next = new BroadcastProducer(); + const track = next.createTrack("media", { timescale: Timescale.MILLI }); + track.appendGroup().writeFrame(textFrame("c", Timestamp.fromMillis(500))); + + second.close(); + broadcast.close(); + next.close(); +}); + // A name keeps one sequence namespace across producer replacements. The application owns where a // replacement starts: an append that would reuse an earlier producer's sequence throws. test("a replacement producer refuses to append a sequence the name already wrote", async () => { diff --git a/js/net/src/error.ts b/js/net/src/error.ts index 41dba873aba..66dce375505 100644 --- a/js/net/src/error.ts +++ b/js/net/src/error.ts @@ -257,8 +257,8 @@ export class GroupTooLarge extends Stream { } /** - * A frame's timedness doesn't match its track: a timestamp on a track with no timescale, or - * none on a track that has one. + * A frame's timestamp doesn't fit its track: a timestamp on a track with no timescale, none + * on a track that has one, or a group's first frame starting before the group below it. * * Raised locally by a frame or datagram write, and decoded from a moq-lite peer's * `TIMESTAMP_MISMATCH` reset. Mirrors the Rust `Error::TimestampMismatch`. @@ -269,7 +269,7 @@ export class TimestampMismatch extends Stream { constructor(options?: { cause?: unknown }) { super(StreamCode.TimestampMismatch, { ...options, - message: "frame timestamp doesn't match track timescale", + message: "frame timestamp doesn't fit its track", }); this.name = "TimestampMismatch"; } diff --git a/js/net/src/group.ts b/js/net/src/group.ts index 35182d54c34..2bdc8aaca71 100644 --- a/js/net/src/group.ts +++ b/js/net/src/group.ts @@ -69,6 +69,9 @@ class GroupState { timestamp?: Timestamp; // The newest frame's timestamp: where a reader that has taken every frame sits. latest?: Timestamp; + // Accepted below the track's eviction watermark, where nothing proves where it starts, + // so this mirror drives no media-time expiry. + unverified = false; // When the group joined the track this mirror belongs to, at this hop: the wall clock a // superseded predecessor ages from. Per track, since one group can join several. arrived = Instant.zero; @@ -159,6 +162,10 @@ export class Producer { // not from its creation, so a group still being filled is never reclaimed. #activity = now(); + // The tracks this group joined before its first frame, each checking that frame's + // timestamp against its neighbors' starts. + #admits?: ((start: Timestamp) => () => void)[]; + constructor(sequence: number) { this.#state = new GroupState(sequence); this.sequence = sequence; @@ -174,6 +181,11 @@ export class Producer { for (const frame of state.frames.peek()) checkTimed(timed, frame); state.timed = timed; }; + hooks.admitGroupStart = (group, admit) => { + group.#admits ??= []; + group.#admits.push(admit); + }; + hooks.groupStart = (group) => (group.#state.total.peek() > 0 ? group.#state.timestamp : undefined); hooks.limitGroup = (group, frames) => { // A closed group takes no more frames, but one already past the cap still // contradicts the end. @@ -266,6 +278,13 @@ export class Producer { if (this.#limit !== undefined && this.#state.total.peek() >= this.#limit) { throw new Error(`frame ${this.#state.total.peek()} is past the track's end`); } + // Frame 0 must fit between its neighbors in every track it joined, checked before + // any of them records it, so a refusal leaves the group and its tracks as they were. + let commits: (() => void)[] | undefined; + if (this.#admits && frame.timestamp !== undefined && this.#state.total.peek() === 0) { + const start = frame.timestamp; + commits = this.#admits.map((admit) => admit(start)); + } if ( this.#state.total.peek() >= MAX_GROUP_FRAMES || this.#state.cacheBytes + frame.payload.byteLength > MAX_GROUP_CACHE_BYTES @@ -285,6 +304,10 @@ export class Producer { this.#activity = now(); appendFrame(this.#state, frame); + if (commits) { + this.#admits = undefined; + for (const commit of commits) commit(); + } if (this.#mirrors) { for (const mirror of this.#mirrors) { @@ -431,8 +454,11 @@ export class Consumer { static { makeConsumer = (state) => new Consumer(state); - hooks.groupTimestamp = (group) => group.#state.timestamp; - hooks.groupLatest = (group) => group.#state.latest; + hooks.groupTimestamp = (group) => (group.#state.unverified ? undefined : group.#state.timestamp); + hooks.groupLatest = (group) => (group.#state.unverified ? undefined : group.#state.latest); + hooks.unverifyGroup = (group) => { + group.#state.unverified = true; + }; hooks.groupArrived = (group) => group.#state.arrived; hooks.arriveGroup = (group, at) => { group.#state.arrived = at; diff --git a/js/net/src/ietf/subscriber.ts b/js/net/src/ietf/subscriber.ts index 82f37956c48..cb550b62b19 100644 --- a/js/net/src/ietf/subscriber.ts +++ b/js/net/src/ietf/subscriber.ts @@ -663,6 +663,7 @@ export class Subscriber { // Keep the request pending until SUBSCRIBE_OK supplies immutable track metadata. // Group streams already wait on the alias, so early data stays behind this response. const producer = hooks.pendingTrackProducer(request); + hooks.receiveTrack(producer); const subscription: Subscription = { track: producer, tail: new Tail() }; // Open the stream and wait for SUBSCRIBE_OK under a timeout. State diff --git a/js/net/src/internal.ts b/js/net/src/internal.ts index 9f5fee77ee6..48c006f0cc4 100644 --- a/js/net/src/internal.ts +++ b/js/net/src/internal.ts @@ -174,6 +174,11 @@ export interface TrackSequence { next: number; /** Open producers writing through this namespace. */ holders: number; + /** + * The highest-sequence checked group start any producer of the name wrote, which bounds a + * replacement's groups the way the eviction watermark bounds a track's own. + */ + bound?: { sequence: number; start: Timestamp }; } /** @@ -239,6 +244,12 @@ export const hooks: { * any other second end. Mirrors the Rust `finish_at_pending`; assigned by `track.ts`. */ endTrack: (producer: Producer, end: Location) => void; + // pending: /quest/m1/public-model/ingest.md + /** + * Mark a track as a session's copy of a remote one: its groups are never checked for a + * rewind, since a received timestamp may not be media time. + */ + receiveTrack: (producer: Producer) => void; /** Mint a track {@link Request}; assigned by `track.ts`. */ makeRequest: (options: TrackRequestOptions) => Request; // pending: /quest/m1/public-model/ingest.md @@ -282,6 +293,16 @@ export const hooks: { * Throws `ProtocolViolation` if a frame at or past the cap already exists. */ limitGroup: (group: GroupProducer, frames: number) => void; + /** + * Check a group's first frame against a track it joined before the frame lands. `admit` + * throws `TimestampMismatch` on a rewind, or returns the commit to run once every track + * the group joined accepted it. + */ + admitGroupStart: (group: GroupProducer, admit: (start: Timestamp) => () => void) => void; + /** Return a group producer's first timestamp, if it already holds a frame. */ + groupStart: (group: GroupProducer) => Timestamp | undefined; + /** Keep a mirror of a group accepted unverified below its track's eviction watermark out of media-time expiry. */ + unverifyGroup: (group: GroupConsumer) => void; /** Return a group's first timestamp, retained even after its first frame is read. */ groupTimestamp: (group: GroupConsumer) => Timestamp | undefined; groupLatest: (group: GroupConsumer) => Timestamp | undefined; @@ -315,6 +336,9 @@ export const hooks: { /** Name the epoch of the route an origin resolved a broadcast handle through. */ stampEpoch: (target: BroadcastConsumer, epoch: Epoch.Valid | undefined) => void; } = { + receiveTrack: () => { + throw new Error("track.ts not loaded"); + }, endTrack: () => { throw new Error("track.ts not loaded"); }, @@ -351,6 +375,15 @@ export const hooks: { limitGroup: () => { throw new Error("group.ts not loaded"); }, + admitGroupStart: () => { + throw new Error("group.ts not loaded"); + }, + groupStart: () => { + throw new Error("group.ts not loaded"); + }, + unverifyGroup: () => { + throw new Error("group.ts not loaded"); + }, arriveGroup: () => { throw new Error("group.ts not loaded"); }, diff --git a/js/net/src/lite/publisher.test.ts b/js/net/src/lite/publisher.test.ts index 3a26d380ec9..c6455a316a1 100644 --- a/js/net/src/lite/publisher.test.ts +++ b/js/net/src/lite/publisher.test.ts @@ -205,7 +205,7 @@ async function subscribeEnd(sequences: number[], version: Version = Version.DRAF // track ending rather than resolving a subscribe against an already-closed one. for (const sequence of sequences) { const group = new GroupProducer(sequence); - group.writeFrame(textFrame("hello")); + group.writeFrame(textFrame("hello", Timestamp.fromMillis(sequence))); group.close(); track.writeGroup(group); @@ -641,9 +641,10 @@ async function servedSubscription( serving, parked: gate.parked, release: gate.release, + // Stamped by sequence, so a group served out of order still starts in order. serve(sequence: number) { const group = new GroupProducer(sequence); - for (const frame of frames) group.writeFrame(textFrame(frame)); + for (const frame of frames) group.writeFrame(textFrame(frame, Timestamp.fromMillis(sequence))); group.close(); track.writeGroup(group); }, diff --git a/js/net/src/lite/subscriber.test.ts b/js/net/src/lite/subscriber.test.ts index 8789e8fb2f9..5f9bb6d055e 100644 --- a/js/net/src/lite/subscriber.test.ts +++ b/js/net/src/lite/subscriber.test.ts @@ -941,13 +941,17 @@ function fakeSession(park: number[] = [], stall: number[] = []) { return { quic, streams, opened, closes }; } -async function answerTrackInfo(stream: FakeStream, version: Version = Version.DRAFT_05): Promise { +async function answerTrackInfo( + stream: FakeStream, + version: Version = Version.DRAFT_05, + timescale?: number, +): Promise { const chunks: Uint8Array[] = []; const writer = new Writer( new WritableStream({ write: (chunk) => void chunks.push(new Uint8Array(chunk)) }), version, ); - await new TrackInfo({}).encode(writer, version); + await new TrackInfo({ timescale }).encode(writer, version); for (const chunk of chunks) stream.inbound.enqueue(chunk); stream.inbound.close(); } @@ -1690,6 +1694,67 @@ test("a local unsubscribe owes no SUBSCRIBE_END", async () => { } }); +// A group stream carrying one frame at `timestamp` (zigzag delta from 0) in the track's scale. +async function groupStream(timestamp: number, version: Version = Version.DRAFT_07): Promise { + const chunks: Uint8Array[] = []; + const writer = new Writer( + new WritableStream({ write: (chunk) => void chunks.push(new Uint8Array(chunk)) }), + version, + ); + await writer.u62(BigInt(timestamp) * 2n); + await writer.u53(1); + await writer.write(new Uint8Array([0])); + return new Reader( + new ReadableStream({ + start: (controller) => { + for (const chunk of chunks) controller.enqueue(chunk); + controller.close(); + }, + }), + undefined, + version, + ); +} + +// 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(); + } + }, +); + test("a publisher FIN without SUBSCRIBE_END still violates an open subscription", async () => { const { quic, streams, opened, closes } = fakeSession(); const subscriber = new Subscriber(quic, Version.DRAFT_07, HopSchema.parse(1n)); diff --git a/js/net/src/lite/subscriber.ts b/js/net/src/lite/subscriber.ts index c43c88f4b01..c33da2df61d 100644 --- a/js/net/src/lite/subscriber.ts +++ b/js/net/src/lite/subscriber.ts @@ -831,6 +831,7 @@ export class Subscriber { timescale.set(0); drainOk = true; } + hooks.receiveTrack(producer); // Register before opening SUBSCRIBE so a racing GROUP stream finds the entry. const entry: SubscribeEntry = { diff --git a/js/net/src/mock.ts b/js/net/src/mock.ts index d817e695c1b..60fb5da9ed1 100644 --- a/js/net/src/mock.ts +++ b/js/net/src/mock.ts @@ -9,9 +9,9 @@ import type { Frame } from "./group.ts"; import type { SendStream } from "./stream.ts"; import { Timestamp } from "./time.ts"; -/** A UTF-8 text frame stamped now, for a test writing to a timed track. */ -export function textFrame(text: string): Frame { - return { payload: new TextEncoder().encode(text), timestamp: Timestamp.now() }; +/** A UTF-8 text frame stamped `timestamp` (now by default), for a test writing to a timed track. */ +export function textFrame(text: string, timestamp = Timestamp.now()): Frame { + return { payload: new TextEncoder().encode(text), timestamp }; } // High watermark to prevent writes from blocking on backpressure. diff --git a/js/net/src/track.test.ts b/js/net/src/track.test.ts index 6219f943f51..56764241ddf 100644 --- a/js/net/src/track.test.ts +++ b/js/net/src/track.test.ts @@ -619,18 +619,18 @@ test("ordered frame reads skip a late lower-sequence group", async () => { const track = producer.subscribe({ maxDelay: Milli(5000) }).ordered(); const five = new GroupProducer(5); - five.writeFrame(textFrame("five")); + five.writeFrame(textFrame("five", Timestamp.fromMillis(5))); five.close(); producer.writeGroup(five); expect(await track.readString()).toBe("five"); const three = new GroupProducer(3); - three.writeFrame(textFrame("three")); + three.writeFrame(textFrame("three", Timestamp.fromMillis(3))); three.close(); producer.writeGroup(three); const six = new GroupProducer(6); - six.writeFrame(textFrame("six")); + six.writeFrame(textFrame("six", Timestamp.fromMillis(6))); six.close(); producer.writeGroup(six); @@ -668,9 +668,9 @@ test("latency budget admits groups within its presentation-time window", async ( test("a late lower group within the budget is delivered", async () => { const producer = new TrackProducer("test").accept({ timescale: Timescale.MILLI, maxAge: Milli(30_000) }); for (const [sequence, timestamp] of [ - [5, 0], - [6, 1000], - [7, 2000], + [5, 1000], + [6, 2000], + [7, 3000], ]) { const group = new GroupProducer(sequence); producer.writeGroup(group); @@ -2506,3 +2506,129 @@ test.each([true, false])("a mid-group end caps later writes (declared first: %s) expect(() => producer.writeGroup(overfull)).toThrow(ProtocolViolation); producer.close(); }); + +// A closed group `sequence` written into `producer` whose one frame starts at `ms`, aborted +// as ingest does when that frame is refused. +function stampGroup(producer: TrackProducer, sequence: number, ms: number): GroupProducer { + const group = new GroupProducer(sequence); + producer.writeGroup(group); + try { + group.writeFrame({ payload: enc.encode(`${ms}`), timestamp: Timestamp.fromMillis(ms) }); + } catch (err) { + group.close(err as Error); + throw err; + } + group.close(); + return group; +} + +test("an appended rewind is refused and an equal start is not", () => { + const producer = new TrackProducer("test").accept({ timescale: Timescale.MILLI }); + producer.writeFrame({ payload: enc.encode("a"), timestamp: Timestamp.fromMillis(1000) }); + producer.writeFrame({ payload: enc.encode("b"), timestamp: Timestamp.fromMillis(1000) }); + expect(() => producer.writeFrame({ payload: enc.encode("c"), timestamp: Timestamp.fromMillis(999) })).toThrow( + TimestampMismatch, + ); + producer.writeFrame({ payload: enc.encode("d"), timestamp: Timestamp.fromMillis(1001) }); +}); + +test("a rewind is refused whichever group lands first", () => { + const producer = new TrackProducer("test").accept({ timescale: Timescale.MILLI }); + stampGroup(producer, 5, 1000); + expect(() => stampGroup(producer, 3, 2000)).toThrow(TimestampMismatch); + stampGroup(producer, 3, 500); + expect(() => stampGroup(producer, 4, 400)).toThrow(TimestampMismatch); + expect(() => stampGroup(producer, 4, 1100)).toThrow(TimestampMismatch); + stampGroup(producer, 4, 700); +}); + +// Subscribers may have seen an aborted group's start, so a replacement at its sequence leaves +// that start as the watermark, and no later group can start before it. +test("a replaced aborted group still bounds later ones", () => { + const producer = new TrackProducer("test").accept({ timescale: Timescale.MILLI }); + const first = new GroupProducer(0); + producer.writeGroup(first); + first.writeFrame({ payload: enc.encode("1000"), timestamp: Timestamp.fromMillis(1000) }); + first.close(new Error("aborted")); + + const replacement = new GroupProducer(0); + replacement.writeFrame({ payload: enc.encode("500"), timestamp: Timestamp.fromMillis(500) }); + producer.writeGroup(replacement); + replacement.close(); + expect(() => stampGroup(producer, 1, 600)).toThrow(TimestampMismatch); + stampGroup(producer, 1, 1000); +}); + +test("a rewind across large timescales is refused", () => { + const m = Number.MAX_SAFE_INTEGER; + const producer = new TrackProducer("test").accept({ timescale: Timescale.MILLI }); + producer.writeFrame({ payload: enc.encode("a"), timestamp: new Timestamp(m - 1, Timescale(m)) }); + // (m - 2) / (m - 1) is earlier than (m - 1) / m, though both cross-products round alike. + expect(() => + producer.writeFrame({ payload: enc.encode("b"), timestamp: new Timestamp(m - 2, Timescale(m - 1)) }), + ).toThrow(TimestampMismatch); + // Exact for fractions too: 0.75s is later than 0.5s. + const fractions = new TrackProducer("test").accept({ timescale: Timescale.MILLI }); + fractions.writeFrame({ payload: enc.encode("a"), timestamp: new Timestamp(0.5, Timescale.SECOND) }); + fractions.writeFrame({ payload: enc.encode("b"), timestamp: new Timestamp(750.5, Timescale.MILLI) }); + expect(() => + fractions.writeFrame({ payload: enc.encode("c"), timestamp: new Timestamp(1.5, Timescale(4)) }), + ).toThrow(TimestampMismatch); +}); + +test("a group stamped before it is written is checked on write", () => { + const producer = new TrackProducer("test").accept({ timescale: Timescale.MILLI }); + stampGroup(producer, 1, 6000); + const group = new GroupProducer(2); + group.writeFrame({ payload: enc.encode("early"), timestamp: Timestamp.fromMillis(5000) }); + expect(() => producer.writeGroup(group)).toThrow(TimestampMismatch); +}); + +test("an evicted group still bounds, and a late group below it is unverified", async () => { + const clock = manualTime(10_000); + try { + const producer = new TrackProducer("test").accept({ timescale: Timescale.MILLI, maxAge: Milli(60_000) }); + for (const [sequence, ms] of [ + [0, 1000], + [1, 2000], + [2, 3000], + ]) { + stampGroup(producer, sequence, ms); + } + + // Idle past the cache window, so the next write evicts every group but the new edge. + clock.set(10_000 + CACHE_WINDOW_MS + 200); + const edge = producer.appendGroup(); + expect(() => edge.writeFrame({ payload: enc.encode("rewind"), timestamp: Timestamp.fromMillis(2500) })).toThrow( + TimestampMismatch, + ); + edge.writeFrame({ payload: enc.encode("edge"), timestamp: Timestamp.fromMillis(3000) }); + + // At or below the watermark the lower neighbor may be gone, so the watermark only caps + // the group and it is accepted without being trusted to time anything. + const track = producer.subscribe({ maxDelay: Milli(60_000) }); + expect(() => stampGroup(producer, 1, 4000)).toThrow(TimestampMismatch); + stampGroup(producer, 1, 500); + producer.close(); + + const timestamps = new Map(); + for (let group = await track.recvGroup(); group; group = await track.recvGroup()) { + timestamps.set(group.sequence, hooks.groupTimestamp(group)); + } + expect(timestamps.get(3)?.value).toBe(3000); + expect(timestamps.has(1)).toBe(true); + expect(timestamps.get(1)).toBeUndefined(); + } finally { + clock.restore(); + } +}); + +test("an untimed track takes groups in any order", () => { + const producer = new TrackProducer("test").accept({}); + for (const sequence of [5, 3, 4]) { + const group = new GroupProducer(sequence); + producer.writeGroup(group); + group.writeFrame({ payload: enc.encode(`${sequence}`) }); + group.close(); + } +}); diff --git a/js/net/src/track.ts b/js/net/src/track.ts index e993d0023c0..0a83fc4fb76 100644 --- a/js/net/src/track.ts +++ b/js/net/src/track.ts @@ -375,7 +375,7 @@ export class Consumer { } // The index of the first timeline group at or after `sequence`. -function timelineIndex(timeline: GroupConsumer[], sequence: number): number { +function timelineIndex(timeline: readonly { sequence: number }[], sequence: number): number { let lo = 0; let hi = timeline.length; while (lo < hi) { @@ -470,7 +470,39 @@ async function resolveInfo(state: TrackState): Promise { // A source group retained in the producer cache, with the mirror handed to each sink // so eviction can drop them together. // `arrived` is when the group joined this track, which its mirrors age a predecessor from. -type CachedGroup = { group: GroupProducer; mirrors: Map; arrived: Instant }; +// `bound` is its first frame's timestamp once checked to fit between its neighbors', which +// bounds theirs; `unverified` marks one accepted below the eviction watermark instead. +type CachedGroup = { + group: GroupProducer; + mirrors: Map; + arrived: Instant; + bound?: Timestamp; + unverified?: boolean; +}; + +// A group's checked start, kept by sequence. +type Start = { sequence: number; start: Timestamp }; + +// Whether `a` presents before `b`, whatever their scales. Cross-scale products can pass +// 2^53, so they are compared exactly in BigInt, as Rust does in u128. +function earlier(a: Timestamp, b: Timestamp): boolean { + if (a.scale === b.scale) return a.value < b.value; + const [ma, ea] = binary(a.value); + const [mb, eb] = binary(b.value); + const e = Math.min(ea, eb); + return (ma * BigInt(b.scale)) << BigInt(ea - e) < (mb * BigInt(a.scale)) << BigInt(eb - e); +} + +// A finite, non-negative `value` exactly as `mantissa * 2^exponent`. Doubling is exact, so +// a fraction becomes an integer within its binary digits. +function binary(value: number): [bigint, number] { + let exponent = 0; + while (!Number.isInteger(value)) { + value *= 2; + exponent--; + } + return [BigInt(value), exponent]; +} let bindProducerSequence: (producer: Producer, sequence: TrackSequence) => void; @@ -577,6 +609,16 @@ export class Producer { #cache: CachedGroup[] = []; // The same entries by sequence, so a write finds a duplicate without a scan. #cached = new Map(); + // The cached groups' checked starts, sorted by sequence, so a first frame finds its + // neighbors' with a search. + #bounds: Start[] = []; + // The highest-sequence checked start evicted. Eviction is not a sequence prefix, so a + // 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. @@ -602,11 +644,21 @@ export class Producer { static { hooks.endTrack = (producer, end) => producer.#endPending(end); + 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; if (own === sequence) return; sequence.next = Math.max(sequence.next, producer.#received); + // What earlier producers of the name wrote is gone from this track's cache, so their + // highest checked start is its eviction watermark. + const previous = sequence.bound; + if (previous && (!producer.#evicted || producer.#evicted.sequence < previous.sequence)) { + producer.#evicted = previous; + } + if (own.bound && (!previous || previous.sequence < own.bound.sequence)) sequence.bound = own.bound; producer.#sequence = sequence; if (producer.#state.closed.peek() !== undefined) return; own.holders--; @@ -792,6 +844,7 @@ export class Producer { #mirror(entry: CachedGroup, sink: TrackState): void { const dst = entry.group.mirror(); hooks.arriveGroup(dst, entry.arrived); + if (entry.unverified) hooks.unverifyGroup(dst); entry.mirrors.set(sink, dst); timelineInsert(sink.timeline, dst); void dst.readable().then(() => sink.timelineChanged.update((revision) => revision + 1)); @@ -831,9 +884,60 @@ export class Producer { } // Take a cached group out of the cache lookups. - #uncache(entry: CachedGroup): void { + #uncache(entry: CachedGroup, evicted = false): void { this.#cache.splice(this.#cache.indexOf(entry), 1); this.#cached.delete(entry.group.sequence); + this.#unbound(entry, evicted); + } + + // The starts a first frame of group `sequence` must fall between: the nearest checked + // starts cached on either side, since groups land out of sequence order, and the eviction + // watermark. Throws {@link TimestampMismatch} on a rewind; an equal start is allowed. Returns + // whether the group is late, at or below the watermark, where its floor may be gone, so + // the watermark caps it instead and it is accepted unverified. + #admit(sequence: number, start: Timestamp): boolean { + const bounds = this.#bounds; + const index = timelineIndex(bounds, sequence); + let floor = bounds[index - 1]?.start; + let ceiling = bounds[bounds[index]?.sequence === sequence ? index + 1 : index]?.start; + const evicted = this.#evicted; + const late = evicted !== undefined && sequence <= evicted.sequence; + if (evicted && late) { + if (ceiling === undefined || earlier(evicted.start, ceiling)) ceiling = evicted.start; + } else if (evicted) { + if (floor === undefined || earlier(floor, evicted.start)) floor = evicted.start; + } + if ((floor && earlier(start, floor)) || (ceiling && earlier(ceiling, start))) throw new TimestampMismatch(); + return late; + } + + // Record a cached group's checked start: a bound for its neighbors, or unverified. + #record(entry: CachedGroup, start: Timestamp, late: boolean): void { + if (late) { + entry.unverified = true; + for (const mirror of entry.mirrors.values()) hooks.unverifyGroup(mirror); + return; + } + entry.bound = start; + const sequence = entry.group.sequence; + const shared = this.#sequence.bound; + if (!shared || shared.sequence < sequence) this.#sequence.bound = { sequence, start }; + const last = this.#bounds.at(-1); + if (!last || last.sequence < sequence) this.#bounds.push({ sequence, start }); + else this.#bounds.splice(timelineIndex(this.#bounds, sequence), 0, { sequence, start }); + } + + // Drop a group's checked start as it leaves the cache, raising the watermark to it when + // it was evicted. + #unbound(entry: CachedGroup, evicted = false): void { + const start = entry.bound; + if (start === undefined) return; + const sequence = entry.group.sequence; + const index = timelineIndex(this.#bounds, sequence); + if (this.#bounds[index]?.sequence === sequence) this.#bounds.splice(index, 1); + if (evicted && (this.#evicted === undefined || sequence > this.#evicted.sequence)) { + this.#evicted = { sequence, start }; + } } // The newest group remains replayable while the track is live, even after the group @@ -897,6 +1001,7 @@ export class Producer { if (oldest === undefined || activity < oldest) oldest = activity; } else { this.#cached.delete(entry.group.sequence); + this.#unbound(entry, true); this.#evict(entry); } } @@ -931,8 +1036,10 @@ export class Producer { this.#pruneTimerAt = at; } - // Retain a source group and fan it out to every live sink. - #publish(group: GroupProducer): void { + // Retain a source group and fan it out to every live sink. A group that already holds + // frame 0 passes its checked `start` and whether it is `late`; any other checks frame 0 + // against this track when it lands. + #publish(group: GroupProducer, start?: Timestamp, late = false): void { const entry: CachedGroup = { group, mirrors: new Map(), arrived: now() }; this.#undrop(group.sequence); group.closed.then((closed) => { @@ -942,6 +1049,17 @@ export class Producer { if (current && current.group !== group) return; this.#drop(group.sequence, group.sequence + 1); }); + const sequence = group.sequence; + if (this.#remote) { + // Only a local publisher's writes are checked. + } else if (start) { + this.#record(entry, start, late); + } else { + hooks.admitGroupStart(group, (start) => { + const late = this.#admit(sequence, start); + return () => this.#record(entry, start, late); + }); + } this.#cache.push(entry); this.#cached.set(group.sequence, entry); this.#advance(group.sequence); @@ -1008,6 +1126,7 @@ export class Producer { * * Its frames must match the track's {@link Info.timescale}: a timestamp on every frame of * a timed track and none on an untimed one, or the write throws {@link TimestampMismatch}. + * So does a first frame that starts before the group below it or after the one above it. */ appendGroup(): GroupProducer { const group = new GroupProducer(this.#nextSequence()); @@ -1026,23 +1145,28 @@ export class Producer { * sequence again. Best effort (mirrors Rust): nothing remembers a sequence whose cache * entry is already gone, so a long-evicted sequence is accepted as new. Throws * {@link TimestampMismatch} if the group already holds a frame whose timedness - * disagrees with the track's {@link Info.timescale}. + * disagrees with the track's {@link Info.timescale}, or whose first frame starts before + * the group below it or after the one above it. */ writeGroup(group: GroupProducer) { this.#writable(group.sequence); this.#bind(group); this.#cap(group); - const existing = this.#cached.get(group.sequence); if (existing) { if (!(existing.group.closed.peek() instanceof Error)) { throw new Error(`duplicate group: sequence=${group.sequence}`); } + // Subscribers may have seen the aborted incarnation's start, so it stays a watermark. this.#evict(existing); - this.#uncache(existing); + this.#uncache(existing, true); } - this.#publish(group); + // 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 late = start !== undefined && this.#admit(group.sequence, start); + + this.#publish(group, start, late); } // Fan a datagram out to every live subscriber, dropping the oldest once the ring is full. @@ -1278,11 +1402,9 @@ export class Subscriber { // // Its first timestamp is the furthest presentation time `sequence` could still reach. // An upper bound, deliberately: a frame's duration is not on the wire, so a group's own - // last timestamp says where it starts presenting, not where it ends. Only the - // *immediate* successor counts: timestamps need not rise with sequence (a rewind - // reorders them), so a later stamped group proves nothing about where an unstamped - // successor will begin, and shrinking the bound is the unsafe direction. An unstamped - // successor leaves the reach unbounded until it presents a frame. + // last timestamp says where it starts presenting, not where it ends. The bound is the + // *immediate* successor's start, so an unstamped successor leaves the reach unbounded + // until it presents a frame. #successor(sequence: number, end?: number): GroupConsumer | undefined { const timeline = this.#state.timeline; for (let i = timelineIndex(timeline, sequence + 1); i < timeline.length; i++) { diff --git a/quest/m1/README.md b/quest/m1/README.md index ca4135f7fcd..bb7ee53e67d 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -101,7 +101,7 @@ lands before the cut. - [qmux no-wait opens](/quest/m1/qmux-no-wait.md) - @moq/qmux rejects an over-limit create when waitUntilAvailable is false, like Chrome - [TS fidelity and recovery](/quest/m1/ts-reliability/README.md) - both modes have explicit profiles, automated validation, loss tests, and a dedicated guide - [Resume reorder](/quest/m1/resume-reorder.md) - a group whose stream header arrives after its successor's still resumes on failover; only a dropped or long-gone group fails -- [Refuse rewinds](/quest/m1/refuse-rewinds.md) - a timed track refuses a group whose first frame is earlier than the previous group's, in Rust and js/net, on every hop +- [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 - [JS pending tail](/quest/m1/js-pending-tail.md) - a JS reader holds for a track's pending tail like Rust - [WebGPU on Safari](/quest/m1/webgpu-safari.md) - the WebGPU renderer is verified on Safari 26 for macOS and iOS - [Plan: HLS export failure signal](/quest/m1/hls-failure-signal.md) - decide what players see when the exporter's timeline fails, live or history, so a truncated recording never looks complete diff --git a/quest/m1/lite-time-provenance.md b/quest/m1/lite-time-provenance.md new file mode 100644 index 00000000000..586562c9b38 --- /dev/null +++ b/quest/m1/lite-time-provenance.md @@ -0,0 +1,38 @@ +# [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 237202f3479..a9fa9d8ee86 100644 --- a/quest/m1/public-model/ingest.md +++ b/quest/m1/public-model/ingest.md @@ -152,6 +152,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. - 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/refuse-rewinds.md b/quest/m1/refuse-rewinds.md deleted file mode 100644 index 852c6c6c725..00000000000 --- a/quest/m1/refuse-rewinds.md +++ /dev/null @@ -1,140 +0,0 @@ -# [S] A track refuses a timestamp that rewinds - -## Goal - -A timed track never holds a group whose first frame is earlier than the -previous group's, in Rust and js/net. Writing one fails loudly instead of -being cached and served. Untimed tracks are exempt. - -## Plan - -Decided by the maintainer on 2026-10-13 -([#5446](https://github.com/moq-dev/moq/pull/5446)): never support timestamp -rewinds. Today the model accepts them, and its expiry logic works around -them. - -Where a group's start is stamped: - -- Rust: `GroupState::stamp` (`rs/moq-net/src/model/group.rs`) sets a - group's start from its first frame. Every write path calls it: - `group::Producer::write_frame`, `write_frames`, and `open_frame` behind - `create_frame` and `create_frame_owned`, which lite and IETF ingress use. - The track keeps groups by sequence in its `lookup` map - (`rs/moq-net/src/model/track.rs`). -- JS: `state.timestamp ??= frame.timestamp` in `js/net/src/group.ts` stamps - the start; the track keeps groups by sequence in its sorted `timeline` - (`js/net/src/track.ts`). - -The rule: when a group's first frame lands, its start must be at or after -the nearest cached, stamped, non-aborted group below it, and at or before -the nearest one above it. Both sides are checked because groups can be -created and received out of sequence order. An equal start is allowed, as -moq-mux's `container::Producer` already allows a keyframe at the previous -group's start. The write fails without writing: every Rust write path -commits the frame (or installs the partial one) before `stamp` today, so -the check runs before any cache, frame, or index change, and the start is -stamped only once it passes. A session receiving such a group aborts the -track with that error and keeps the session. The ingress handlers must -route it there: today Rust's IETF subscriber (`rs/moq-net/src/ietf/subscriber.rs`) -sends `TimestampMismatch` down the generic branch that aborts only the -group, and js/net's (`js/net/src/ietf/subscriber.ts`) closes the track only -for `ProtocolViolation` or `MalformedTrack`. Check the lite subscribers the -same way, and test that the track, not just the group, closes. - -Cases the rule must cover (from review): - -- Only an original start counts. A group received from a later frame - (`start_at(hdr.frame_start)` in `rs/moq-net/src/lite/subscriber.rs`, or a - partial fetch) stamps its first *received* frame, which can sit above a - valid successor's start, since moq-mux lets a keyframe overlap the previous - group's last frame. A group whose first held frame is not index 0 has no - known start: it is neither checked nor used as a neighbor's bound. -- The neighbor check and the first stamp are one atomic step under the - shared track state, taken in the existing track-before-group lock order. - Groups are written on independent streams, so two unstamped neighbors - checked under only their own locks could each accept a rewind. -- A JS `GroupProducer` can hold frames before `track.writeGroup` attaches - it, so its start is stamped with no timeline to check. Check an already - started group when it is attached, before it is cached or fanned out. -- Eviction must not drop the bound. An unstamped group whose stamped - predecessor expires (Rust's `evict_expired_scan`, JS's `#prune`) would - otherwise have no lower neighbor and accept a rewind. The track keeps the - sequence and original start of the highest-sequence group it evicted. A - group above it is checked against both its cached neighbors and that - watermark, never one in place of the other, since eviction is not a - sequence prefix (each recently written group is retained on its own). One - entry, so no history grows with the cache. -- A late or refetched group at or below that watermark (JS `writeGroup` - accepts long-evicted sequences; Rust takes older groups after `finish`) - is checked against its surviving cached neighbors plus the watermark as an - upper bound, but its lower bound may be gone. Open, for the implementer to - confirm with the maintainer. Recommended: accept it unverified, keep it out - of every other group's bounds and out of timestamp-driven expiry, so - backfill below the watermark keeps working (fetched backfill is already - served by sequence only). Alternative: refuse a late group whose lower - bound was evicted, which keeps the Goal absolute but refuses legitimate - backfill of evicted history. - -The check costs one ordered neighbor lookup per group, not per frame. The -existing `group_write_frames` benchmark writes into an otherwise empty -track, so add a row that stamps groups swept over the retained group count, -where a scan of `lookup` would show as a slope. - -Recommended: every track producer refuses, local or received. The check -sits where the start is stamped, which publishers, relays, and Rust and JS -subscribers all share, so one rule covers every hop. A buggy or foreign -publisher (including a moq-transport peer) fails at the first hop instead -of being forwarded, and every hop's expiry and delivery logic (`reach`, the -drift budget, #5446's delivered floor) can rely on starts never falling. -Alternative: refuse only local writes (`append_group`, `create_group`, and -the JS equivalents) and let received groups through. It is cheaper on the -forwarding path, but relays forward a peer's rewind and every hop must still -tolerate one. - -Error: recommended, reuse `TimestampMismatch` (stream code 0x39 on the wire) -and broaden its doc, in Rust, js/net, and the code's description in -`drafts/draft-lcurley-moq-lite.md`, to "a frame's timestamp doesn't fit its -track". No new code, no new public variant. Alternative: a distinct -`TimestampRewind` error with its own stream code, a wire addition that -needs a draft entry. - -Reconcile the docs: - -- `TrackState::reach` in track.rs and `#reach` in js/net's track.ts say - "timestamps need not rise with sequence (a rewind reorders them)". Keep the - immediate-successor rule and drop the rewind rationale: the bound is the - immediate servable successor's start, unbounded while that successor is - unstamped. Tightening `reach` to the first stamped group above, which the - refusal makes safe, is out of scope; suggest a follow-up if a stall shows - it matters. -- Delete tests whose premise is a rewind, such as - `reach_follows_the_immediate_successor_not_a_later_rewind`. -- The container layers already refuse rewinds (moq-mux's - `container::Producer` returns `TimestampRewind`; the Rust and js/hang - container consumers abort one). Keep all of them: they floor every frame - against the previous group's start, while the model checks only a group's - first frame, and a foreign wire publisher bypasses the producer's floor. -- If #5446 lands first, update its note that late-group delivery after a - rewind is unspecified. - -Regression tests in both languages: a timed track refuses a group whose -first frame rewinds, whether appended, created out of order, received -through the streamed-frame path, written after its predecessor was evicted -(including a delayed stamp above the watermark while an older group -survives a non-prefix eviction), or (JS) written before `writeGroup`; an -equal start is accepted; a resumed partial group of a valid overlapping -pair is accepted; a late group below the watermark is refused against a -surviving cached neighbor, follows the chosen rule otherwise, and never -causes a later refetch to be refused; an untimed track accepts groups in -any order. - -Public API: write paths can now fail with `TimestampMismatch` on a rewind. -Wire: none if `TimestampMismatch` is reused; the draft's description of -code 0x39 broadens. - -## Related - -- [#5446](https://github.com/moq-dev/moq/pull/5446) - where the maintainer decided rewinds are never supported -- [JS track handover](/quest/m1/js-group-handover.md) - mirrors Rust's resume contract, which relies on starts never falling -- [Same-epoch importers](/quest/m2/hop-aligned-import.md) - importers already refuse a rewind at the container layer -- [Epoch anchor](/quest/m2/epoch-anchor.md) - a rewind stays fatal to an importer, which restarts under a successor epoch diff --git a/rs/moq-archive/src/writer.rs b/rs/moq-archive/src/writer.rs index 1b63846ecd6..3d55b7e132f 100644 --- a/rs/moq-archive/src/writer.rs +++ b/rs/moq-archive/src/writer.rs @@ -1644,56 +1644,6 @@ mod tests { ); } - #[tokio::test] - async fn a_decreasing_timestamp_fails_the_recording() { - let source = broadcast::Info::new().produce(); - let video = track(&source, "video"); - - let store = Store::new(InMemory::new(), "rec"); - let writer = Writer::new(store, source.consume(), Config::default()).await.unwrap(); - writer.control().track("video", media()).await.unwrap(); - - group(&video, 0, &[1000]); - group(&video, 1, &[500]); - video.finish().unwrap(); - source.close(); - - assert_eq!( - writer.run().await, - Err(Error::Source("track video group 1: timestamp 500 precedes 1000".into())) - ); - } - - /// A malformed group fails the recording only after committing everything reported before - /// it, including the open record, so the archive stays readable up to the bad frame. - #[tokio::test] - async fn a_failed_recording_flushes_what_was_reported() { - let source = broadcast::Info::new().produce(); - let video = track(&source, "video"); - - let store = Store::new(InMemory::new(), "rec"); - let writer = Writer::new(store.clone(), source.consume(), Config::default()) - .await - .unwrap(); - writer.control().track("video", media()).await.unwrap(); - - group(&video, 0, &[0]); - group(&video, 1, &[1000]); - // Under the minimum, so groups 1 and 2 share the record still open at the failure. - group(&video, 2, &[1500]); - group(&video, 3, &[1200]); - - assert_eq!( - writer.run().await, - Err(Error::Source( - "track video group 3: timestamp 1200 precedes 1500".into() - )) - ); - let records = window(&store, "video").await; - assert_eq!(groups(&records), vec![(0, 0), (1, 2)]); - check_objects(&store, "video", &records).await; - } - /// Stopping a recording whose source is still live finishes every timeline and ends `run`. #[tokio::test] async fn stopping_ends_without_the_source() { diff --git a/rs/moq-cli/src/play/media.rs b/rs/moq-cli/src/play/media.rs index 91048ac73fe..b7257d34862 100644 --- a/rs/moq-cli/src/play/media.rs +++ b/rs/moq-cli/src/play/media.rs @@ -865,17 +865,17 @@ mod tests { moq_mux::container::Consumer::new(subscriber, Container::Legacy(moq_mux::container::Kind::Video)), Buffered(Vec::new()), )); - let picture = |index: u64| moq_mux::container::Frame { + let picture = |index: u64, keyframe: bool| moq_mux::container::Frame { timestamp: moq_net::Timestamp::from_millis(index * 20).unwrap(), duration: Some(moq_net::Timestamp::from_millis(20).unwrap()), - keyframe: index == 0, + keyframe, payload: Bytes::from_static(b"picture"), }; for index in 0..3 { let mut pcm = packet(index, 0.25); pcm.keyframe = index == 0; audio.write(pcm).unwrap(); - video.write(picture(index)).unwrap(); + video.write(picture(index, index == 0)).unwrap(); tokio::time::sleep(PACKET_DURATION).await; } if audio_end { @@ -886,12 +886,12 @@ mod tests { } for index in 0..3 { if audio_end || index == 0 { - let mut pcm = packet(if audio_end { index } else { index + 3 }, 0.5); + let mut pcm = packet(index + 3, 0.5); pcm.keyframe = index == 0; audio.write(pcm).unwrap(); } if video_end || index == 0 { - video.write(picture(if video_end { index } else { index + 3 })).unwrap(); + video.write(picture(index + 3, video_end && index == 0)).unwrap(); } } if audio_end { @@ -908,7 +908,7 @@ mod tests { audio.write(packet(index, 0.5)).unwrap(); } if !video_end { - video.write(picture(index)).unwrap(); + video.write(picture(index, false)).unwrap(); } } tokio::task::yield_now().await; @@ -1214,7 +1214,7 @@ mod tests { video.discontinuity().unwrap(); // A burst longer than the age budget evicts this replacement's only keyframe. for index in 0..15 { - video.write(picture(index * 20, index == 0)).unwrap(); + video.write(picture(1_000 + index * 20, index == 0)).unwrap(); } video.discontinuity().unwrap(); for index in 0..3 { diff --git a/rs/moq-cli/src/play/video.rs b/rs/moq-cli/src/play/video.rs index 7733c7bc55c..4a4207e05e1 100644 --- a/rs/moq-cli/src/play/video.rs +++ b/rs/moq-cli/src/play/video.rs @@ -1014,8 +1014,8 @@ mod tests { producer.write(frame(ms, ms == 0)).unwrap(); } producer.discontinuity().unwrap(); - for ms in [0, 33, 66] { - producer.write(frame(ms, ms == 0)).unwrap(); + for ms in [100, 133, 166] { + producer.write(frame(ms, ms == 100)).unwrap(); } producer.finish().unwrap(); let decoder = Buffered::new(2); diff --git a/rs/moq-ffi/src/test.rs b/rs/moq-ffi/src/test.rs index 410534358af..d961ef574af 100644 --- a/rs/moq-ffi/src/test.rs +++ b/rs/moq-ffi/src/test.rs @@ -1051,7 +1051,7 @@ async fn whole_media_frames_require_the_callers_timestamp() { broadcast.close().unwrap(); } -#[tokio::test(start_paused = true)] +#[tokio::test] async fn catalog_now_fixes_the_clock_for_later_media() { let broadcast = MoqBroadcastProducer::new().unwrap(); let catalog = MoqMediaCatalogProducer::new(broadcast.clone()).unwrap(); diff --git a/rs/moq-flate/src/stream/mod.rs b/rs/moq-flate/src/stream/mod.rs index 1664ddbb8fb..342165026b0 100644 --- a/rs/moq-flate/src/stream/mod.rs +++ b/rs/moq-flate/src/stream/mod.rs @@ -516,13 +516,13 @@ mod test { .unwrap(); let subscriber = track.subscribe(replaying()); - // Publish sequence 1 before sequence 0, the way reordering delivers them. + // Publish sequence 1 before sequence 0, the way reordering delivers them. Each still + // starts after the one before it. for sequence in [1u64, 0] { let mut flate = crate::Encoder::new(); let mut group = track.create_group(moq_net::group::Info { sequence }).unwrap(); - group - .write_frame(moq_net::Timestamp::now(), flate.frame(&[sequence as u8; 8])) - .unwrap(); + let timestamp = moq_net::Timestamp::from_millis(sequence).unwrap(); + group.write_frame(timestamp, flate.frame(&[sequence as u8; 8])).unwrap(); group.finish().unwrap(); } track.finish().unwrap(); diff --git a/rs/moq-mux/src/container/consumer.rs b/rs/moq-mux/src/container/consumer.rs index f3bb074f0ee..fe9c03c825c 100644 --- a/rs/moq-mux/src/container/consumer.rs +++ b/rs/moq-mux/src/container/consumer.rs @@ -1658,15 +1658,23 @@ mod tests { // ---- Malformed rewind ---- + /// The track refuses a group whose first frame rewinds, but not a later frame that does, + /// so the consumer floors every frame against the previous group's start. #[tokio::test] - async fn a_group_starting_below_the_previous_start_aborts() { + async fn a_frame_below_the_previous_start_aborts() { let mut track = track_producer("test", hang::container::track_info(hang::catalog::PRIORITY.video)); let mut consumer = Consumer::new(track.subscribe(None), Container::Legacy(crate::container::Kind::Data)); write_group(&mut track, 0, &[ts(100_000)]); assert_eq!(consumer.read_frame().await.unwrap().unwrap().timestamp, ts(100_000)); - write_group(&mut track, 1, &[ts(0)]); + write_group(&mut track, 1, &[ts(100_000), ts(0)]); track.finish().unwrap(); - let err = consumer.read().await.unwrap_err(); + let err = loop { + match consumer.read().await { + Ok(Some(_)) => continue, + Ok(None) => panic!("the rewound frame ended the track cleanly"), + Err(err) => break err, + } + }; assert!(matches!( err, crate::Error::TimestampRewind(TimestampRewind { timestamp, floor }) diff --git a/rs/moq-mux/src/container/fmp4/import.rs b/rs/moq-mux/src/container/fmp4/import.rs index b09395a6095..ad69473d10c 100644 --- a/rs/moq-mux/src/container/fmp4/import.rs +++ b/rs/moq-mux/src/container/fmp4/import.rs @@ -1096,7 +1096,8 @@ impl Import { } /// The source signalled a new timeline (an HLS `EXT-X-DISCONTINUITY`), so the next - /// fragment's decode time is not compared with the last one on each track. + /// fragment's decode time is not compared with the last one on each track. A timeline that + /// restarts below where it was still fails: a track refuses a group that rewinds. pub fn discontinuity(&mut self) { for track in self.text.values_mut() { track.end_decode_time = None; diff --git a/rs/moq-mux/src/container/fmp4/import_test.rs b/rs/moq-mux/src/container/fmp4/import_test.rs index 0831233ee2d..09b1dc5fee3 100644 --- a/rs/moq-mux/src/container/fmp4/import_test.rs +++ b/rs/moq-mux/src/container/fmp4/import_test.rs @@ -1643,7 +1643,8 @@ async fn import_refuses_a_restart() { } /// Skipping sequences (an HLS media-sequence gap) is not a new timeline: a rewound decode time -/// after a seek is still refused. Only a signalled discontinuity starts the comparison afresh. +/// after a seek is still refused. A signalled discontinuity skips the import's own comparison, +/// but the track still refuses a group that starts before the one before it. #[tokio::test] async fn seek_keeps_the_decode_time_until_a_discontinuity() { let (init, _, _) = bbb_init(); @@ -1665,8 +1666,11 @@ async fn seek_keeps_the_decode_time_until_a_discontinuity() { fmp4.discontinuity(); fmp4.seek(200).unwrap(); - fmp4.decode(&live_session(0, 5)) - .expect("a signalled discontinuity starts a new timeline"); + let err = fmp4.decode(&live_session(0, 5)).unwrap_err(); + assert!( + matches!(err, crate::Error::Moq(moq_net::Error::TimestampMismatch)), + "{err:?}" + ); } /// The catalog is first published at the first fragment, carrying the clock that fragment diff --git a/rs/moq-mux/src/container/ts/export_test.rs b/rs/moq-mux/src/container/ts/export_test.rs index 36a9a8900fe..dc7a300d25a 100644 --- a/rs/moq-mux/src/container/ts/export_test.rs +++ b/rs/moq-mux/src/container/ts/export_test.rs @@ -7881,14 +7881,16 @@ async fn a_timeline_restarting_at_zero_fails_the_export() { .unwrap() .with_delay(Duration::from_millis(500)); let start = tokio::time::Instant::now(); - // One-second groups of 20 ms frames for 10 s, then a group back at zero. + // One-second groups of 20 ms frames for 10 s, then a group back at zero. The track refuses + // a group whose first frame rewinds, so that group opens at the previous start. for sequence in 0..11u64 { tokio::time::sleep_until(start + Duration::from_secs(sequence)).await; let base = if sequence < 10 { sequence * 1_000 } else { 0 }; let mut group = track.create_group(moq_net::group::Info { sequence }).unwrap(); for k in 0..50u64 { + let millis = if sequence == 10 && k == 0 { 9_000 } else { base + k * 20 }; let frame = Frame { - timestamp: Timestamp::from_millis(base + k * 20).unwrap(), + timestamp: Timestamp::from_millis(millis).unwrap(), duration: None, payload: Bytes::from_static(&[0x21, 0x10, 0x04, 0x60]), keyframe: k == 0, diff --git a/rs/moq-net/MIGRATE.md b/rs/moq-net/MIGRATE.md index 9060a2c08eb..1a66b269119 100644 --- a/rs/moq-net/MIGRATE.md +++ b/rs/moq-net/MIGRATE.md @@ -4,6 +4,12 @@ How to update your code for each breaking change, newest first. ## Unreleased +### 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 publisher whose timeline restarts, such as an HLS `EXT-X-DISCONTINUITY` that resets decode times, publishes a new broadcast instead. + ### A bare track handle is not demand (#5820) Holding a `track::Consumer` sends nothing and no longer keeps the producer used. diff --git a/rs/moq-net/benches/group.rs b/rs/moq-net/benches/group.rs index 1446fbcd405..0d05dc0cbb1 100644 --- a/rs/moq-net/benches/group.rs +++ b/rs/moq-net/benches/group.rs @@ -8,9 +8,16 @@ //! cached group, swept over cache depth so a per-delivery scan shows up as a slope. //! It also covers the steady-state path after a subscriber has reached the edge. //! +//! `track_stamp_groups` covers stamping a group's first frame, which checks it +//! against its neighbors' starts, swept over cache depth so a scan shows as a slope: +//! `append` among stamped groups, `sparse` among groups created but not yet stamped. +//! //! Run with `cargo bench -p moq-net`. -use std::{hint::black_box, time::Duration}; +use std::{ + hint::black_box, + time::{Duration, Instant}, +}; use bytes::Bytes; use criterion::{BatchSize, BenchmarkId, Criterion, Throughput, criterion_group, criterion_main}; @@ -354,12 +361,65 @@ fn bench_relay_frames(c: &mut Criterion) { let _ = c; } +/// Append and stamp groups on a track already holding N cached groups. +/// +/// A group's first frame is checked against the nearest stamped starts on either side, +/// once per group. Each fresh track takes at most N more groups, so the cache stays +/// within 2N for the whole measurement. +fn bench_stamp(c: &mut Criterion) { + let payload = Bytes::from(vec![0u8; PAYLOAD]); + let mut g = c.benchmark_group("track_stamp_groups"); + for &n in &DEPTHS { + g.bench_with_input(BenchmarkId::new("append", n), &n, |b, &n| { + b.iter_custom(|iters| { + let mut total = Duration::ZERO; + let mut left = iters; + while left > 0 { + let ctx = filled_track(n, &payload); + let batch = left.min(n as u64); + let start = Instant::now(); + for _ in 0..batch { + let mut group = ctx.track.append_group().unwrap(); + group.write_frame(Timestamp::ZERO, payload.clone()).unwrap(); + group.finish().unwrap(); + } + total += start.elapsed(); + left -= batch; + } + total + }); + }); + // N groups created before any is stamped, then stamped in order: each one's upper + // neighbor is a run of unstamped groups, which a slot scan walks. + g.bench_with_input(BenchmarkId::new("sparse", n), &n, |b, &n| { + b.iter_custom(|iters| { + let mut total = Duration::ZERO; + let mut left = iters; + while left > 0 { + let ctx = filled_track(0, &payload); + let mut groups: Vec<_> = (0..n).map(|_| ctx.track.append_group().unwrap()).collect(); + let batch = left.min(n as u64); + let start = Instant::now(); + for group in groups.iter_mut().take(batch as usize) { + group.write_frame(Timestamp::ZERO, payload.clone()).unwrap(); + } + total += start.elapsed(); + left -= batch; + } + total + }); + }); + } + g.finish(); +} + criterion_group!( benches, bench_write, bench_read, bench_roundtrip, bench_track_recv, + bench_stamp, bench_relay_frames ); criterion_main!(benches); diff --git a/rs/moq-net/src/error.rs b/rs/moq-net/src/error.rs index eb86c552f40..134b1e11cc0 100644 --- a/rs/moq-net/src/error.rs +++ b/rs/moq-net/src/error.rs @@ -235,9 +235,10 @@ pub enum StreamError { #[error("group too large")] GroupTooLarge, - /// A frame's timestamp doesn't match its track: missing on a timed track, present on - /// an untimed one, or out of range for the track's timescale. - #[error("frame timestamp doesn't match track timescale")] + /// A frame's timestamp doesn't fit its track: missing on a timed track, present on an + /// untimed one, out of range for the track's timescale, or starting a group earlier + /// than the group before it. + #[error("frame timestamp doesn't fit its track")] TimestampMismatch, /// The grant does not cover this subscription, fetch, or announcement, or no longer @@ -490,9 +491,10 @@ pub enum Error { #[error("frame already open")] FrameOpen, - /// A frame's timestamp doesn't match its track: it's missing on a timed track, - /// present on an untimed track, or out of range for the track's timescale. - #[error("frame timestamp doesn't match track timescale")] + /// A frame's timestamp doesn't fit its track: it's missing on a timed track, present + /// on an untimed track, out of range for the track's timescale, or starts a group + /// earlier than the group before it. + #[error("frame timestamp doesn't fit its track")] TimestampMismatch, /// The group was evicted by its own track to pay eviction debt under memory diff --git a/rs/moq-net/src/ietf/subscriber.rs b/rs/moq-net/src/ietf/subscriber.rs index fabe21d3d10..69707d87402 100644 --- a/rs/moq-net/src/ietf/subscriber.rs +++ b/rs/moq-net/src/ietf/subscriber.rs @@ -2786,6 +2786,7 @@ 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. @@ -3559,6 +3560,7 @@ 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)); + track.received(); // Nothing feeds this copy live: what it caches is fetched, never the live edge. track.set_idle(); Some(Target::Resume(Idle { @@ -10987,6 +10989,7 @@ mod stitch_tests { "video", track::Info::default().with_timescale(timescale.map(|_| Timescale::MICRO)), ); + track.received(); let fill = kio::Producer::new(fill); { @@ -12686,6 +12689,37 @@ 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. + #[moq_clock::test] + async fn a_received_rewind_is_accepted() { + use futures::FutureExt; + + let mut h = Harness::new(Fill::Done, vec![draft18_subgroup(SEQUENCE, true, 0, &[b"rewound"])]); + h.subscriber.version = Version::Draft18; + // Starts past the incoming group's first object, at `timestamp(0)`. + let mut before = h.track.create_group((SEQUENCE - 1).into()).unwrap(); + before.write_frame(timestamp(4000), b"before".as_slice()).unwrap(); + before.finish().unwrap(); + + let mut consumer = h.track.subscribe(None); + assert_eq!(read_group(&mut consumer).await.0, SEQUENCE - 1); + + let mut stream = read_draft18(&h.session).await; + h.subscriber.clone().recv_group(&mut stream).await.unwrap(); + // The stream was read to its end, so the group is already delivered. + let mut group = consumer + .recv_group() + .now_or_never() + .expect("delivered") + .unwrap() + .unwrap(); + assert_eq!(group.sequence, SEQUENCE); + let frame = group.read_frame().now_or_never().unwrap().unwrap().unwrap(); + assert_eq!(frame.payload.as_ref(), b"rewound"); + assert!(h.track.closed().now_or_never().is_none()); + } + /// A head that stops short of where the tail starts would leave a hole in the middle of /// the group, which the model cannot express. Both halves go, and the head is published /// as the prefix it is. diff --git a/rs/moq-net/src/lite/subscriber.rs b/rs/moq-net/src/lite/subscriber.rs index 5f59f20715d..8550aaa3b88 100644 --- a/rs/moq-net/src/lite/subscriber.rs +++ b/rs/moq-net/src/lite/subscriber.rs @@ -2221,6 +2221,75 @@ 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. + async fn receive_rewind(version: Version) -> (Result<(), Error>, track::Producer) { + use crate::transport::poll::Session as _; + + let mut script = Vec::new(); + let mut encoder = crate::coding::Encoder::new(&mut script, version.into()); + lite::Group { + subscribe: 7, + sequence: 5, + frame_start: 0, + } + .encode(&mut encoder, version) + .unwrap(); + // Frame 0 at 500ms: the first delta is absolute. + encoder.varint(crate::coding::varint::zigzag(500)).unwrap(); + encoder.bytes(b"rewound").unwrap(); + let mut session = crate::lite::test_transport::ScriptedSession::eof(script); + let subscriber = Subscriber::new(SubscriberConfig { + runtime: crate::time::sim(), + session: session.clone(), + origin: origin::Config::new(crate::Hop::new(1).unwrap()).produce(), + recv_bandwidth: None, + version, + peer_setup: crate::lite::PeerSetup::new(false, crate::transport::Binding::InBand), + peer_hop: None, + cost: None, + going_away: Default::default(), + 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(); + let mut before = track.create_group(4u64.into()).unwrap(); + before + .write_frame(Timestamp::from_millis(1000).unwrap(), &b"before"[..]) + .unwrap(); + before.finish().unwrap(); + subscriber.subscribes.lock().insert( + 7, + TrackEntry { + producer: track.clone(), + timescale: Some(Timescale::MILLI), + tail: Default::default(), + }, + ); + + let (_, recv) = session.open_bi().await.unwrap(); + let mut group = GroupRecv::new(subscriber, Reader::new(recv, version)); + let res = kio::wait(|waiter| group.poll_serve(waiter)).await; + (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. + #[moq_clock::test] + async fn a_received_rewind_is_accepted_at_every_version() { + for version in [Version::Lite05, Version::Lite06, Version::Lite07] { + let (res, track) = receive_rewind(version).await; + res.unwrap_or_else(|err| panic!("{version}: {err:?}")); + assert!(track.closed().now_or_never().is_none(), "{version}"); + } + } + /// 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. @@ -5544,6 +5613,7 @@ impl ServeLoop { false => request, }; let serving = request.accept(info); + serving.received(); Self { demand: serving.demand(), serving, diff --git a/rs/moq-net/src/model/broadcast.rs b/rs/moq-net/src/model/broadcast.rs index af171978233..33c7f564e4a 100644 --- a/rs/moq-net/src/model/broadcast.rs +++ b/rs/moq-net/src/model/broadcast.rs @@ -1539,6 +1539,36 @@ mod test { ); } + /// A replacement continues the name's timeline, so its groups can't start before the + /// highest one an earlier producer checked, while a new broadcast starts over. + #[test] + fn replacement_refuses_a_rewind_past_its_predecessor() { + let at = |millis| crate::Timestamp::from_millis(millis).unwrap(); + let producer = Info::new().produce(); + + let first = producer.create_track("media", crate::track::Info::timed()).unwrap(); + let mut group = first.append_group().unwrap(); + group.write_frame(at(1000), b"a".as_slice()).unwrap(); + group.finish().unwrap(); + drop((group, first)); + + let second = producer.create_track("media", crate::track::Info::timed()).unwrap(); + let mut group = second.create_group(crate::group::Info { sequence: 1 }).unwrap(); + assert!(matches!( + group.write_frame(at(500), b"b".as_slice()), + Err(Error::TimestampMismatch) + )); + group.write_frame(at(1000), b"b".as_slice()).unwrap(); + + let next = Info::new().produce(); + let track = next.create_track("media", crate::track::Info::timed()).unwrap(); + track + .append_group() + .unwrap() + .write_frame(at(500), b"c".as_slice()) + .unwrap(); + } + /// Names requested but never served leave nothing behind, so a peer asking for /// arbitrary names cannot grow the broadcast. #[test] diff --git a/rs/moq-net/src/model/cache.rs b/rs/moq-net/src/model/cache.rs index 1f485c8951b..54151413e38 100644 --- a/rs/moq-net/src/model/cache.rs +++ b/rs/moq-net/src/model/cache.rs @@ -667,6 +667,11 @@ impl Track { &self.pool } + /// The track holding this account's groups, while it is still open. + pub(super) fn track(&self) -> Option> { + self.state.upgrade() + } + /// The track's reads parked on their drift budget. pub(super) fn wakes(&self) -> &expiry::Wakes { &self.wakes diff --git a/rs/moq-net/src/model/delivered.rs b/rs/moq-net/src/model/delivered.rs index 696038238c9..c7f0893dbf2 100644 --- a/rs/moq-net/src/model/delivered.rs +++ b/rs/moq-net/src/model/delivered.rs @@ -56,7 +56,7 @@ impl Delivered { } /// The lowest sequence handed out above a gap: once it is past the reader's budget, - /// so is the gap, since group timestamps on a track never rewind. + /// so is the gap, since a timed track refuses a group whose start rewinds. pub fn above_gap(&self) -> Option { let mut runs = self.runs.keys(); match runs.next()? { diff --git a/rs/moq-net/src/model/group.rs b/rs/moq-net/src/model/group.rs index e7ddd6685f2..7ec1eb170ed 100644 --- a/rs/moq-net/src/model/group.rs +++ b/rs/moq-net/src/model/group.rs @@ -176,6 +176,9 @@ pub(super) struct GroupState { // rather than read off `frames` so an abort doesn't erase it. timeline: Timeline, + // How the track judged the start against its neighbors' when frame 0 landed. + origin: Origin, + // Once finalized, the total number of frames the group will ever contain. Recorded // at finish so the count outlives an abort that clears the cache. pub(super) fin: Option, @@ -231,6 +234,23 @@ enum Timeline { Timed { start: Timestamp, latest: Timestamp }, } +/// How a timed group's start was checked against its neighbors' (see +/// [`track::Neighbors`]), which decides whether it bounds theirs. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub(super) enum Origin { + /// Not checked: untimed, started past frame 0, written with no open track, or received + /// from a session. A start read off a later frame is not where the group began, and a + /// received one may not be media time, so it bounds nothing. + #[default] + Unchecked, + /// Frame 0 fit between its neighbors, so its start bounds theirs. + Bound, + /// Frame 0 landed at or below the track's eviction watermark, where its lower bound + /// may be gone. Accepted so backfill keeps working, but it neither bounds a neighbor + /// nor drives media-time expiry. + Unverified, +} + impl GroupState { /// Refuse a write that would advance the group to `next_index`, past the track's end. fn admit(&self, next_index: usize) -> Result<()> { @@ -359,6 +379,25 @@ impl GroupState { self.poll_terminal(index) } + /// Check frame 0's `timestamp` against the track's `neighbors`, returning how to record + /// it. Anything but a timed frame 0 checked under an open local track is unchecked. + fn check_start(&self, neighbors: Option<&track::Neighbors>, timestamp: Option) -> Result { + match (neighbors, timestamp) { + (Some(neighbors), Some(start)) if self.next_index == 0 => neighbors.admit(start), + _ => Ok(self.origin), + } + } + + /// Settle how the group holds its start, returning it if frame 0 just became a bound. + fn settle(&mut self, origin: Origin) -> Option { + let bound = origin == Origin::Bound && self.origin != Origin::Bound; + self.origin = origin; + match self.timeline { + Timeline::Timed { start, .. } if bound => Some(start), + _ => None, + } + } + /// Record a frame's place in presentation time: the first frame starts the group, /// and each later one extends a timed group's end. Every frame matches its track /// ([`on_track`]), so a group never mixes the two. @@ -539,6 +578,9 @@ struct Alive { // out of `GroupState` has to read both under one guard, or the two halves can // straddle the abort: see `Producer::live_first_frame`. aborted: AtomicBool, + // Set once frame 0's rewind check ran, so later writes skip the track lock it takes. A + // stale false only takes that lock once more. + checked: AtomicBool, // Whether the abort was the publisher giving the group up rather than cache policy // (`Old`, `Timeout`, `Evicted`). Stored before `aborted`, under the same rules. cut: AtomicBool, @@ -666,6 +708,7 @@ impl Producer { info, state: state.clone(), aborted: AtomicBool::new(false), + checked: AtomicBool::new(false), cut: AtomicBool::new(false), access, }); @@ -698,6 +741,18 @@ impl Producer { self.track.timescale } + /// The track to check this group's start against, while frame 0 may still be unwritten. + /// + /// The caller holds its lock until the frame is recorded, so the check and the stamp are + /// one step against every other group's (locks are ordered track then group). Only a + /// timed group not yet checked pays for it, once. + fn first_frame_track(&self) -> Option> { + if self.track.timescale.is_none() || self.alive.checked.load(Ordering::Relaxed) { + return None; + } + self.cache.track() + } + /// Start the group at frame `index` rather than 0, so the first frame written lands /// there. /// @@ -743,6 +798,9 @@ impl Producer { return Err(Error::FrameTooLarge); } + let track = self.first_frame_track(); + let track = track.as_ref().map(|track| track.read()); + let neighbors = track.as_ref().and_then(|track| track.neighbors(self.info.sequence)); let mut state = modify(&self.state)?; if state.fin.is_some() { return Err(Error::Closed); @@ -755,6 +813,7 @@ impl Producer { .checked_add(1) .ok_or(Error::BoundsExceeded(crate::coding::BoundsExceeded))?; state.admit(next_index)?; + let origin = state.check_start(neighbors.as_ref(), timestamp)?; debug_assert!(state.partial.is_none(), "a frame is already open"); let size = payload.len() as u64; if state.would_overflow(1, size) { @@ -767,7 +826,14 @@ impl Producer { state.next_index = next_index; state.committed = state.next_index; state.stamp(timestamp); + let bound = state.settle(origin); + self.alive.checked.store(true, Ordering::Relaxed); drop(state); + // The group's lock is released first: sharing reads it back through its slot. + if let (Some(track), Some(start)) = (&track, bound) { + track.share_bound(self.info.sequence, start); + } + drop(track); // With the group lock released (lock order is track then group), settle // eviction debt if enough has been written since the track last paid. @@ -803,6 +869,9 @@ impl Producer { let count = frames.len(); let bytes: u64 = frames.filled().iter().map(|frame| frame.payload.len() as u64).sum(); + let track = self.first_frame_track(); + let track = track.as_ref().map(|track| track.read()); + let neighbors = track.as_ref().and_then(|track| track.neighbors(self.info.sequence)); let mut state = modify(&self.state)?; if state.fin.is_some() { return Err(Error::Closed); @@ -815,6 +884,14 @@ impl Producer { .checked_add(count) .ok_or(Error::BoundsExceeded(crate::coding::BoundsExceeded))?; state.admit(next_index)?; + let first = frames.filled().first().map(|frame| frame.timestamp); + let origin = match first { + Some(timestamp) => state.check_start( + neighbors.as_ref(), + on_track(timestamp, self.track.timescale).expect("timestamp scale checked above"), + )?, + None => state.origin, + }; if state.would_overflow(count, bytes) { return Err(self.abort_too_large(state)); } @@ -834,7 +911,16 @@ impl Producer { } state.next_index = next_index; state.committed = next_index; + let bound = state.settle(origin); + if count > 0 { + self.alive.checked.store(true, Ordering::Relaxed); + } drop(state); + // The group's lock is released first: sharing reads it back through its slot. + if let (Some(track), Some(start)) = (&track, bound) { + track.share_bound(self.info.sequence, start); + } + drop(track); self.cache.settle(now); if let Some(latest) = latest { @@ -898,6 +984,9 @@ impl Producer { return Err(Error::FrameTooLarge); } + let track = self.first_frame_track(); + let track = track.as_ref().map(|track| track.read()); + let neighbors = track.as_ref().and_then(|track| track.neighbors(self.info.sequence)); let mut state = modify(&self.state)?; if state.fin.is_some() { return Err(Error::Closed); @@ -910,6 +999,7 @@ impl Producer { .checked_add(1) .ok_or(Error::BoundsExceeded(crate::coding::BoundsExceeded))?; state.admit(next_index)?; + let origin = state.check_start(neighbors.as_ref(), timestamp)?; // Only one frame is ever in flight, so the declared size is the most the cache // can grow by before the next check. let metadata_bytes = header.fields.as_ref().map_or(0, |fields| fields.heap()); @@ -940,7 +1030,14 @@ impl Producer { // Opening the frame is enough: the header carries the timestamp, so the group's // place in time is known before a single payload byte streams in. state.stamp(timestamp); + let bound = state.settle(origin); + self.alive.checked.store(true, Ordering::Relaxed); drop(state); + // The group's lock is released first: sharing reads it back through its slot. + if let (Some(track), Some(start)) = (&track, bound) { + track.share_bound(self.info.sequence, start); + } + drop(track); // With the group lock released (lock order is track then group), settle // eviction debt if enough has been written since the track last paid. @@ -1174,10 +1271,24 @@ impl Producer { /// place in the media timeline rather than when it happened to be delivered. That /// is what lets the track tell a burst of old content apart from live content (see /// [`track::Subscriber`]). + /// + /// `None` for a group accepted unverified below the track's eviction watermark too: + /// nothing proves where it starts, so it drives no media-time expiry. pub(super) fn timestamp(&self) -> Option { - match self.state.read().timeline { - Timeline::Timed { start, .. } => Some(start), - Timeline::Empty | Timeline::Untimed => None, + let state = self.state.read(); + match state.timeline { + Timeline::Timed { start, .. } if state.origin != Origin::Unverified => Some(start), + _ => None, + } + } + + /// The start this group lends its neighbors' rewind check: its first frame's timestamp, + /// once frame 0 was checked to fit between theirs. + pub(super) fn bound(&self) -> Option { + let state = self.state.read(); + match state.timeline { + Timeline::Timed { start, .. } if state.origin == Origin::Bound => Some(start), + _ => None, } } @@ -1190,9 +1301,10 @@ impl Producer { /// fallen behind is there nothing left worth delivering. Grows as the group does, so /// a group still receiving frames stays fresh and a stalled one ages in place. pub(super) fn latest(&self) -> Option { - match self.state.read().timeline { - Timeline::Timed { latest, .. } => Some(latest), - Timeline::Empty | Timeline::Untimed => None, + let state = self.state.read(); + match state.timeline { + Timeline::Timed { latest, .. } if state.origin != Origin::Unverified => Some(latest), + _ => None, } } diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index 4409d592ba6..599b76664ed 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -35,6 +35,7 @@ use std::{ sync::Arc, sync::OnceLock, sync::atomic::{AtomicBool, AtomicU64, Ordering}, + sync::{Mutex, PoisonError}, task::{Poll, ready}, time::Duration, }; @@ -191,19 +192,40 @@ impl Info { } /// The sequence namespace a track name keeps within its broadcast, shared by every -/// producer that serves the name, so a replacement that would append a sequence an earlier -/// one wrote is refused. Holds one past the highest sequence written. +/// producer that serves the name, so a replacement continues the same timeline: it can't +/// append a sequence an earlier one wrote, nor start a group before an earlier one's. #[derive(Clone, Default)] -pub(super) struct Sequence(Arc); +pub(super) struct Sequence(Arc); + +#[derive(Default)] +struct SequenceState { + // One past the highest sequence written. + next: AtomicU64, + // The highest-sequence checked start written, which bounds a replacement's groups the + // way the eviction watermark bounds a track's own. + bound: Mutex>, +} impl Sequence { /// One past the highest sequence any producer of the name wrote. fn next(&self) -> u64 { - self.0.load(Ordering::Relaxed) + self.0.next.load(Ordering::Relaxed) } fn advance(&self, sequence: u64) { - self.0.fetch_max(sequence.saturating_add(1), Ordering::Relaxed); + self.0.next.fetch_max(sequence.saturating_add(1), Ordering::Relaxed); + } + + fn bound(&self) -> Option<(u64, Timestamp)> { + *self.0.bound.lock().unwrap_or_else(PoisonError::into_inner) + } + + /// Keep group `sequence`'s checked `start` if it is the highest-sequence one yet. + fn share_bound(&self, sequence: u64, start: Timestamp) { + let mut bound = self.0.bound.lock().unwrap_or_else(PoisonError::into_inner); + if bound.is_none_or(|(highest, _)| sequence > highest) { + *bound = Some((sequence, start)); + } } /// Whether nothing holds it and nothing was written, so forgetting it changes nothing. @@ -299,6 +321,17 @@ pub(crate) struct TrackState { // `max_sequence` when an earlier producer of the name wrote further. sequence: Sequence, + // The checked starts cached groups hold, by sequence (see [`group::Origin::Bound`]), so a + // first frame finds its neighbors without walking groups not yet stamped. Filled through + // the read guard a first frame holds, so it needs its own lock, never contended. + bounds: Mutex>, + + // The sequence and start of the highest-sequence group that left the cache holding a + // checked start (see [`group::Origin::Bound`]). Eviction is not a sequence prefix, so a + // 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: Option<(u64, Timestamp)>, + // The sequence of the newest cached group: the live edge, protected from // eviction by never entering the eviction order until the track is `closed`. // Tracked separately from `max_sequence` because datagrams advance that shared @@ -348,6 +381,9 @@ 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: 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 costs one entry. @@ -394,6 +430,33 @@ pub(crate) struct TrackState { routes: Option, } +/// The starts a group's first frame must fall between: a timed track never holds a group +/// whose start is earlier than the one before it. Read under the track lock and checked +/// under it, so no two groups stamped concurrently can each accept a rewind. +#[derive(Clone, Copy, Debug, Default)] +pub(super) struct Neighbors { + floor: Option, + ceiling: Option, + // At or below the eviction watermark, where the floor may have been evicted. + late: bool, +} + +impl Neighbors { + /// Refuse a `start` that rewinds past a neighbor's with [`Error::TimestampMismatch`], + /// or say how the group holds it. An equal start is allowed: a keyframe may land at the + /// previous group's start. + pub(super) fn admit(&self, start: Timestamp) -> Result { + if self.floor.is_some_and(|floor| start < floor) || self.ceiling.is_some_and(|ceiling| start > ceiling) { + return Err(Error::TimestampMismatch); + } + Ok(if self.late { + group::Origin::Unverified + } else { + group::Origin::Bound + }) + } +} + /// A cached group plus its bookkeeping in the track's `lookup` map. /// /// Access times and the evictable-population sample live in the group's own @@ -495,6 +558,11 @@ impl Slot { } } + /// The checked start of whichever copy began at frame 0. + fn bound(&self) -> Option { + self.copies().find_map(group::Producer::bound) + } + /// Abort every copy this slot holds. fn abort(&self, err: Error) { for group in self.copies() { @@ -553,6 +621,67 @@ pub(crate) struct FetchOutcome { } impl TrackState { + /// The starts a first frame of group `sequence` must fall between: the nearest checked + /// starts cached on either side, since groups land out of sequence order, and the + /// eviction watermark. A group at or below the watermark is late: its floor may be gone, + /// 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. + pub(super) fn neighbors(&self, sequence: u64) -> Option { + if self.received { + return None; + } + let bounds = self.bounds.lock().unwrap_or_else(PoisonError::into_inner); + let below = bounds.range(..sequence).next_back().map(|(_, start)| *start); + let above = bounds + .range((Bound::Excluded(sequence), Bound::Unbounded)) + .next() + .map(|(_, start)| *start); + drop(bounds); + Some(match self.evicted { + Some((evicted, start)) if sequence <= evicted => Neighbors { + floor: below, + ceiling: Some(above.map_or(start, |above| above.min(start))), + late: true, + }, + Some((_, start)) => Neighbors { + floor: Some(below.map_or(start, |below| below.max(start))), + ceiling: above, + late: false, + }, + None => Neighbors { + floor: below, + ceiling: above, + late: false, + }, + }) + } + + /// Record group `sequence`'s just-checked `start` as a bound for its neighbors and the + /// name's later producers. A group stamped after it left the cache bounds nothing. + pub(super) fn share_bound(&self, sequence: u64, start: Timestamp) { + if self.lookup.get(&sequence).and_then(Slot::bound) != Some(start) { + return; + } + self.bounds + .lock() + .unwrap_or_else(PoisonError::into_inner) + .insert(sequence, start); + self.sequence.share_bound(sequence, start); + } + + /// Take group `sequence` out of the cache, raising the eviction watermark to its start. + fn uncache(&mut self, sequence: u64) -> Option { + let slot = self.remove_slot(sequence)?; + if let Some(start) = slot.bound() + && self.evicted.is_none_or(|(evicted, _)| sequence > evicted) + { + self.evicted = Some((sequence, start)); + } + Some(slot) + } + /// Record sequences the upstream publisher gave up. fn drop_groups(&mut self, groups: Range) { if groups.is_empty() { @@ -565,7 +694,7 @@ impl TrackState { /// Remove an aborted slot, keeping a cut as a drop: reclaiming it must not erase a /// disposition a relay still has to forward. fn reclaim_aborted(&mut self, sequence: u64) { - if let Some(slot) = self.remove_slot(sequence) + if let Some(slot) = self.uncache(sequence) && slot.visible && slot.group.is_cut() && let Some(end) = sequence.checked_add(1) @@ -824,12 +953,10 @@ impl TrackState { /// /// An upper bound, deliberately. A frame's duration is not on the wire, so a group's /// own last timestamp says where it *starts* presenting, not where it ends; only its - /// successor's start proves it cannot run past it. And only the *immediate* servable - /// successor counts: timestamps need not rise with sequence (a rewind reorders them), - /// so a later stamped group proves nothing about where an unstamped successor will - /// begin, and shrinking the bound is the unsafe direction. An unstamped successor - /// therefore leaves the reach unbounded until it presents its first frame, and so - /// does having no servable successor below `cap` at all. + /// successor's start proves it cannot run past it. The bound is the *immediate* + /// servable successor's start, so an unstamped successor leaves the reach unbounded + /// until it presents its first frame, and so does having no servable successor below + /// `cap` at all. fn reach(&self, sequence: u64, cap: Option) -> Option { self.successor(sequence, cap)?.group.timestamp() } @@ -915,8 +1042,8 @@ impl TrackState { /// budget fall out for free instead of needing a special case. /// /// The edge must sit strictly above the candidate. The live edge is never late - /// against itself, and backfill or the tail of a rewound timeline can carry a high - /// timestamp on a low sequence without being an edge at all. + /// against itself, and a group received from past its first frame carries a later + /// frame's timestamp on a low sequence without being an edge at all. fn is_stale(&self, sequence: u64, edge: &Edge, budget: Duration) -> bool { self.lookup.contains_key(&sequence) && self.drifted(sequence, edge, budget) } @@ -1156,7 +1283,7 @@ impl TrackState { // Take the group out of the cache and abort it, so any consumer // still reading surfaces `Error::Old` instead of blocking forever // on a frame that will never arrive. - self.remove_slot(sequence).unwrap().abort(Error::Old); + self.uncache(sequence).unwrap().abort(Error::Old); } } @@ -1229,12 +1356,21 @@ impl TrackState { self.lookup .retain(|_, slot| slot.copies().any(group::Producer::is_finished)); self.withheld = self.lookup.values().filter(|slot| slot.pending).count(); + let lookup = &self.lookup; + self.bounds + .get_mut() + .unwrap_or_else(PoisonError::into_inner) + .retain(|sequence, _| lookup.contains_key(sequence)); } /// Take `sequence` out of the cache, keeping the withheld count in step. fn remove_slot(&mut self, sequence: u64) -> Option { let slot = self.lookup.remove(&sequence)?; self.withheld -= usize::from(slot.pending); + self.bounds + .get_mut() + .unwrap_or_else(PoisonError::into_inner) + .remove(&sequence); Some(slot) } @@ -1304,7 +1440,8 @@ impl TrackState { } /// Reject a sequence that is still cached; a dead (aborted or evicted) - /// incarnation is removed so a fresh group can serve the sequence again. + /// incarnation is removed so a fresh group can serve the sequence again. Subscribers may + /// have seen its start, so it stays as the eviction watermark. /// /// Best effort: nothing remembers a sequence whose slot is already gone, so a /// publisher re-sending a long-evicted sequence is accepted as new. @@ -1316,7 +1453,7 @@ impl TrackState { if self.covering_group(sequence, frame_start).is_some() { return Err(Error::Duplicate); } - self.remove_slot(sequence); + self.uncache(sequence); Ok(()) } @@ -1377,7 +1514,13 @@ impl TrackState { head: None, }, ); - self.withheld -= replaced.map_or(0, |slot| usize::from(slot.pending)); + if let Some(replaced) = replaced { + self.withheld -= usize::from(replaced.pending); + self.bounds + .get_mut() + .unwrap_or_else(PoisonError::into_inner) + .remove(&sequence); + } self.cache.wakes().arrived(sequence..=sequence); if visible && !pending { self.arrival.push_back((sequence, stamp)); @@ -1560,7 +1703,7 @@ impl TrackState { self.debt -= size; paid = paid.saturating_add(size); - self.remove_slot(sequence).unwrap().abort(Error::Evicted); + self.uncache(sequence).unwrap().abort(Error::Evicted); } } @@ -2206,7 +2349,11 @@ impl Producer { let timescale = self.modify()?.info.as_ref().unwrap().timescale; let timestamp = group::on_track(timestamp.into(), timescale)?; let mut group = self.append_group()?; - group.write_frame(timestamp, frame)?; + if let Err(err) = group.write_frame(timestamp, frame) { + // A rewind is only known once the group exists; end it rather than leave it open. + let _ = group.abort(err.clone()); + return Err(err); + } group.finish()?; Ok(()) } @@ -2334,6 +2481,16 @@ 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. #[cfg_attr(not(feature = "async"), allow(dead_code))] // pending: /quest/m1/public-model/ingest.md @@ -2791,6 +2948,9 @@ fn set_sequence(state: &kio::Producer, sequence: Sequence) { state.max_sequence.is_none(), "a track joins its namespace before writing" ); + // What earlier producers of the name wrote is gone from this track's cache, so their + // highest checked start is its eviction watermark. + state.evicted = sequence.bound(); state.sequence = sequence; } } @@ -4798,8 +4958,8 @@ impl Subscriber { /// Once means once per sequence: a group the publisher aborts and re-creates, or that /// a second route delivers too, is not handed out again. Once a group is past the /// budget below, the subscriber stops expecting every group before it, so a gap that - /// old is never filled. That assumes group timestamps on a track never rewind; which - /// late groups are handed out after a rewind is unspecified. + /// old is never filled, which holds because a timed track refuses a group whose start + /// rewinds. /// /// Groups are semi-reliable, and the [`Subscription::max_delay`] budget is the other /// thing (alongside eviction and a moving start) that decides which of them arrive: @@ -7288,7 +7448,7 @@ mod test { #[cfg(feature = "async")] fn a_late_lower_group_within_the_budget_is_delivered() { let producer = track_producer("test", Info::timed()); - for (sequence, millis) in [(5, 0), (6, 1000), (7, 2000)] { + for (sequence, millis) in [(5, 1000), (6, 2000), (7, 3000)] { let mut group = producer.create_group(group::Info { sequence }).unwrap(); group .write_frame(Timestamp::from_millis(millis).unwrap(), bytes::Bytes::from_static(b"x")) @@ -7364,39 +7524,6 @@ mod test { assert!(matches!(result, Ok(None)), "the held group ends: {result:?}"); } - /// A rewound successor can keep a drained group within budget until its abort - /// moves the reach to the next cached group, without changing the track itself. - #[moq_clock::test] - #[cfg(feature = "async")] - async fn aborted_stamped_successor_wakes_a_parked_read() { - let mut producer = track_producer("test", Info::timed()); - let mut head = producer.append_group().unwrap(); - head.write_frame(Timestamp::ZERO, b"head".as_slice()).unwrap(); - let mut successor = producer.append_group().unwrap(); - successor - .write_frame(Timestamp::from_millis(30_000).unwrap(), b"next".as_slice()) - .unwrap(); - append_at(&mut producer, 1000); - append_at(&mut producer, 20_000); - let mut sub = producer.subscribe(Subscription::default().with_max_delay(Duration::from_secs(1))); - let mut reading = sub.recv_group().await.unwrap().unwrap(); - assert_eq!(reading.sequence, 0, "the rewound successor extends the reach"); - assert!(reading.read_frame().await.unwrap().is_some()); - - let woken = Arc::new(std::sync::atomic::AtomicBool::new(false)); - let waker = futures::task::waker(Arc::new(FlagWake(woken.clone()))); - let mut cx = std::task::Context::from_waker(&waker); - let mut next = std::pin::pin!(reading.read_frame()); - assert!(next.as_mut().poll(&mut cx).is_pending()); - successor.abort(Error::Cancel).unwrap(); - assert!( - woken.load(Ordering::SeqCst), - "the stamped successor's abort lost its wakeup" - ); - let result = next.as_mut().poll(&mut cx); - assert!(matches!(result, Poll::Ready(Ok(None))), "the head is stale: {result:?}"); - } - /// A first timestamp on a *newer* group can convict a held one, so the held reader has /// to be woken by it. The conviction needs a group beyond the held one's successor: /// a group is bounded by where its successor begins, so the successor itself never @@ -7577,7 +7704,7 @@ mod test { } /// An aborted edge hands the edge to a group between it and the successor, and neither - /// the abort nor that group's first frame touches the track, so the read must still wake. + /// the abort nor that group's frames touch the track, so the read must still wake. #[test] #[cfg(feature = "async")] fn a_parked_read_watches_its_edge_abort() { @@ -7609,9 +7736,12 @@ mod test { assert!(next.as_mut().poll(&mut cx).is_pending()); edge.abort(Error::Cancel).unwrap(); - between - .write_frame(Timestamp::from_millis(10_000).unwrap(), bytes::Bytes::from_static(b"x")) - .unwrap(); + // Starts between its neighbors, then runs past the deadline. + for millis in [1500, 10_000] { + between + .write_frame(Timestamp::from_millis(millis).unwrap(), bytes::Bytes::from_static(b"x")) + .unwrap(); + } assert!( woken.load(Ordering::SeqCst), "the edge's abort, then a new edge below it, lost the wakeup" @@ -7671,19 +7801,25 @@ mod test { assert!(matches!(result, Poll::Ready(Ok(None))), "the head is stale: {result:?}"); } - /// A rewound edge's abort hands the edge back to a lower group that already presented - /// past the deadline. No frame write crosses it afterwards, and the abort never touches - /// the track, so only the read watching the edge's abort sees the head go stale. + /// An edge's abort hands the edge back to a lower group whose frames already ran past + /// the deadline. No frame write crosses it afterwards, and the abort never touches the + /// track, so only the read watching the edge's abort sees the head go stale. #[test] #[cfg(feature = "async")] - fn a_parked_read_watches_its_rewound_edge_abort() { + fn a_parked_read_watches_its_overtaken_edge_abort() { let mut producer = track_producer("test", crate::track::Info::timed()); let mut subscriber = producer.subscribe(Subscription::default().with_max_delay(Duration::from_secs(5))); let mut head = producer.append_group().unwrap(); head.write_frame(Timestamp::ZERO, bytes::Bytes::from_static(b"head")) .unwrap(); append_at(&mut producer, 1000); // the successor bounds the head's reach - append_at(&mut producer, 10_000); // past the deadline, but below the edge + // Starts below the edge but presents past the deadline. + let mut long = producer.append_group().unwrap(); + for millis in [1500, 10_000] { + long.write_frame(Timestamp::from_millis(millis).unwrap(), bytes::Bytes::from_static(b"x")) + .unwrap(); + } + long.finish().unwrap(); let mut edge = producer.append_group().unwrap(); edge.write_frame(Timestamp::from_millis(2000).unwrap(), bytes::Bytes::from_static(b"x")) .unwrap(); @@ -7695,7 +7831,7 @@ mod test { .unwrap() .unwrap() .expect("head group"); - assert_eq!(held.sequence, 0, "the rewound edge keeps the head within budget"); + assert_eq!(held.sequence, 0, "the edge keeps the head within budget"); assert!(held.read_frame().now_or_never().unwrap().unwrap().is_some()); let woken = Arc::new(std::sync::atomic::AtomicBool::new(false)); @@ -8141,32 +8277,6 @@ mod test { assert_eq!(drain(&mut subscriber), vec![0, 2, 3]); } - /// Reach is the *immediate* successor's start, never a minimum across later groups. - /// Timestamps need not rise with sequence: a rewind can put a much earlier timestamp on - /// a much later group, and that group proves nothing about where the candidate's own - /// successor begins. Taking the minimum would shrink the bound and discard content that - /// is still well inside the budget. - #[test] - #[cfg(feature = "async")] - fn reach_follows_the_immediate_successor_not_a_later_rewind() { - let mut producer = track_producer("test", Info::timed()); - - // Group 0 is bounded by group 1 at 10s. Groups 2 and 3 rewind to 1s and 2s. - append_at(&mut producer, 0); - append_at(&mut producer, 10_000); - append_at(&mut producer, 1_000); - append_at(&mut producer, 2_000); - - // Group 0 reaches 10s, so nothing here proves it is past a 500ms budget: it could - // hold frames through nearly 10s. A minimum over later groups would put its reach - // at 1s and drop it. - let mut sub = producer.subscribe(Subscription::default().with_max_delay(Duration::from_millis(500))); - assert!( - drain(&mut sub).contains(&0), - "group 0 is bounded by its successor at 10s, not by a later rewind" - ); - } - fn untimed_producer() -> Producer { track_producer("test", None) } @@ -8598,21 +8708,21 @@ mod test { #[cfg(feature = "async")] fn a_lower_sequence_is_never_the_live_edge() { let producer = track_producer("test", Info::timed()); - // A high timestamp on a lower sequence is not a live edge: backfill served on - // demand sits there, and so does the tail of a timeline the publisher rewound. - // Only groups above the candidate anchor the measure, so neither can convict - // the group that follows it. + // A high timestamp on a lower sequence is not a live edge: a group whose frames run + // past where its successor starts sits there. Only groups above the candidate + // anchor the measure, so it cannot convict the group that follows it. let mut straggler = producer.create_group(0u64.into()).unwrap(); - straggler - .write_frame(Timestamp::from_millis(60_000).unwrap(), bytes::Bytes::from_static(b"x")) - .unwrap(); + for millis in [0, 60_000] { + straggler + .write_frame(Timestamp::from_millis(millis).unwrap(), bytes::Bytes::from_static(b"x")) + .unwrap(); + } straggler.finish().unwrap(); - let mut rewound = producer.create_group(1u64.into()).unwrap(); - rewound - .write_frame(Timestamp::from_millis(0).unwrap(), bytes::Bytes::from_static(b"x")) + let mut next = producer.create_group(1u64.into()).unwrap(); + next.write_frame(Timestamp::from_millis(0).unwrap(), bytes::Bytes::from_static(b"x")) .unwrap(); - rewound.finish().unwrap(); + next.finish().unwrap(); let mut subscriber = producer.subscribe(replay()); assert_eq!(drain(&mut subscriber), vec![0, 1]); @@ -12387,4 +12497,193 @@ mod test { } assert_eq!(subscriber.delivered_runs(), 1000); } + + /// A group `sequence` whose first frame lands at `millis`, or the write's refusal. A + /// refused group is aborted, as a caller would. + fn start_at_millis(producer: &Producer, sequence: u64, millis: u64) -> Result { + let mut group = producer.create_group(group::Info { sequence })?; + match group.write_frame(Timestamp::from_millis(millis).unwrap(), &b"x"[..]) { + Ok(()) => Ok(group), + Err(err) => { + group.abort(err.clone()).unwrap(); + Err(err) + } + } + } + + /// Take group `sequence` out of the cache as eviction does. + fn evict(producer: &Producer, sequence: u64) { + producer + .modify() + .unwrap() + .uncache(sequence) + .unwrap() + .abort(Error::Evicted); + } + + /// A rewound first frame is refused and leaves the group as it was, so the publisher can + /// still write a start at or after the group before it. + #[test] + fn an_appended_rewind_is_refused() { + let mut producer = track_producer("test", Info::timed()); + append_at(&mut producer, 1000); + + let mut group = producer.append_group().unwrap(); + let refused = group.write_frame(Timestamp::from_millis(500).unwrap(), &b"x"[..]); + assert!(matches!(refused, Err(Error::TimestampMismatch)), "{refused:?}"); + assert_eq!(group.frame_count(), 0); + assert_eq!(group.timestamp(), None); + + // An equal start is a keyframe at the previous group's start. + group + .write_frame(Timestamp::from_millis(1000).unwrap(), &b"x"[..]) + .unwrap(); + } + + /// The one-shot write ends the group it appended rather than leaving it open. + #[test] + fn a_rewound_single_frame_group_is_aborted() { + let mut producer = track_producer("test", Info::timed()); + append_at(&mut producer, 1000); + let refused = producer.write_frame(Timestamp::from_millis(500).unwrap(), &b"x"[..]); + assert!(matches!(refused, Err(Error::TimestampMismatch)), "{refused:?}"); + let state = producer.state.read(); + assert!(state.lookup.get(&1).unwrap().group.is_aborted()); + } + + /// Groups land out of sequence order, so a start is bounded from above as well as below. + #[test] + fn a_rewind_is_refused_whichever_group_lands_first() { + let producer = track_producer("test", Info::timed()); + start_at_millis(&producer, 5, 1000).unwrap(); + assert!(matches!( + start_at_millis(&producer, 3, 2000), + Err(Error::TimestampMismatch) + )); + start_at_millis(&producer, 3, 500).unwrap(); + assert!(matches!( + start_at_millis(&producer, 4, 400), + Err(Error::TimestampMismatch) + )); + start_at_millis(&producer, 4, 1000).unwrap(); + } + + /// The batched and streamed writes check frame 0 the same way, before writing anything. + #[test] + fn every_write_path_refuses_a_rewind() { + let mut producer = track_producer("test", Info::timed()); + append_at(&mut producer, 1000); + let rewound = Timestamp::from_millis(500).unwrap(); + + let mut batched = producer.append_group().unwrap(); + let mut frames = frame::Buffer::<4>::new(); + frames + .push(frame::Frame { + timestamp: Some(rewound), + payload: bytes::Bytes::from_static(b"x"), + }) + .unwrap(); + assert!(matches!( + batched.write_frames(&mut frames), + Err(Error::TimestampMismatch) + )); + assert_eq!(frames.len(), 1, "the batch is left to retry"); + assert_eq!(batched.frame_count(), 0); + + let mut streamed = producer.append_group().unwrap(); + let refused = streamed + .create_frame(frame::Info { + size: 1, + timestamp: Some(rewound), + }) + .err(); + assert!(matches!(refused, Some(Error::TimestampMismatch)), "{refused:?}"); + assert_eq!(streamed.frame_count(), 0); + } + + /// Eviction is not a sequence prefix: a group stamped after its predecessor was evicted + /// is still held above it, though an older group survives below. + #[test] + fn an_evicted_predecessor_still_bounds_a_delayed_stamp() { + let producer = track_producer("test", Info::timed()); + start_at_millis(&producer, 0, 0).unwrap(); + start_at_millis(&producer, 1, 1000).unwrap(); + let mut delayed = producer.create_group(group::Info { sequence: 2 }).unwrap(); + evict(&producer, 1); + + let refused = delayed.write_frame(Timestamp::from_millis(500).unwrap(), &b"x"[..]); + assert!(matches!(refused, Err(Error::TimestampMismatch)), "{refused:?}"); + delayed + .write_frame(Timestamp::from_millis(1000).unwrap(), &b"x"[..]) + .unwrap(); + } + + /// A group at or below the eviction watermark is checked against what survives, and + /// otherwise accepted unverified: it bounds no later group, so a refetch of evicted + /// history is never refused on its account, and it drives no media-time expiry. + #[test] + fn a_late_group_below_the_watermark_is_unverified() { + let producer = track_producer("test", Info::timed()); + for (sequence, millis) in [(0, 1000), (1, 2000), (2, 3000), (3, 4000)] { + start_at_millis(&producer, sequence, millis).unwrap(); + } + evict(&producer, 1); + evict(&producer, 2); + + // Below the surviving group 0, or above the watermark at group 2. + for millis in [500, 3500] { + assert!(matches!( + start_at_millis(&producer, 1, millis), + Err(Error::TimestampMismatch) + )); + } + let late = start_at_millis(&producer, 1, 2900).unwrap(); + assert_eq!(late.bound(), None, "a late group bounds nothing"); + assert_eq!(late.timestamp(), None, "nor drives media-time expiry"); + + // Group 2 really started before late group 1 claimed to. + start_at_millis(&producer, 2, 2500).unwrap(); + } + + /// Subscribers may have seen an aborted group's start, so a replacement at its sequence + /// leaves that start as the watermark, and no later group can start before it. + #[test] + fn a_replaced_aborted_group_still_bounds_later_ones() { + let producer = track_producer("test", Info::timed()); + start_at_millis(&producer, 0, 1000) + .unwrap() + .abort(Error::Cancel) + .unwrap(); + let replacement = start_at_millis(&producer, 0, 500).unwrap(); + assert_eq!(replacement.bound(), None, "a late replacement bounds nothing"); + assert!(matches!( + start_at_millis(&producer, 1, 600), + Err(Error::TimestampMismatch) + )); + start_at_millis(&producer, 1, 1000).unwrap(); + } + + /// A group received from past its first frame stamps a later frame, which may overlap + /// the next group's start. It has no known start, so it is neither checked nor a bound. + #[test] + fn a_resumed_partial_group_is_neither_checked_nor_a_bound() { + let producer = track_producer("test", Info::timed()); + start_at_millis(&producer, 0, 1000).unwrap(); + let mut partial = producer.create_group(group::Info { sequence: 1 }).unwrap(); + partial.start_at(3).unwrap(); + partial + .write_frame(Timestamp::from_millis(2100).unwrap(), &b"x"[..]) + .unwrap(); + start_at_millis(&producer, 2, 2000).unwrap(); + } + + /// An untimed track has no timeline, so its groups land in any order. + #[test] + fn an_untimed_track_takes_groups_in_any_order() { + let producer = untimed_producer(); + for sequence in [5, 3, 4] { + let mut group = producer.create_group(group::Info { sequence }).unwrap(); + group.write_frame(None, &b"x"[..]).unwrap(); + } + } } diff --git a/rs/moq-net/tests/integration/late_lower_group.rs b/rs/moq-net/tests/integration/late_lower_group.rs index 3d9ca6c09ca..753c0e0181b 100644 --- a/rs/moq-net/tests/integration/late_lower_group.rs +++ b/rs/moq-net/tests/integration/late_lower_group.rs @@ -34,10 +34,13 @@ fn produce_origin(hop: u64) -> moq_net::origin::Producer { producer } +/// Four frames starting at `sequence` seconds, so a group created late still starts +/// below the one after it. fn write_group(track: &moq_net::track::Producer, sequence: u64) { let mut group = track.create_group(moq_net::group::Info { sequence }).unwrap(); - for _ in 0..4 { - group.write_frame(Timestamp::now(), &b"x"[..]).unwrap(); + for frame in 0..4 { + let timestamp = Timestamp::from_millis(sequence * 1000 + frame).unwrap(); + group.write_frame(timestamp, &b"x"[..]).unwrap(); } group.finish().unwrap(); } diff --git a/rs/moq-net/tests/integration/main.rs b/rs/moq-net/tests/integration/main.rs index d155bb218f5..be987e7bd03 100644 --- a/rs/moq-net/tests/integration/main.rs +++ b/rs/moq-net/tests/integration/main.rs @@ -33,6 +33,7 @@ mod mesh_withdraw; mod model; mod multi_cdn; mod quiet_catalog; +mod received_rewind; mod rejoin; mod replacement; mod request_caps; diff --git a/rs/moq-net/tests/integration/received_rewind.rs b/rs/moq-net/tests/integration/received_rewind.rs new file mode 100644 index 00000000000..1f860b26b96 --- /dev/null +++ b/rs/moq-net/tests/integration/received_rewind.rs @@ -0,0 +1,181 @@ +#![cfg(feature = "async")] + +//! A relay re-serves what it received without checking it for a rewind. +//! +//! 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. + +use crate::support; + +use std::time::Duration; + +use moq_net::Version; +use moq_net::track::{Info, Position, Subscription}; +use support::harness::{MockConnectOptions, connect_mock}; + +const TIMEOUT: Duration = Duration::from_secs(10); + +/// Long enough that both groups are fresh. +const BUDGET: Duration = Duration::from_secs(60); + +fn produce_origin(hop: u64) -> moq_net::origin::Producer { + let (producer, driver) = + moq_net::origin::Producer::new(moq_net::origin::Config::new(moq_net::Hop::new(hop).unwrap())); + support::harness::spawn(driver); + producer +} + +fn write_group(track: &moq_net::track::Producer, sequence: u64) { + let mut group = track.create_group(moq_net::group::Info { sequence }).unwrap(); + group.write_frame(None, &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) { + let publisher = produce_origin(1); + let broadcast = publisher.create_broadcast("bcast").unwrap(); + let track = broadcast.create_track("data", Info::default()).unwrap(); + broadcast.announce(Default::default()).unwrap(); + + let relay_origin = produce_origin(2); + let mut options = MockConnectOptions::new("moq-lite-06".parse::().unwrap()); + options.server_publish = Some(publisher.consume()); + options.client_subscribe = Some(relay_origin.clone()); + let upstream = connect_mock(options).await; + + let subscriber = produce_origin(3); + let mut options = MockConnectOptions::new(downstream.parse::().unwrap()); + options.server_publish = Some(relay_origin.consume()); + options.client_subscribe = Some(subscriber.clone()); + let pair = connect_mock(options).await; + + let consumer = subscriber.consume(); + moq_clock::timeout(TIMEOUT, consumer.routed("bcast")) + .await + .expect("announce timeout") + .expect("routed"); + let remote = moq_clock::timeout(TIMEOUT, consumer.request_broadcast("bcast", None)) + .await + .expect("resolve timeout") + .expect("broadcast resolves"); + + let subscription = Subscription::default() + .with_max_delay(BUDGET) + .with_start(Some(Position::group(0))); + let reader = moq_clock::sim::spawn(async move { + let mut sub = remote + .track("data") + .unwrap() + .subscribe(subscription) + .await + .expect("subscribe"); + let timescale = sub.info().timescale; + let mut got = Vec::new(); + loop { + let next = moq_clock::timeout(TIMEOUT, sub.recv_group()) + .await + .expect("recv_group timed out"); + let Some(mut group) = next.expect("recv_group") else { + break; + }; + while moq_clock::timeout(TIMEOUT, group.read_frame()) + .await + .expect("read_frame timed out") + .expect("read_frame") + .is_some() + {} + got.push(group.sequence); + } + (timescale, got) + }); + + moq_clock::timeout(TIMEOUT, track.demand().used()) + .await + .expect("no subscriber appeared") + .unwrap(); + moq_clock::sleep(Duration::from_millis(10)).await; + write_group(&track, 1); + // Sent later, so group 0's send time lands after group 1's. + moq_clock::sleep(Duration::from_millis(10)).await; + write_group(&track, 0); + track.finish().unwrap(); + + let got = moq_clock::timeout(TIMEOUT * 3, reader) + .await + .expect("reader hung") + .expect("reader panicked"); + drop((track, pair, upstream, relay_origin, broadcast, publisher, subscriber)); + got +} + +#[moq_clock::test] +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] { + failures.push(format!("{downstream}: {got:?}")); + } + } + assert!(failures.is_empty(), "{failures:#?}"); +} + +/// 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. +async fn fetch_only(version: &str) -> Vec> { + let publisher = produce_origin(1); + let broadcast = publisher.create_broadcast("bcast").unwrap(); + let track = broadcast.create_track("data", Info::default()).unwrap(); + broadcast.announce(Default::default()).unwrap(); + write_group(&track, 0); + write_group(&track, 1); + track.finish().unwrap(); + + let subscriber = produce_origin(2); + let mut options = MockConnectOptions::new(version.parse::().unwrap()); + options.server_publish = Some(publisher.consume()); + options.client_subscribe = Some(subscriber.clone()); + let pair = connect_mock(options).await; + + let consumer = subscriber.consume(); + moq_clock::timeout(TIMEOUT, consumer.routed("bcast")) + .await + .expect("announce timeout") + .expect("routed"); + let remote = moq_clock::timeout(TIMEOUT, consumer.request_broadcast("bcast", None)) + .await + .expect("resolve timeout") + .expect("broadcast resolves"); + + let mut got = Vec::new(); + for sequence in [1, 0] { + let read = async { + let mut group = remote.track("data").unwrap().fetch_group(sequence, None).await?; + while group.read_frame().await?.is_some() {} + Ok(group.sequence) + }; + got.push(moq_clock::timeout(TIMEOUT, read).await.expect("fetch timed out")); + // Answered later, so group 0's send time lands after group 1's. + moq_clock::sleep(Duration::from_millis(10)).await; + } + drop((pair, remote, broadcast, publisher, subscriber)); + got +} + +#[moq_clock::test] +async fn fetches_alone_accept_descending_send_times() { + let mut failures = Vec::new(); + for version in ["moq-lite-05", "moq-lite-06"] { + let got = fetch_only(version).await; + if got.iter().any(Result::is_err) { + failures.push(format!("{version}: {got:?}")); + } + } + assert!(failures.is_empty(), "{failures:#?}"); +} diff --git a/rs/moq-stats/src/aggregate.rs b/rs/moq-stats/src/aggregate.rs index 6cfde68ac2d..05cd778dc10 100644 --- a/rs/moq-stats/src/aggregate.rs +++ b/rs/moq-stats/src/aggregate.rs @@ -1133,7 +1133,7 @@ mod tests { fn fail_traffic(&mut self) { let mut group = self.track.append_group().expect("append group"); group - .write_frame(Timestamp::ZERO, b"not json".to_vec()) + .write_frame(Timestamp::now(), b"not json".to_vec()) .expect("write frame"); group.finish().expect("finish group"); }