fix(json): roll a snapshot before a delta evicts its own base

`delta_allowed` gated only on the delta ratio, so with a nonzero ratio a large
patch could be admitted into a group that no longer has room for it. The group
cache evicts from the front, and frame 0 is the snapshot every delta applies to,
so a ~4 MiB snapshot followed by a ~29 MiB patch dropped the base and left a late
subscriber with `Lagged` instead of the current value.

The budget is now part of the decision: a patch that would push the group past
the cache rolls a fresh snapshot instead. Checked before the window advances, so
falling back is clean, and measured on the plaintext, which bounds the
compressed frame.

Also preserves the fields belonging to a mode or compression this build does not
recognize. The draft and docs claim such an entry round-trips verbatim, but only
the discriminant survived: `{"mode":"windowed","window":10}` kept `windowed` and
dropped `window`, so a relay that reparsed and republished emitted an entry
nothing can act on, which is worse than dropping it outright. Both configs now
carry a flattened `extra`, and the browser schemas are loose, matching what
`Container` already does for an unknown container.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

Author: kixelated

js/hang/src/catalog/binary.ts

@@ -10,8 +10,11 @@ import { TimelineSchema } from "./timeline";
  * The entry says how to read the track, so a consumer needs the track name and nothing else about
  * the application. Prefer a {@link JsonConfig} track when the payloads are JSON, so a generic
  * consumer can read them.
+ *
+ * A *loose* object: fields this build doesn't recognize pass through untouched, so an entry using
+ * a future mode or compression round-trips rather than losing the fields that describe it.
  */
