feat(json): a public snapshot group budget, refusing an oversized snapshot before parsing

.github/workflows/nightly.yml

@@ -355,6 +355,7 @@ jobs:
         # viewer: diff-independent, since a break can arrive through the relay,
         # js/net, or the wire between them.
         # `rs quic-sim` sweeps moq-quic's simulated workloads from a fresh seed; PRs run a fixed 64.
+        # `test resume-model` sweeps resume against its model from a fresh seed; PRs run a fixed 64.
         # It also runs the bandwidth probe's acceptance suite, of which PRs run the short cases.
         # `test drill impaired` runs the relay drills over a freshly seeded shaper; PRs run loopback.
         # `rs cli-export` runs moq-cli's export process tests, which take tens of seconds of real time.
@@ -364,7 +365,7 @@ jobs:
         # `rs multicast` runs the mDNS tests, which loop multicast back on the runner.
         # `rs render-sw` renders through wgpu on Mesa's software Vulkan driver.
         # `test qmux` runs js/qmux against the in-tree Rust qmux, per draft.
-        recipe: ["gate", "rs doctest --workspace", "rs loom", "test drill-sensitivity", "rs uring", "test drain", "test json-properties", "rs quic-sim", "test drill impaired", "rs cli-export", "rs quic-stress", "rs audio-devices", "rs multicast", "rs render-sw", "test qmux"]
+        recipe: ["gate", "rs doctest --workspace", "rs loom", "test drill-sensitivity", "rs uring", "test drain", "test json-properties", "test resume-model", "rs quic-sim", "test drill impaired", "rs cli-export", "rs quic-stress", "rs audio-devices", "rs multicast", "rs render-sw", "test qmux"]
         include:
           - recipe: rs send-depth
             nightly: true

.github/workflows/smoke.yml

@@ -559,8 +559,9 @@ jobs:
           path: runner-upstream
           persist-credentials: false
 
-      # Any failed pair or setup step fails the job. A complete run still writes
-      # its report first, since a failing run is the one worth looking at.
+      # Only moq-dev clients against this checkout's relay gate the job, plus any
+      # setup step; the other pairs are reported only. A complete run still
+      # writes its report first, since a failing run is the one worth looking at.
       - name: Run upstream Docker matrices
         run: just runner "$GITHUB_WORKSPACE/runner-upstream" "$GITHUB_WORKSPACE/interop-runner-report"
 

AGENTS.md

@@ -102,7 +102,7 @@ just fix          # Auto-fix lint/formatting, same scope
 These diff the branch against its base and only run the affected packages.
 
 Quests: before starting work, run `quest guide` and follow it.
-The `quest` binary comes from the kixelated/quest flake input and serves the quest skills; change them upstream and bump the input.
+The `quest` binary comes from the moq-dev/quest flake input and serves the quest skills; change them upstream and bump the input.
 
 # Cross-Package Sync
 

bench/README.md

@@ -286,7 +286,7 @@ process CPU (user + system, every thread) per MB into
 budget (default 1); every received frame is checked byte for byte.
 
 The binding benchmark itself lives in
