From dcc640ac75e63316f050238ae245991e8c7187fb Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sun, 11 Oct 2026 00:35:37 -0700 Subject: [PATCH 1/3] chore(quest): claim JSON window timestamps Co-Authored-By: GPT-6 Astra From f57fb1a7b9bb6668302267429c3ca5637b12628e Mon Sep 17 00:00:00 2001 From: Luke Curley <432854+kixelated@users.noreply.github.com> Date: Wed, 21 Oct 2026 23:02:46 -0700 Subject: [PATCH 2/3] feat(json)!: window consumers return each event's frame timestamp The Rust and JS window consumers yield `Timed>`, matching the snapshot and stream consumers. A record restated by a group header carries the header frame's time, since the wire has no per-record time. Room chat subscribers pass the send time through; timeline readers drop it, since each record carries its own pts. Co-Authored-By: Claude Opus 5.5 --- js/hang/src/timeline.ts | 3 +- js/json/MIGRATE.md | 3 +- js/json/src/window/consumer.ts | 20 +++++--- js/json/src/window/window.test.ts | 61 ++++++++++++++++++++--- js/room/MIGRATE.md | 5 ++ js/room/README.md | 2 +- js/room/src/chat.test.ts | 9 ++-- js/room/src/chat.ts | 18 ++++--- quest/m1/README.md | 1 - quest/m1/json-window-timestamps.md | 27 ---------- quest/m1/refresh/README.md | 1 - quest/m1/refresh/json.md | 4 -- rs/moq-archive/src/reader/tests.rs | 2 +- rs/moq-json/MIGRATE.md | 3 +- rs/moq-json/src/window/consumer.rs | 21 +++++--- rs/moq-json/src/window/mod.rs | 79 ++++++++++++++++++++++++++++++ rs/moq-mux/src/timeline.rs | 5 +- rs/moq-room/MIGRATE.md | 4 ++ rs/moq-room/src/chat.rs | 52 +++++++++++++------- 19 files changed, 233 insertions(+), 87 deletions(-) delete mode 100644 quest/m1/json-window-timestamps.md diff --git a/js/hang/src/timeline.ts b/js/hang/src/timeline.ts index 1dabc7f394a..e9e6e1c4acf 100644 --- a/js/hang/src/timeline.ts +++ b/js/hang/src/timeline.ts @@ -85,7 +85,8 @@ export class Consumer implements AsyncIterable { /** Get the next timeline event, or `undefined` when the track ends. Throws on an untimed record. */ async next(): Promise { - const event = await this.#window.next(); + // Each record carries its own pts, so the frame timestamp adds nothing. + const event = (await this.#window.next())?.value; if (event === undefined) return undefined; if ("push" in event) { const { index } = event.push; diff --git a/js/json/MIGRATE.md b/js/json/MIGRATE.md index fb5fcf6ef66..3bf664a5d5e 100644 --- a/js/json/MIGRATE.md +++ b/js/json/MIGRATE.md @@ -11,7 +11,8 @@ Fix any publisher that writes a repeated key. ### Consumers return `Timed` values (#5099) -The snapshot and stream consumers' `next()` and async iterator yield `{ value, at }`, where `at` is the frame's timestamp and absent on an untimed track. +The snapshot, stream, and window consumers' `next()` and async iterator yield `{ value, at }`, where `at` is the frame's timestamp and absent on an untimed track. +A window event's `value` is the `Event` that `next()` used to return; a record restated by a group header carries the header frame's time. A snapshot consumer's `next()` yields every state in order; call `latest()` to skip to the newest, as `next()` used to. Where you awaited `next()` for the newest value, read `(await consumer.latest())?.value`. A caller that passes the result on as `any` or `unknown` still compiles, so search for every `next()` and `for await` rather than relying on type errors. diff --git a/js/json/src/window/consumer.ts b/js/json/src/window/consumer.ts index 7b949aac4e7..e2f1ad484b4 100644 --- a/js/json/src/window/consumer.ts +++ b/js/json/src/window/consumer.ts @@ -26,6 +26,10 @@ const GAPS: Moq.StreamCode[] = [ * window yields nothing for records already delivered, so this reads as one continuous stream of * {@link Event}s regardless of how the publisher framed them. * + * Each event carries the timestamp of the frame that produced it, absent on an untimed track. The + * wire has no per-record time, so a record a reader first learns of from a group header (one it + * joined after, or missed while behind) gets the header frame's time, not its original push time. + * * @public */ export class Consumer { @@ -34,6 +38,8 @@ export class Consumer { #group?: Moq.Group.Consumer; #codec?: Group; + // Timestamp of the frame whose events the decoder is draining. + #at?: Moq.Time.Timestamp; #reading = false; constructor(config: ConsumerConfig) { @@ -46,8 +52,8 @@ export class Consumer { return this.#decoder.offset; } - /** Get the next event, or `undefined` once the track ends. */ - next(): Promise | undefined> { + /** Get the next event and its frame's timestamp, or `undefined` once the track ends. */ + next(): Promise> | undefined> { if (this.#reading) throw new Error("multiple calls to next not supported"); this.#reading = true; return this.#read().finally(() => { @@ -55,11 +61,12 @@ export class Consumer { }); } - async #read(): Promise | undefined> { + async #read(): Promise> | undefined> { for (;;) { - // Drain what the frames already decoded produced before reading more. + // Drain what the last frame produced before reading more, so every queued event is that + // frame's and shares its timestamp. const event = this.#decoder.next(); - if (event) return event; + if (event) return { value: event, at: this.#at }; if (!this.#group) { this.#group = await this.#track.nextGroup(); @@ -89,11 +96,12 @@ export class Consumer { if (!this.#codec) throw new Error("an open MoQ group has a window codec"); this.#codec.decode(frame.payload); + this.#at = frame.timestamp; } } /** Iterate over events until the track ends. */ - async *[Symbol.asyncIterator](): AsyncIterator> { + async *[Symbol.asyncIterator](): AsyncIterator>> { for (;;) { const event = await this.next(); if (event === undefined) return; diff --git a/js/json/src/window/window.test.ts b/js/json/src/window/window.test.ts index bd0c3da9a0f..9afc87b48dd 100644 --- a/js/json/src/window/window.test.ts +++ b/js/json/src/window/window.test.ts @@ -1,5 +1,5 @@ import { expect, test } from "bun:test"; -import { Group, Error as NetError, StreamCode, Time, Track } from "@moq/net"; +import { Group, Error as NetError, StreamCode, Time, type Timed, Track } from "@moq/net"; import { Consumer } from "./consumer.ts"; import { Decoder, Encoder, type Event, Producer, type Span } from "./index.ts"; @@ -16,7 +16,7 @@ class Live { producer: Producer; consumer: Consumer; events: Event[] = []; - #next?: Promise | undefined>; + #next?: Promise> | undefined>; constructor(config: { opRatio?: number; compression?: boolean; checkpointRecords?: number } = {}) { const track = new Track.Producer("test"); @@ -46,7 +46,7 @@ class Live { ]); if (event === idle || event === undefined) return; this.#next = undefined; - this.events.push(event); + this.events.push(event.value); } } @@ -292,7 +292,7 @@ test("consumer resumes at a checkpoint after losing a group", async () => { let group = track.appendGroup(); group.writeFrame({ payload: frame.payload, timestamp: Time.Timestamp.now() }); frame.commit(); - expect(await consumer.next()).toEqual({ push: { index: 0, value: { n: 0 } } }); + expect((await consumer.next())?.value).toEqual({ push: { index: 0, value: { n: 0 } } }); group.close(error); frame = encoder.push({ n: 1 }); @@ -302,7 +302,7 @@ test("consumer resumes at a checkpoint after losing a group", async () => { group.close(); track.close(); - expect(await consumer.next()).toEqual({ push: { index: 1, value: { n: 1 } } }); + expect((await consumer.next())?.value).toEqual({ push: { index: 1, value: { n: 1 } } }); expect(await consumer.next()).toBeUndefined(); } }); @@ -318,7 +318,7 @@ test("a fresh consumer adopts the current offset", async () => { const consumer = new Consumer({ track: subscriber }); producer.finish(); const events: Event[] = []; - for await (const event of consumer) events.push(event); + for await (const event of consumer) events.push(event.value); expect(events).toEqual([{ push: { index: 3, value: { n: 3 } } }, { push: { index: 4, value: { n: 4 } } }]); }); @@ -454,3 +454,52 @@ test("edits keep their timestamps, and an edit without one goes out untimed", as } } }); + +test("events carry their frame's timestamp, and a header's restated records get the header's", async () => { + const at = (ms: number) => Time.Timestamp.fromMillis(ms); + const millis = (event: Timed> | undefined) => event?.at?.as(Time.Timescale.MILLI); + const track = new Track.Producer("test").accept({ timescale: Time.Timescale.MILLI }); + const live = new Consumer({ track: track.subscribe() }); + const late = track.subscribe(); + const producer = new Producer({ track }); + + producer.push({ value: { n: 1 }, at: at(100) }); + producer.push({ value: { n: 2 }, at: at(200) }); + producer.pop(1, at(300)); + for (const [value, ms] of [ + [{ push: { index: 0, value: { n: 1 } } }, 100], + [{ push: { index: 1, value: { n: 2 } } }, 200], + [{ pop: { start: 0, end: 1 } }, 300], + ] as const) { + const event = await live.next(); + expect(event?.value).toEqual(value); + expect(millis(event)).toBe(ms); + } + + // A reader joining at a later group learns of the retained record from its header. + producer.cut(); + producer.push({ value: { n: 3 }, at: at(400) }); + producer.finish(); + late.setGroups({ start: { included: 1 } }); + const events: Timed>[] = []; + for await (const event of new Consumer({ track: late })) events.push(event); + expect(events.map((event) => event.value)).toEqual([ + { push: { index: 1, value: { n: 2 } } }, + { push: { index: 2, value: { n: 3 } } }, + ]); + expect(events.map(millis)).toEqual([400, 400]); +}); + +test("events on an untimed track have no timestamp", async () => { + const track = new Track.Producer("test"); + const consumer = new Consumer({ track: track.subscribe() }); + const producer = new Producer({ track }); + producer.push({ value: { n: 0 } }); + producer.pop(1); + producer.finish(); + + const events: Timed>[] = []; + for await (const event of consumer) events.push(event); + expect(events.length).toBe(2); + expect(events.every((event) => event.at === undefined)).toBeTrue(); +}); diff --git a/js/room/MIGRATE.md b/js/room/MIGRATE.md index f5d865cd37c..323ee407a1f 100644 --- a/js/room/MIGRATE.md +++ b/js/room/MIGRATE.md @@ -4,6 +4,11 @@ How to update your code for each breaking change, newest first. ## Unreleased +### Chat events carry their send time + +`Chat.Subscriber.recv()` returns `{ value, at }`: `value` is the event `recv()` used to return, and `at` is its frame's timestamp. +Read `(await subscriber.recv())?.value` where you used the event directly. + ### `HISTORY` is a `Nano` (#5706) `HISTORY` is a `Nano` from `@moq/clock` instead of milliseconds. diff --git a/js/room/README.md b/js/room/README.md index 6928799bfa6..f7945081810 100644 --- a/js/room/README.md +++ b/js/room/README.md @@ -74,7 +74,7 @@ const publisher = Chat.Publisher.create(broadcast); publisher.send("hello"); const subscriber = Chat.Subscriber.subscribe(broadcast.consume()); -const event = await subscriber.recv(); // push, pop, or skip +const event = await subscriber.recv(); // { value: push, pop, or skip, at: when it was sent } // Call publisher.finish() and subscriber.close() when done. ``` diff --git a/js/room/src/chat.test.ts b/js/room/src/chat.test.ts index fb41ff4a986..48e15d6fe0a 100644 --- a/js/room/src/chat.test.ts +++ b/js/room/src/chat.test.ts @@ -11,16 +11,19 @@ test("chat expires ten-second history and late readers only see retained message const subscriber = Subscriber.subscribe(broadcast.consume()); try { publisher.send("first"); - expect(await subscriber.recv()).toEqual({ push: { index: 0, value: "first" } }); + const first = await subscriber.recv(); + expect(first?.value).toEqual({ push: { index: 0, value: "first" } }); + // A sent message carries its send time. + expect(first?.at).toBeDefined(); await clock.advanceAsync(Nano(HISTORY - 1)); publisher.expire(); // The publisher's own timer retires the message once its history elapses. const popped = subscriber.recv(); await clock.advanceAsync(Nano(1)); - expect(await popped).toEqual({ pop: { start: 0, end: 1 } }); + expect((await popped)?.value).toEqual({ pop: { start: 0, end: 1 } }); publisher.send("second"); const late = Subscriber.subscribe(broadcast.consume()); - expect(await late.recv()).toEqual({ push: { index: 1, value: "second" } }); + expect((await late.recv())?.value).toEqual({ push: { index: 1, value: "second" } }); publisher.finish(); expect(await late.recv()).toBeUndefined(); late.close(); diff --git a/js/room/src/chat.ts b/js/room/src/chat.ts index d322d4905b8..944c137ec47 100644 --- a/js/room/src/chat.ts +++ b/js/room/src/chat.ts @@ -6,7 +6,7 @@ import { Instant, Nano, now, Second } from "@moq/clock"; import * as Json from "@moq/json"; -import { type Broadcast, Time, type Track } from "@moq/net"; +import { type Broadcast, Time, type Timed, type Track } from "@moq/net"; import { Effect, Signal } from "@moq/signals"; /** Name of the track carrying the chat window. */ @@ -92,16 +92,20 @@ export class Subscriber { }); } - /** Return the next window change, or undefined on clean completion; failures throw. */ - async recv(): Promise { + /** + * Return the next window change and when its frame was sent, or undefined on clean completion; + * failures throw. A message first seen in a later window's header carries that header's time. + */ + async recv(): Promise | undefined> { await this.#positioned; - const event = await this.#consumer.next(); - if (!event) return undefined; + const next = await this.#consumer.next(); + if (!next) return undefined; + const { value: event, at } = next; if ("push" in event) { if (typeof event.push.value !== "string") throw new Error("chat record must be a string"); - return { push: { index: event.push.index, value: event.push.value } }; + return { value: { push: { index: event.push.index, value: event.push.value } }, at }; } - return event; + return { value: event, at }; } /** Cancel the subscription and release buffered records. */ diff --git a/quest/m1/README.md b/quest/m1/README.md index 5e5e87c05b5..3fe1d01080a 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -85,7 +85,6 @@ lite-07 stream type here. - [More tests under load](/quest/m1/test-flakes-2/README.md) - the second round of load-only failures, one quest per flake, fixed at the cause - [Lazy discovery](/quest/m1/lazy-discovery/README.md) - a `consume` session requests announcements only for prefixes the origin is watching, in JS then Rust; `follow`, `broadcasts`, and the `discovery` option go away - [A public model](/quest/m1/public-model/README.md) - every model item a built-in session reaches is public, unnecessary, or marked deliberately private, so a custom publisher, subscriber, or relay builds from the public model alone -- [JSON window timestamps](/quest/m1/json-window-timestamps.md) - Rust and JS window consumers return each event's frame timestamp like the snapshot and stream consumers - [JavaScript FETCH](/quest/m1/js-fetch.md) - browser publishers answer IETF FETCH through the JS ranges request surface - [Report relays that refuse TRACK_STATUS](/quest/m1/track-status-race.md) - moqx and any other relay refusing TRACK_STATUS with an undefined code get an upstream report - [IETF fill timeout](/quest/m1/ietf-fill-timeout.md) - native FILL_TIMEOUT bounds upstream waiting on drafts 18-22, alongside local content-age enforcement diff --git a/quest/m1/json-window-timestamps.md b/quest/m1/json-window-timestamps.md deleted file mode 100644 index 962b76724a9..00000000000 --- a/quest/m1/json-window-timestamps.md +++ /dev/null @@ -1,27 +0,0 @@ -# [S] JSON window consumers return each event's timestamp - -## Goal - -The moq-json and `@moq/json` window consumers return each event with a -timestamp. The Rust and JS snapshot and stream consumers already return -`Timed` (#5099, #5677); the window consumer is the last one. - -## Plan - -Follow-up from #5099, which gave `@moq/json` and `@moq/flate` snapshot and -stream consumers `Timed`. The window consumer (`window::Consumer::next` in -`rs/moq-json` and `js/json`) still yields a bare `Event`. - -- Decided (2026-10-10): `next()` returns `Timed>`. `at` is the - timestamp of the frame that carried the event, and `None` when untimed. A - record restated by a group header gets the header frame's time; document that - on the type. There is no per-record time on the wire, so no wire change. -- Decided: Rust and JS together, same names and shape. Update moq-ffi and the - bindings if they expose the window consumer (see the Cross-Package Sync table). -- Update in-repo readers (`js/hang` timeline, `js/room` chat) and add a line to - `## Unreleased` of each affected package's `MIGRATE.md`. No new doc page. -- Land before the next `@moq/json` release, so its break ships together with - #5099's. The Rust break uses the same `moq_net::Timed` as #5677 and ships in - the next moq-json release with it. - -Public API: breaking (Rust and JS). Wire: none. diff --git a/quest/m1/refresh/README.md b/quest/m1/refresh/README.md index 4b8a4fd781f..d3d57d3dea1 100644 --- a/quest/m1/refresh/README.md +++ b/quest/m1/refresh/README.md @@ -68,4 +68,3 @@ Work owned by this README, after the children land: ## Related - [moq-transport passthrough](/quest/m1/ietf-extension/README.md) - its dynamic-groups child is the IETF half this epic promotes -- [JSON window timestamps](/quest/m1/json-window-timestamps.md) - touches the same window consumers diff --git a/quest/m1/refresh/json.md b/quest/m1/refresh/json.md index c499af0a2ab..a4f93b7b5d7 100644 --- a/quest/m1/refresh/json.md +++ b/quest/m1/refresh/json.md @@ -87,7 +87,3 @@ Public API: breaking (`delta_ratio` and `op_ratio` replaced by `deltas` and `ops` in Rust, JS, and every binding that exposes them; `Encoder::snapshot()` and Rust `Producer::refresh` / `poll_refresh` added; bindings gain no refresh method because moq-ffi drives it internally). Wire: none. - -## Related - -- [JSON window timestamps](/quest/m1/json-window-timestamps.md) - touches the same window code diff --git a/rs/moq-archive/src/reader/tests.rs b/rs/moq-archive/src/reader/tests.rs index d3bd64b6adb..150d9d0c25e 100644 --- a/rs/moq-archive/src/reader/tests.rs +++ b/rs/moq-archive/src/reader/tests.rs @@ -694,7 +694,7 @@ async fn timeline_tracks_are_republished_and_finished_on_request() { let mut records = Vec::new(); while let Some(event) = timeline.next().await.unwrap() { - if let window::Event::Push { value, .. } = event { + if let window::Event::Push { value, .. } = event.value { records.push(value.sequence); } } diff --git a/rs/moq-json/MIGRATE.md b/rs/moq-json/MIGRATE.md index 9eeae6499ab..d888087cec3 100644 --- a/rs/moq-json/MIGRATE.md +++ b/rs/moq-json/MIGRATE.md @@ -6,7 +6,8 @@ How to update your code for each breaking change, newest first. ### Consumers return `Timed` (#5677) -Snapshot and stream `next` / `poll_next` return `moq_net::Timed`: read `.value` for the value and `.at` for the frame's exact timestamp, `None` when untimed. +Snapshot, stream, and window `next` / `poll_next` return `moq_net::Timed`: read `.value` for the value and `.at` for the frame's exact timestamp, `None` when untimed. +A window event's `.value` is the `Event` that `next` used to return; a record restated by a group header carries the header frame's time. Snapshot `next` now yields every state in order inside the subscription's `max_delay`; call `latest` / `poll_latest` for the old collapse-to-newest read. ### Duplicate keys are refused diff --git a/rs/moq-json/src/window/consumer.rs b/rs/moq-json/src/window/consumer.rs index 75819be091f..0332c77b91d 100644 --- a/rs/moq-json/src/window/consumer.rs +++ b/rs/moq-json/src/window/consumer.rs @@ -17,11 +17,17 @@ use crate::Result; /// Group rolls never surface. A publisher rolls for compression's sake, and a header restating the /// window yields nothing for records already delivered, so this reads as one continuous stream of /// [`Event`]s regardless of how the publisher framed them. +/// +/// Each event carries the timestamp of the frame that produced it, `None` on an untimed track. The +/// wire has no per-record time, so a record a reader first learns of from a group header (one it +/// joined after, or missed while behind) gets the header frame's time, not its original push time. pub struct Consumer { track: moq_net::track::Ordered, group: Option, codec: Option, decoder: Decoder, + /// Timestamp of the frame whose events the decoder is draining. + at: Option, } impl Consumer { @@ -32,6 +38,7 @@ impl Consumer { group: None, codec: None, decoder: Decoder::new(config), + at: None, } } @@ -40,8 +47,8 @@ impl Consumer { self.decoder.range() } - /// Get the next event, or `None` once the track ends. - pub async fn next(&mut self) -> Result>> + /// Get the next event and its frame's timestamp, or `None` once the track ends. + pub async fn next(&mut self) -> Result>>> where T: Unpin, { @@ -49,11 +56,12 @@ impl Consumer { } /// Poll for the next event, without blocking. - pub fn poll_next(&mut self, waiter: &kio::Waiter) -> Poll>>> { + pub fn poll_next(&mut self, waiter: &kio::Waiter) -> Poll>>>> { loop { - // Drain what the frames already decoded produced before reading more. - if let Some(event) = self.decoder.next_event() { - return Poll::Ready(Ok(Some(event))); + // Drain what the last frame produced before reading more, so every queued event is that + // frame's and shares its timestamp. + if let Some(value) = self.decoder.next_event() { + return Poll::Ready(Ok(Some(moq_net::Timed { value, at: self.at }))); } let Some(group) = &mut self.group else { @@ -89,6 +97,7 @@ impl Consumer { Poll::Ready(Ok(Some(frame))) => { let codec = self.codec.as_mut().expect("an open MoQ group has a window codec"); self.decoder.decode(codec, &frame.payload)?; + self.at = frame.timestamp; } Poll::Ready(Ok(None)) => { // This group is exhausted. Clear it and poll for a later one, which restates the diff --git a/rs/moq-json/src/window/mod.rs b/rs/moq-json/src/window/mod.rs index b0b5f1d012a..bc28c879e27 100644 --- a/rs/moq-json/src/window/mod.rs +++ b/rs/moq-json/src/window/mod.rs @@ -120,6 +120,80 @@ mod test { assert_eq!(pushed, vec![rec(1), rec(2)]); } + #[test] + fn events_carry_their_frames_timestamp() { + let track = moq_net::broadcast::Info::new() + .produce() + .create_track("test", timed()) + .unwrap(); + let mut live = consumer(track.subscribe(None), false); + let mut late = track.subscribe(None); + let mut producer = Producer::new(track, ProducerConfig::default()); + let at = |ms| moq_net::Timestamp::from_millis(ms).unwrap(); + + producer.push(moq_net::Timed::from(&rec(1)).at(at(100))).unwrap(); + producer.push(moq_net::Timed::from(&rec(2)).at(at(200))).unwrap(); + producer.pop(moq_net::Timed::from(1u64).at(at(300))).unwrap(); + let events: Vec<_> = drain_timed(&mut live).into_iter().map(|e| (e.value, e.at)).collect(); + assert_eq!( + events, + vec![ + ( + Event::Push { + index: 0, + value: rec(1) + }, + Some(at(100)) + ), + ( + Event::Push { + index: 1, + value: rec(2) + }, + Some(at(200)) + ), + (Event::Pop(0..1), Some(at(300))), + ] + ); + + // A reader joining at a later group learns of the retained record from its header, so it + // gets the header frame's time rather than the record's original push time. + producer.cut().unwrap(); + producer.push(moq_net::Timed::from(&rec(3)).at(at(400))).unwrap(); + late.set_groups(1..); + let mut late = consumer(late, false); + let events: Vec<_> = drain_timed(&mut late).into_iter().map(|e| (e.value, e.at)).collect(); + assert_eq!( + events, + vec![ + ( + Event::Push { + index: 1, + value: rec(2) + }, + Some(at(400)) + ), + ( + Event::Push { + index: 2, + value: rec(3) + }, + Some(at(400)) + ), + ] + ); + } + + #[test] + fn untimed_events_have_no_timestamp() { + let (mut producer, track) = producer(ProducerConfig::default()); + producer.push(&rec(1)).unwrap(); + producer.pop(1).unwrap(); + let events = drain_timed(&mut consumer(track, false)); + assert_eq!(events.len(), 2); + assert!(events.iter().all(|event| event.at.is_none())); + } + #[test] fn a_bare_publish_stays_untimed() { let track = moq_net::broadcast::Info::new() @@ -187,6 +261,11 @@ mod test { /// Drain every event currently available without blocking. fn drain(consumer: &mut Consumer) -> Vec> { + drain_timed(consumer).into_iter().map(|event| event.value).collect() + } + + /// Drain every event currently available, with its frame timestamp. + fn drain_timed(consumer: &mut Consumer) -> Vec>> { let waiter = kio::Waiter::noop(); let mut out = Vec::new(); while let Poll::Ready(Ok(Some(event))) = consumer.poll_next(&waiter) { diff --git a/rs/moq-mux/src/timeline.rs b/rs/moq-mux/src/timeline.rs index 4fed58ca324..4015446a65c 100644 --- a/rs/moq-mux/src/timeline.rs +++ b/rs/moq-mux/src/timeline.rs @@ -864,7 +864,7 @@ impl Consumer { /// Get the next window event, or `None` once the track ends. pub async fn next(&mut self) -> crate::Result>> { match self.inner.next().await? { - Some(event) => Ok(Some(self.decode_event(event)?)), + Some(event) => Ok(Some(self.decode_event(event.value)?)), None => Ok(None), } } @@ -872,7 +872,8 @@ impl Consumer { /// Poll for the next window event, without blocking. pub fn poll_next(&mut self, waiter: &kio::Waiter) -> Poll>>> { match self.inner.poll_next(waiter)? { - Poll::Ready(Some(event)) => Poll::Ready(self.decode_event(event).map(Some)), + // Each record carries its own pts, so the frame timestamp adds nothing. + Poll::Ready(Some(event)) => Poll::Ready(self.decode_event(event.value).map(Some)), Poll::Ready(None) => Poll::Ready(Ok(None)), Poll::Pending => Poll::Pending, } diff --git a/rs/moq-room/MIGRATE.md b/rs/moq-room/MIGRATE.md index fab249cc52d..de8cc3119cd 100644 --- a/rs/moq-room/MIGRATE.md +++ b/rs/moq-room/MIGRATE.md @@ -3,3 +3,7 @@ How to update your code for each breaking change, newest first. ## Unreleased + +### Chat events carry their send time + +`chat::Subscriber::recv` / `poll_recv` return `moq_net::Timed`: `.value` is the event they used to return, and `.at` is its frame's timestamp. diff --git a/rs/moq-room/src/chat.rs b/rs/moq-room/src/chat.rs index 22ebb790c6c..0c7efb389e6 100644 --- a/rs/moq-room/src/chat.rs +++ b/rs/moq-room/src/chat.rs @@ -151,8 +151,9 @@ impl Subscriber { } } - /// Poll the next window change, returning errors separately from clean completion. - pub fn poll_recv(&mut self, waiter: &kio::Waiter) -> Poll, Error>> { + /// Poll the next window change and when its frame was sent, returning errors separately from + /// clean completion. + pub fn poll_recv(&mut self, waiter: &kio::Waiter) -> Poll>, Error>> { if let State::Positioning(pending) = &mut self.state { let track = pending.as_mut().expect("positioning holds the track"); // The newest window is known only once the feed is live: a relay copy still waiting @@ -170,8 +171,10 @@ impl Subscriber { Poll::Ready(Ok(ready!(consumer.poll_next(waiter))?)) } - /// Wait for the next window change, or None on clean completion. - pub async fn recv(&mut self) -> Result, Error> { + /// Wait for the next window change and when its frame was sent, or None on clean completion. + /// + /// A message first seen in a later window's header carries that header's time, not its send time. + pub async fn recv(&mut self) -> Result>, Error> { kio::wait(|waiter| self.poll_recv(waiter)).await } } @@ -187,19 +190,24 @@ mod tests { let mut publisher = Publisher::create(&mut broadcast).unwrap(); let mut subscriber = Subscriber::subscribe(&broadcast.consume()).await.unwrap(); publisher.send("first").unwrap(); + let first = subscriber.recv().await.unwrap().unwrap(); + assert!(first.at.is_some(), "a sent message carries its send time"); assert_eq!( - subscriber.recv().await.unwrap(), - Some(Event::Push { + first.value, + Event::Push { index: 0, value: "first".into() - }) + } ); publisher.expire().await.unwrap(); - assert_eq!(subscriber.recv().await.unwrap(), Some(Event::Pop(0..1))); + assert_eq!( + subscriber.recv().await.unwrap().map(|e| e.value), + Some(Event::Pop(0..1)) + ); publisher.send("second").unwrap(); let mut late = Subscriber::subscribe(&broadcast.consume()).await.unwrap(); assert_eq!( - late.recv().await.unwrap(), + late.recv().await.unwrap().map(|e| e.value), Some(Event::Push { index: 1, value: "second".into() @@ -292,11 +300,14 @@ mod tests { { let mut warm = Subscriber::subscribe(&remote).await.unwrap(); assert_eq!( - warm.recv().await.unwrap(), - Some(Event::Push { - index: 0, - value: "first".into() - }) + warm.recv().await.unwrap().map(|e| (e.value, e.at)), + Some(( + Event::Push { + index: 0, + value: "first".into() + }, + Some(at(0)) + )) ); } // The relay cancels the unread upstream subscription; its copy keeps the window. @@ -308,11 +319,14 @@ mod tests { let subscription = track::Subscription::default().with_max_delay(Duration::from_secs(10)); let mut late = Subscriber::new(remote.track(TRACK).unwrap().subscribe(subscription).await.unwrap()); assert_eq!( - late.recv().await.unwrap(), - Some(Event::Push { - index: 1, - value: "second".into() - }), + late.recv().await.unwrap().map(|e| (e.value, e.at)), + Some(( + Event::Push { + index: 1, + value: "second".into() + }, + Some(at(1)) + )), "the warm copy's backlog was replayed" ); } From 087bccc83a2ea9f0755f73268ab44d1706e36b83 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 21 Oct 2026 23:26:47 -0700 Subject: [PATCH 3/3] docs(room): note chat recv returns the send time Co-Authored-By: Claude Opus 5.5 --- rs/moq-room/README.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/rs/moq-room/README.md b/rs/moq-room/README.md index 98718b40ced..6b25df11ec2 100644 --- a/rs/moq-room/README.md +++ b/rs/moq-room/README.md @@ -29,4 +29,4 @@ while let Some(event) = room.next().await { Chat uses `moq-json::window` with ten seconds of history. Drive `Publisher::expire()` alongside incoming sends (or call `poll_expire` from a poll loop) so messages expire while idle. `Subscriber::recv` returns push/pop/skip -events and propagates errors. The format differs from iroh-live's raw UTF-8 chat. +events with their send time and propagates errors. The format differs from iroh-live's raw UTF-8 chat.