fix(net): retain the highest group PTS

bench/README.md

@@ -376,6 +376,12 @@ and group setup stay outside drift timing; each parked read checks its live
 payload. `ledger` uses one fixture against every checkout's internal ledger, sweeping
 ascending, descending, shuffled, and rolling-expiration arrivals over the retained
 gap count. It includes each arrival's expiration and coverage checks.
+The drift sweep includes untimed controls and timed groups with decode-order
+B-frames, independently varying publisher and reader counts and retained history.
+Its `time_mode` column is 0 for untimed, 1 for millisecond timestamps, and 2
+for mixed millisecond/microsecond timestamps, which exercise exact cross-scale
+comparison.
+
 The browser smoke lane preserves one measured round of all three fixtures, and drift
 and ledger also run in the nightly JS benchmark leg. Timing is
 observational, with the A/A spread showing the host's noise floor.

bench/browser/micro.ts

@@ -109,7 +109,8 @@ try {
 			rows.set(variant, []);
 			await pages.get(variant)?.evaluate("globalThis.measure()");
 			const measured = rows.get(variant);
-			if (measured?.length !== (benchmark === "frames" ? 26 : 12)) throw new Error(`incomplete rows: ${variant}`);
+			if (measured?.length !== (benchmark === "frames" ? 26 : benchmark === "drift" ? 36 : 12))
+				throw new Error(`incomplete rows: ${variant}`);
 			samples.push({ round, variant, rows: measured });
 			await Bun.write(
 				values.out,

js/net/bench/drift.ts

@@ -1,63 +1,106 @@
 /** Sweep publishers, readers, and retained history for a parked frame read's drift check. */
 import { Manual, Milli } from "@moq/clock";
-import { type Group, Track } from "@moq/net";
+import { type Group, Time, Track } from "@moq/net";
 
 /** Read live frames after a retained window, checking every delivered payload. */
 export async function run(): Promise<void> {
 	const clock = new Manual();
 	const installed = clock.install();
 	const payload = new Uint8Array([7]);
+	const iterations = 2000;
+	const warmup = 20;
+	const offsets = [0, 80, 40];
 	try {
-		console.log("retained,publishers,readers,ns_per_frame");
+		// The same reordered-PTS verdict runs here in Chromium before timing any work.
+		const check = new Track.Producer("reordered").accept({ timescale: Time.Timescale.MILLI });
+		const reader = check.subscribe({ maxDelay: Milli(80) });
+		try {
+			for (const timestamps of [[0], [100], [150, 200, 160]]) {
+				const group = check.appendGroup();
+				for (const ms of timestamps) group.writeFrame({ payload, timestamp: Time.Timestamp.fromMillis(ms) });
+				group.finish();
+			}
+			if ((await reader.recvGroup())?.sequence !== 1) throw new Error("B-frame tail rewound the drift edge");
+		} finally {
+			reader.close();
+			check.finish();
+			check.close();
+		}
+		console.log("time_mode,retained,publishers,readers,ns_per_frame");
 		for (const retained of [1, 64, 1024])
 			for (const publishers of [1, 4])
-				for (const readers of [1, 8]) {
-					const tracks = Array.from({ length: publishers }, (_, i) => new Track.Producer(`p${i}`).accept({}));
-					const sources: Group.Producer[] = [];
-					const consumers: Group.Consumer[] = [];
-					try {
-						for (const track of tracks) {
-							for (let sequence = 0; sequence < retained; sequence++) {
-								const group = track.appendGroup();
-								group.writeFrame({ payload });
-								group.finish();
-							}
-							const subscribers = Array.from({ length: readers }, () =>
-								track.subscribe({ maxDelay: Milli(60_000) }),
-							);
-							// Drain replay through the normal ordered cursor while retaining its timeline.
-							const ordered = subscribers.map((sub) => sub.ordered());
-							for (let sequence = 0; sequence < retained; sequence++)
+				for (const readers of [1, 8])
+					for (const mode of [0, 1, 2]) {
+						const stamp = (ms: number, index: number) =>
+							mode === 0
+								? undefined
+								: mode === 2 && index % 2 !== 0
+									? Time.Timestamp.fromMicros(ms * 1000)
+									: Time.Timestamp.fromMillis(ms);
+						const tracks = Array.from({ length: publishers }, (_, i) =>
+							new Track.Producer(`p${i}`).accept(mode !== 0 ? { timescale: Time.Timescale.MILLI } : {}),
+						);
+						const sources: Group.Producer[] = [];
+						const consumers: Group.Consumer[] = [];
+						try {
+							for (const track of tracks) {
+								for (let sequence = 0; sequence < retained; sequence++) {
+									const group = track.appendGroup();
+									group.writeFrame({
+										payload,
+										timestamp: stamp(sequence * 100, sequence),
+									});
+									group.finish();
+								}
+								const subscribers = Array.from({ length: readers }, () =>
+									track.subscribe({
+										maxDelay: Milli(
+											(retained + Math.ceil((iterations + warmup) / offsets.length) + 1) * 100,
+										),
+									}),
+								);
+								// Drain replay through the normal ordered cursor while retaining its timeline.
+								const ordered = subscribers.map((sub) => sub.ordered());
+								for (let sequence = 0; sequence < retained; sequence++)
+									for (const reader of ordered) {
+										const group = await reader.nextGroup();
+										if (group?.sequence !== sequence) throw new Error("history missing");
+										await group.readFrame();
+									}
+								sources.push(track.appendGroup());
 								for (const reader of ordered) {
 									const group = await reader.nextGroup();
-									if (group?.sequence !== sequence) throw new Error("history missing");
-									await group.readFrame();
+									if (!group) throw new Error("live group missing");
+									consumers.push(group);
 								}
-							sources.push(track.appendGroup());
-							for (const reader of ordered) {
-								const group = await reader.nextGroup();
-								if (!group) throw new Error("live group missing");
-								consumers.push(group);
 							}
+							let start = 0;
+							for (let iteration = 0; iteration < iterations + warmup; iteration++) {
+								if (iteration === warmup) start = performance.now();
+								const pending = consumers.map((consumer) => consumer.readFrame());
+								const timestamp = stamp(
+									retained * 100 +
+										Math.floor(iteration / offsets.length) * 100 +
+										offsets[iteration % offsets.length],
+									iteration,
+								);
+								for (const source of sources) source.writeFrame({ payload, timestamp });
+								for (const frame of await Promise.all(pending))
+									if (
+										frame?.payload[0] !== 7 ||
+										frame.timestamp?.asMillis() !== timestamp?.asMillis()
+									)
+										throw new Error("frame missing or reordered");
+							}
+							const elapsed = performance.now() - start;
+							console.log(
+								`${mode},${retained},${publishers},${readers},${((elapsed * 1e6) / (iterations * publishers * readers)).toFixed(0)}`,
+							);
+						} finally {
+							for (const track of tracks) track.finish();
+							for (const track of tracks) track.close();
 						}
-						const iterations = 2000;
-						let start = 0;
-						for (let iteration = 0; iteration < iterations + 20; iteration++) {
-							if (iteration === 20) start = performance.now();
-							const pending = consumers.map((consumer) => consumer.readFrame());
-							for (const source of sources) source.writeFrame({ payload });
-							for (const frame of await Promise.all(pending))
-								if (frame?.payload[0] !== 7) throw new Error("frame missing");
-						}
-						const elapsed = performance.now() - start;
-						console.log(
-							`${retained},${publishers},${readers},${((elapsed * 1e6) / (iterations * publishers * readers)).toFixed(0)}`,
-						);
-					} finally {
-						for (const track of tracks) track.finish();
-						for (const track of tracks) track.close();
 					}
-				}
 	} finally {
 		installed.close();
 	}

js/net/src/group.test.ts

@@ -1,4 +1,5 @@
 import { expect, test } from "bun:test";
+import { Manual } from "@moq/clock";
 import { Signal } from "@moq/signals";
 import { FrameTooLarge, GroupTooLarge, StreamCode, StreamError } from "./error.ts";
 import { MAX_GROUP_CACHE_BYTES, MAX_GROUP_FRAMES, Producer } from "./group.ts";
@@ -8,6 +9,47 @@ import { Timestamp } from "./time.ts";
 
 const dec = new TextDecoder();
 
+test("reordered frames keep the highest group PTS in live and later mirrors", () => {
+	const installed = new Manual().install();
+	try {
+		const producer = new Producer(0);
+		const shared = producer.consume();
+		const live = producer.mirror();
+		for (const timestamp of [Timestamp.fromMillis(0), Timestamp.fromMicros(80_000), Timestamp.fromMillis(40)])
+			producer.writeFrame({ payload: new Uint8Array([1]), timestamp });
+		expect(hooks.groupTimestamp(shared)?.asMillis()).toBe(0);
+		expect(hooks.groupLatest(shared)?.asMillis()).toBe(80);
+		expect(hooks.groupLatest(live)?.asMillis()).toBe(80);
+		// A later mirror only replays the remaining B-frame, but keeps the known high PTS.
+		shared.tryReadFrame();
+		shared.tryReadFrame();
+		const later = producer.mirror();
+		expect(hooks.groupLatest(later)?.asMillis()).toBe(80);
+		expect(later.tryReadFrame()?.timestamp?.asMillis()).toBe(40);
+		for (const reader of [shared, live, later]) reader.close();
+		producer.finish();
+	} finally {
+		installed.close();
+	}
+});
+
+test("zero group PTS keeps its scale and untimed groups stay unstamped", () => {
+	const installed = new Manual().install();
+	try {
+		for (const timestamp of [undefined, Timestamp.fromMicros(0)]) {
+			const producer = new Producer(0);
+			const consumer = producer.consume();
+			expect(hooks.groupLatest(consumer)).toBeUndefined();
+			producer.writeFrame({ payload: new Uint8Array(), timestamp });
+			expect(hooks.groupLatest(consumer)).toBe(timestamp);
+			consumer.close();
+			producer.finish();
+		}
+	} finally {
+		installed.close();
+	}
+});
+
 test("used reflects mirror demand and unused resolves when the last reader leaves", async () => {
 	const producer = new Producer(0);
 

js/net/src/group.ts

@@ -78,7 +78,7 @@ class GroupState {
 	// The first frame's timestamp, retained after reads. Media time only: an untimed group
 	// never has one, so it says nothing about whether a frame was written.
 	timestamp?: Timestamp;
-	// The newest frame's timestamp: where a reader that has taken every frame sits.
+	// The highest frame PTS, including when the decode-order tail presents earlier.
 	latest?: Timestamp;
 	// Accepted below the track's eviction watermark, where nothing proves where it starts,
 	// so this mirror drives no media-time expiry.
@@ -104,7 +104,8 @@ function appendFrame(state: GroupState, frame: Frame) {
 	if (state.closed.peek() !== undefined) throw new Error("group is closed");
 
 	state.timestamp ??= frame.timestamp;
-	state.latest = frame.timestamp;
+	if (frame.timestamp === undefined) state.latest = undefined;
+	else if (state.latest === undefined || state.latest.isBefore(frame.timestamp)) state.latest = frame.timestamp;
 	state.cacheBytes += stored(frame)?.payload.length ?? frame.payload.byteLength;
 	state.frames.mutate((frames) => {
 		frames.push(frame);

js/net/src/track.test.ts

@@ -582,6 +582,31 @@ test("nextGroup returns buffered groups in sequence", async () => {
 	expect((await track.nextGroup())?.sequence).toBe(5);
 });
 
+test.each([false, true])("a reordered tail keeps the group PTS drift verdict (B-frame: %s)", async (reordered) => {
+	const clock = manualTime(0);
+	try {
+		const producer = new TrackProducer("test").accept({ timescale: Timescale.MILLI });
+		const subscriber = producer.subscribe({ maxDelay: Milli(80) });
+		for (const ms of [0, 100]) {
+			const group = producer.appendGroup();
+			group.writeFrame({ payload: new Uint8Array(), timestamp: Timestamp.fromMillis(ms) });
+			group.finish();
+		}
+		const edge = producer.appendGroup();
+		for (const ms of [150, 200])
+			edge.writeFrame({ payload: new Uint8Array(), timestamp: Timestamp.fromMillis(ms) });
+		if (reordered) edge.writeFrame({ payload: new Uint8Array(), timestamp: Timestamp.fromMillis(160) });
+		edge.finish();
+		// Group 0 reaches 100ms and is 100ms behind the highest PTS; group 1 reaches 150ms.
+		expect((await subscriber.recvGroup())?.sequence).toBe(1);
+		subscriber.close();
+		producer.finish();
+		producer.close();
+	} finally {
+		clock.installed.close();
+	}
+});
+
 // A group's age is where its content *ends*, not where it began. A long group whose tail
 // is level with the live edge still owes the reader every frame in it, so judging it by
 // its first timestamp would discard exactly the group being filled.

quest/m1/README.md

@@ -114,7 +114,6 @@ surface and zero-copy texture import, land before the cut.
 
 - [Resume checked against a model](/quest/m1/resume-model-test.md) - a randomized test checks every resumed reader against a one-track reference model in Rust and JS, and passes before the cut
 - [Address groups by timestamp](/quest/m1/time-addressing/README.md) - group start timestamps replace sequence numbers in the moq-net API and lite-07 wire, while old subscribers, old publishers, and published JS clients keep working
-- [A group ends at its highest PTS](/quest/m1/group-end-pts.md) - a B-frame tail no longer understates a group's end, in Rust and JS
 - [Release audit findings](/quest/m1/release-audit/README.md) - resolve newly identified resource, parser, playback, and interoperability risks before the cut
 - [Pre-cut audit findings](/quest/m1/cut-audit/README.md) - fix the crashes, hostile-input costs, and correctness bugs the 2026-10-24 audit of main found
 - [Cancelled moq-transport requests stop](/quest/m1/ietf-request-cancel.md) - a cancelled request releases its state even when the peer withholds credit