-[rust2](https://github.com/kixelated/rust2/blob/main/bench/ffi/README.md), with
+[rust2](https://github.com/moq-dev/rust2/blob/main/bench/ffi/README.md), with
 the UniFFI fork whose changes it measures: its `just bench-ffi` runs these
 runners nightly at a pinned moq revision, against the fork, and reports each
 wrapper's ratio to Rust. So that they keep working here, moq's Nightly runs every

flake.nix

@@ -32,7 +32,7 @@
     # The quest CLI, which also serves the quest guide and skills the stubs in
     # .claude/skills call. Bump the rev to upgrade them.
     quest = {
-      url = "github:kixelated/quest/01d051cf5dca8cd29c664bf8bcbcab63cdc76445";
+      url = "github:moq-dev/quest/01d051cf5dca8cd29c664bf8bcbcab63cdc76445";
       inputs.nixpkgs.follows = "nixpkgs";
       inputs.flake-utils.follows = "flake-utils";
     };

js/json/api/index.api.md

@@ -23,6 +23,7 @@ export namespace Snapshot {
 	export interface Config<T> {
 		deltas?: boolean;
 		initial?: T;
+		maxGroupBytes?: number;
 		schema?: ZodMiniType<T>;
 	}
 	export namespace Consumer {

js/json/src/snapshot/encoder.ts

@@ -1,4 +1,4 @@
-import { Group } from "@moq/net";
+import { Group, Error as NetError } from "@moq/net";
 import type * as z from "@zod/mini";
 
 import { deepEqual, diff } from "../diff.ts";
@@ -30,10 +30,12 @@ export interface Config<T> {
 	initial?: T;
 
 	/**
-	 * Bytes a group may hold before it rolls, defaulting to moq-net's per-group cache limit. Lets a
-	 * test reach the limit without megabytes of JSON. Must fit the snapshot's own frame.
+	 * The most bytes a group may hold, snapshot and deltas together.
 	 *
-	 * @internal
+	 * A delta that would pass it rolls a fresh snapshot instead, and a snapshot past it throws
+	 * `FrameTooLarge`. Lowering it lets a test reach the limit without megabytes of JSON; a value
+	 * above moq-net's per-group cache limit is clamped to it, since moq-net refuses a larger group
+	 * anyway. Defaults to that limit.
 	 */
 	maxGroupBytes?: number;
 }
@@ -126,7 +128,7 @@ export class Encoder<T> {
 
 	constructor(config: Config<T> = {}) {
 		this.#config = config;
-		this.#maxGroupBytes = config.maxGroupBytes ?? Group.MAX_GROUP_CACHE_BYTES;
+		this.#maxGroupBytes = Math.min(config.maxGroupBytes ?? Infinity, Group.MAX_GROUP_CACHE_BYTES);
 	}
 
 	/**
@@ -255,8 +257,11 @@ export class Encoder<T> {
 		return new TextEncoder().encode(JSON.stringify(result.patch));
 	}
 
-	// Encode a group's snapshot (frame 0), returning its payload.
+	// Encode a group's snapshot (frame 0), returning its payload. A snapshot past the budget is
+	// refused before any state changes, so the previous value stays readable rather than superseded
+	// by a frame moq-net would refuse on write.
 	#snapshot(frame: Uint8Array): Uint8Array {
+		if (frame.length > this.#maxGroupBytes) throw new NetError.FrameTooLarge();
 		this.#groupBytes = frame.length;
 		this.#groupFrames = 1;
 		this.#resync = false;

js/json/src/snapshot/snapshot.test.ts

@@ -2,6 +2,7 @@ import { expect, test } from "bun:test";
 import { Milli } from "@moq/clock";
 import { Group, Error as NetError, StreamCode, Time, Track } from "@moq/net";
 import { Consumer } from "./consumer.ts";
+import { Encoder } from "./encoder.ts";
 import { Producer } from "./producer.ts";
 
 type Value = Record<string, unknown>;
@@ -551,6 +552,14 @@ test("a delta that would overflow the snapshot rolls a new one instead", async (
 	expect(values[values.length - 1]).toEqual({ v: "y".repeat(size) });
 });
 
+test("a snapshot past the budget is refused", () => {
+	for (const deltas of [false, true]) {
+		const encoder = new Encoder<Value>({ deltas, maxGroupBytes: 100 });
+		expect(() => encoder.update({ v: "x".repeat(1000) })).toThrow(NetError.FrameTooLarge);
+		expect(encoder.value).toBeUndefined();
+	}
+});
+
 // A publisher gives up a group a reader is still on once a newer one supersedes it, which every
 // moq-lite version reports as OLD. The reader skips to the newer snapshot instead of failing.
 test("snapshot consumer skips a group the publisher dropped mid-read", async () => {

js/net/package.json

@@ -38,6 +38,7 @@
 		"@types/bun": "^1.4.3",
 		"@types/node": "^26.6.4",
 		"@typescript/lib-dom": "npm:@types/web@^0.0.358",
+		"fast-check": "^4.10.2",
 		"rimraf": "^6.1.3",
 		"typescript": "7.0.2"
 	}

js/net/src/group.ts

@@ -666,6 +666,8 @@ export class Consumer {
 	readonly sequence: number;
 
 	#state: GroupState;
+	// The copy it was mirrored from, which that copy's cache evicts.
+	readonly #mirrored: GroupState;
 	#expiry?: Expiry;
 	// Carries the group across its logical track's route changes; see `resume.ts`.
 	#recover?: GroupRecover;
@@ -685,6 +687,7 @@ export class Consumer {
 
 	private constructor(state: GroupState) {
 		this.#state = state;
+		this.#mirrored = state;
 		this.sequence = state.sequence;
 	}
 
@@ -746,6 +749,7 @@ export class Consumer {
 	static {
 		makeConsumer = (state) => new Consumer(state);
 		hooks.releaseGroupReader = (group, abort) => group.#release(abort);
+		hooks.readsMirrored = (group) => group.#state === group.#mirrored;
 		hooks.groupTimestamp = (group) =>
 			group.#state.unverified || group.#state.offset !== 0 ? undefined : group.#state.timestamp;
 		hooks.groupLatest = (group) => (group.#state.unverified ? undefined : group.#state.latest);
@@ -777,9 +781,7 @@ export class Consumer {
 				group.#closed = undefined;
 		};
 		hooks.readGroupFrame = (group, compressed) => group.#readFrame(0, compressed);
-		hooks.evictGroup = (group) => {
-			group.#evict();
-		};
+		hooks.evictGroup = (group) => group.#evict();
 	}
 
 	// Evaluate the drift budget for a read that has nothing buffered and is about to
@@ -801,7 +803,9 @@ export class Consumer {
 		return true;
 	}
 
-	#evict() {
+	#evict(): boolean {
+		// A group read on from another copy is that copy's to reclaim.
+		if (this.#state !== this.#mirrored) return false;
 		const frames = this.#bufferedFrames();
 		if (frames.length > 0) {
 			// Unread content is going away either way. If the budget had already given up
@@ -819,6 +823,9 @@ export class Consumer {
 		this.#state.cacheBytes = 0;
 		this.#state.flate = undefined;
 		this.#state.frames.set([]);
+		// A failed copy holds nothing more to reclaim, and its recovery may still continue the group.
+		const closed = this.#state.closed.peek();
+		return !(closed instanceof Error && this.#recovers(closed));
 	}
 
 	// The next change a parked read waits for.

js/net/src/internal.ts

@@ -554,6 +554,8 @@ export const hooks: {
 	 * Assigned by `group.ts`.
 	 */
 	releaseGroupReader: (group: GroupConsumer, abort?: Error) => void;
+	/** Whether a group reader still reads the copy it was mirrored from, not one its recovery moved it to. */
+	readsMirrored: (group: GroupConsumer) => boolean;
 	groupTimestamp: (group: GroupConsumer) => Timestamp | undefined;
 	groupLatest: (group: GroupConsumer) => Timestamp | undefined;
 	/** Return when a group joined its track at this hop, on the installed clock. */
@@ -572,8 +574,11 @@ export const hooks: {
 	recoverGroup: (group: GroupConsumer, recover: GroupRecover) => void;
 	/** Read the next frame, as its stored DEFLATE slice when `compressed`, so a lite-07 publisher forwards it without inflating. */
 	readGroupFrame: (group: GroupConsumer, compressed: boolean) => Promise<ReadGroupFrame | undefined>;
-	/** Make an evicted mirror terminal while its track timeline still contains it. */
-	evictGroup: (group: GroupConsumer) => void;
+	/**
+	 * Make an evicted mirror terminal while its track timeline still contains it. Whether the
+	 * mirror is released too: not when its group reads on, or may yet, from another copy.
+	 */
+	evictGroup: (group: GroupConsumer) => boolean;
 	/** Attach the origin advertisement of a created broadcast. */
 	attachAnnouncer: (
 		producer: BroadcastProducer,
@@ -771,6 +776,9 @@ export const hooks: {
 	releaseGroupReader: () => {
 		throw new Error("group.ts not loaded");
 	},
+	readsMirrored: () => {
+		throw new Error("group.ts not loaded");
+	},
 	groupTimestamp: () => {
 		throw new Error("group.ts not loaded");
 	},