-export const BinaryConfigSchema = z.object({
+export const BinaryConfigSchema = z.looseObject({
 	// Optional reference to another broadcast that publishes this track, expressed
 	// relative to the broadcast that served this catalog (e.g. "./source").
 	// If unset, the track lives in the same broadcast as the catalog.

js/hang/src/catalog/data.test.ts

@@ -67,3 +67,16 @@ test("the known modes and compressions are recognized", () => {
 	expect(modeSupported("stream")).toBe(true);
 	expect(compressionSupported("deflate")).toBe(true);
 });
+
+test("fields belonging to an unrecognized mode survive a round trip", () => {
+	// Preserving the mode string alone is not enough: a future mode comes with fields describing it,
+	// and a relay that reparsed and republished would otherwise strip them, leaving an entry nothing
+	// can act on. Same guarantee the container schema gives an unknown container.
+	const parsed = RootSchema.parse({
+		json: { tracks: { future: { mode: "windowed", window: 10 } } },
+		binary: { tracks: { future: { mode: "windowed", chunk: "4k" } } },
+	});
+
+	expect(parsed.json?.tracks.future).toMatchObject({ mode: "windowed", window: 10 });
+	expect(parsed.binary?.tracks.future).toMatchObject({ mode: "windowed", chunk: "4k" });
+});

js/hang/src/catalog/json.ts

@@ -9,8 +9,11 @@ import { TimelineSchema } from "./timeline";
  *
  * The entry says how to read the track, so a consumer needs the track name and nothing else about
  * the application.
+ *
+ * A *loose* object: fields this build doesn't recognize pass through untouched, so an entry using
+ * a future mode or compression round-trips rather than losing the fields that describe it.
  */
-export const JsonConfigSchema = z.object({
+export const JsonConfigSchema = z.looseObject({
 	// Optional reference to another broadcast that publishes this track, expressed
 	// relative to the broadcast that served this catalog (e.g. "./source").
 	// If unset, the track lives in the same broadcast as the catalog.

rs/hang/src/catalog/binary.rs

@@ -82,6 +82,15 @@ pub struct BinaryConfig {
 	/// The companion timeline track indexing this track's groups, if the publisher offers one.
 	#[serde(default)]
 	pub timeline: Option<Timeline>,
+
+	/// Fields this build doesn't recognize, kept so the entry round-trips.
+	///
+	/// A future [`Mode`] or [`Compression`] almost certainly comes with fields describing it, and
+	/// preserving the discriminant without them would leave a relay republishing an entry nothing
+	/// can act on. Same reason [`Container`](crate::catalog::Container) keeps an unknown container
+	/// verbatim.
+	#[serde(flatten)]
+	pub extra: serde_json::Map<String, serde_json::Value>,
 }
 
 impl BinaryConfig {
@@ -93,6 +102,7 @@ impl BinaryConfig {
 			compression: None,
 			mime: None,
 			timeline: None,
+			extra: Default::default(),
 		}
 	}
 }

rs/hang/src/catalog/json.rs

@@ -80,6 +80,15 @@ pub struct JsonConfig {
 	/// The companion timeline track indexing this track's groups, if the publisher offers one.
 	#[serde(default)]
 	pub timeline: Option<Timeline>,
+
+	/// Fields this build doesn't recognize, kept so the entry round-trips.
+	///
+	/// A future [`Mode`] or [`Compression`] almost certainly comes with fields describing it, and
+	/// preserving the discriminant without them would leave a relay republishing an entry nothing
+	/// can act on. Same reason [`Container`](crate::catalog::Container) keeps an unknown container
+	/// verbatim.
+	#[serde(flatten)]
+	pub extra: serde_json::Map<String, serde_json::Value>,
 }
 
 impl JsonConfig {
@@ -91,6 +100,7 @@ impl JsonConfig {
 			compression: None,
 			schema: None,
 			timeline: None,
+			extra: Default::default(),
 		}
 	}
 }

rs/hang/src/catalog/root.rs

@@ -507,6 +507,23 @@ mod test {
 		assert_eq!(parsed, reparsed, "re-encoded catalog did not round-trip");
 	}
 
+	/// Preserving the mode string alone is not enough: a future mode comes with fields describing
+	/// it, and a relay that reparsed and republished would otherwise strip them, leaving an entry
+	/// nothing can act on.
+	#[test]
+	fn unknown_mode_fields_round_trip() {
+		let encoded = r#"{"json":{"tracks":{"future":{"mode":"windowed","window":10}}}}"#;
+
+		let parsed = Catalog::from_str(encoded).expect("failed to decode");
+		let future = parsed.json.tracks.get("future").expect("missing track");
+		assert_eq!(future.mode, Mode::Unknown("windowed".to_string()));
+		assert_eq!(future.extra.get("window"), Some(&serde_json::json!(10)));
+
+		let output = parsed.to_json().expect("failed to encode");
+		assert!(output.contains(r#""window":10"#), "unknown fields dropped: {output}");
+		assert_eq!(Catalog::from_str(&output).expect("re-decode"), parsed);
+	}
+
 	/// There is no safe default: reading a stream as a snapshot silently drops every record but the
 	/// last, so an entry without a mode is malformed rather than assumed.
 	#[test]

rs/moq-json/src/snapshot/encoder.rs

@@ -287,6 +287,17 @@ impl<T: Serialize> Encoder<T> {
 		if self.config.compression && bytes.len() as u64 > moq_flate::DEFAULT_MAX_FRAME_SIZE {
 			return Err(moq_flate::Error::TooLarge(moq_flate::DEFAULT_MAX_FRAME_SIZE).into());
 		}
+		// A delta is only readable while the group still holds the snapshot it applies to, and the
+		// group cache evicts from the front. Admitting a patch that pushes the group past that budget
+		// would drop frame 0, leaving a late subscriber with a base-less group (`Lagged`) instead of
+		// the current value. Roll a fresh snapshot instead, which is cheap next to losing the value.
+		//
+		// Measured on the plaintext, which bounds the compressed frame, and checked here because
+		// nothing has mutated yet: the window only advances on the line below.
+		if self.snapshot_len + self.delta_bytes + bytes.len() as u64 > moq_net::group::MAX_GROUP_CACHE {
+			return self.snapshot(value).map(Some);
+		}
+
 		let payload = match self.flate.as_mut() {
 			Some(flate) => flate.frame(&bytes),
 			None => Bytes::from(bytes),