feat(json)!: window consumers return each event's frame timestamp

js/hang/src/timeline.ts

@@ -85,7 +85,8 @@ export class Consumer implements AsyncIterable<Event> {
 
 	/** Get the next timeline event, or `undefined` when the track ends. Throws on an untimed record. */
 	async next(): Promise<Event | undefined> {
-		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;

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.

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<T> {
@@ -34,6 +38,8 @@ export class Consumer<T> {
 
 	#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,20 +52,21 @@ export class Consumer<T> {
 		return this.#decoder.offset;
 	}
 
-	/** Get the next event, or `undefined` once the track ends. */
-	next(): Promise<Event<T> | undefined> {
+	/** Get the next event and its frame's timestamp, or `undefined` once the track ends. */
+	next(): Promise<Moq.Timed<Event<T>> | undefined> {
 		if (this.#reading) throw new Error("multiple calls to next not supported");
 		this.#reading = true;
 		return this.#read().finally(() => {
 			this.#reading = false;
 		});
 	}
 
-	async #read(): Promise<Event<T> | undefined> {
+	async #read(): Promise<Moq.Timed<Event<T>> | 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<T> {
 
 			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<Event<T>> {
+	async *[Symbol.asyncIterator](): AsyncIterator<Moq.Timed<Event<T>>> {
 		for (;;) {
 			const event = await this.next();
 			if (event === undefined) return;

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<Rec>;
 	consumer: Consumer<Rec>;
 	events: Event<Rec>[] = [];
-	#next?: Promise<Event<Rec> | undefined>;
+	#next?: Promise<Timed<Event<Rec>> | 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<Rec>({ track: subscriber });
 	producer.finish();
 	const events: Event<Rec>[] = [];
-	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<Event<Rec>> | undefined) => event?.at?.as(Time.Timescale.MILLI);
+	const track = new Track.Producer("test").accept({ timescale: Time.Timescale.MILLI });
+	const live = new Consumer<Rec>({ track: track.subscribe() });
+	const late = track.subscribe();
+	const producer = new Producer<Rec>({ 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<Event<Rec>>[] = [];
+	for await (const event of new Consumer<Rec>({ 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<Rec>({ track: track.subscribe() });
+	const producer = new Producer<Rec>({ track });
+	producer.push({ value: { n: 0 } });
+	producer.pop(1);
+	producer.finish();
+
+	const events: Timed<Event<Rec>>[] = [];
+	for await (const event of consumer) events.push(event);
+	expect(events.length).toBe(2);
+	expect(events.every((event) => event.at === undefined)).toBeTrue();
+});

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.

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.
 ```
 

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();

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<Event | undefined> {
+	/**
+	 * 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<Timed<Event> | 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. */

quest/m1/README.md

@@ -84,7 +84,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
 - [FETCH_OK whole group](/quest/m1/ietf-fetch-end-location.md) - on drafts 14 to 19, FETCH_OK End Object 0 means the whole group in both directions, and a FETCH ending at the maximum group is refused as unsupported
 - [Pause an IETF SUBSCRIBE](/quest/m1/ietf-subscribe-pause.md) - Forward=0 pauses a SUBSCRIBE without demand upstream and Forward=1 resumes it, instead of ending it
 - [JavaScript FETCH](/quest/m1/js-fetch.md) - browser publishers answer IETF FETCH through the JS ranges request surface

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

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

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);
 		}
 	}

rs/moq-json/MIGRATE.md

@@ -6,7 +6,8 @@ How to update your code for each breaking change, newest first.
 
 ### Consumers return `Timed<T>` (#5677)
 
-Snapshot and stream `next` / `poll_next` return `moq_net::Timed<T>`: 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<T>`: 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

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<T> {
 	track: moq_net::track::Ordered,
 	group: Option<moq_net::group::Consumer>,
 	codec: Option<Codec>,
 	decoder: Decoder<T>,
+	/// Timestamp of the frame whose events the decoder is draining.
+	at: Option<moq_net::Timestamp>,
 }
 
 impl<T: DeserializeOwned> Consumer<T> {
@@ -32,6 +38,7 @@ impl<T: DeserializeOwned> Consumer<T> {
 			group: None,
 			codec: None,
 			decoder: Decoder::new(config),
+			at: None,
 		}
 	}
 
@@ -40,20 +47,21 @@ impl<T: DeserializeOwned> Consumer<T> {
 		self.decoder.range()
 	}
 
-	/// Get the next event, or `None` once the track ends.
-	pub async fn next(&mut self) -> Result<Option<Event<T>>>
+	/// Get the next event and its frame's timestamp, or `None` once the track ends.
+	pub async fn next(&mut self) -> Result<Option<moq_net::Timed<Event<T>>>>
 	where
 		T: Unpin,
 	{
 		kio::wait(|waiter| self.poll_next(waiter)).await
 	}
 
 	/// Poll for the next event, without blocking.
-	pub fn poll_next(&mut self, waiter: &kio::Waiter) -> Poll<Result<Option<Event<T>>>> {
+	pub fn poll_next(&mut self, waiter: &kio::Waiter) -> Poll<Result<Option<moq_net::Timed<Event<T>>>>> {
 		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<T: DeserializeOwned> Consumer<T> {
 				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