feat(moq-data): add metadata-over-MoQ helpers (set + json)

[kixelated]

Summary

Adds a new moq-data crate and @moq/data package: helpers for sending metadata over MoQ tracks, built on the same snapshot/delta machinery as moq-json. Two modules, each behind a feature (Rust) / subpath export (JS):

• set — a HashSet-like collection synced over a track with +/- delta encoding.
• json — re-exports moq-json for now (moq_data::json / @moq/data/json); JSON will migrate here over time.

The motivating use case is a tracks.set track listing all available track names in a broadcast, but the crate is kept generic enough to sync any binary data.

The set module

// Rust
let mut tracks = set::Producer::<String>::new(track, set::Config::default());
tracks.insert("video".to_string())?;   // '+' delta
tracks.remove("audio")?;               // '-' delta

let mut consumer = set::Consumer::<String>::new(track.consume());
while let Some(names) = consumer.next().await? { /* HashSet<String> */ }

// JS — mirror API
import { Producer, Consumer, stringCodec } from "@moq/data/set";
const producer = new Producer(track, { codec: stringCodec });
producer.insert("video");
for await (const names of new Consumer(track, { codec: stringCodec })) { /* Set<string> */ }

Items are arbitrary binary data, not serde-bound. A type plugs in via a tiny Item trait (Rust: encode -> Bytes / decode(Bytes) -> Self) or Codec<T> (JS: encode/decode to Uint8Array). String, Vec<u8>, Bytes and stringCodec/bytesCodec are provided.

Wire format (cross-language, byte-compatible)

Each group is self-contained, mirroring moq-json:

• Frame 0 = atomic snapshot: u32(count) then, per item, u32(len) + item bytes.
• Frames 1+ = a single op: one byte (+ = 0x2B insert, - = 0x2D remove) + item bytes to end of frame.

A consumer jumps to the newest group, decodes the snapshot, replays the deltas, and yields the full set after each frame. Deltas are on by default (delta_ratio / deltaRatio = 2); a new snapshot group rolls once the deltas outgrow the ratio budget or hit MAX_DELTA_FRAMES (256). Lengths are big-endian u32 (not QUIC varints) so the Rust and JS encoders stay trivially byte-compatible without depending on moq-net internals.

Scope / branch targeting

This is an additive new crate/package, so it targets main. The actual tracks.set wiring into the hang catalog is a follow-up — that's a catalog/wire change and belongs on dev. This PR ships only the generic building block, with a string-set test demonstrating the scenario.

Public API surface (all new, nothing breaking)

• rs/moq-data: set::{Producer, Consumer, Config, Item, Error, Result} + Item impls for String/Vec<u8>/Bytes; json re-export. Config is #[non_exhaustive] with a Default; Error is #[non_exhaustive].
• @moq/data: set::{Producer, Consumer, Config, Codec, stringCodec, bytesCodec}; @moq/data/json re-export.

Test plan

• cargo test -p moq-data --all-features — 8 tests pass
• cargo fmt / cargo clippy -p moq-data --all-features clean (via nix)
• Feature combos compile: json-only, set-only, no-default-features
• bun test in js/data — 6 tests pass
• tsc --noEmit and biome check js/data clean

Cross-package sync

Added a row to the CLAUDE.md sync table: rs/moq-data set wire/API ↔ js/data (shared wire format, must stay byte-compatible).

🤖 Generated with Claude Code

(Written by Claude)

[coderabbitai]

Note

Reviews paused

It looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the reviews.auto_review.auto_pause_after_reviewed_commits setting.

Use the following commands to manage reviews:

• @coderabbitai resume to resume automatic reviews.
• @coderabbitai review to trigger a single review.

Use the checkboxes below for quick actions:

• ▶️ Resume reviews
• 🔍 Trigger review

Walkthrough

Two new parallel packages are introduced: rs/moq-data (Rust) and js/data (TypeScript). Both implement a replicated set collection over MoQ tracks using an identical snapshot/delta wire protocol: each track group begins with a full snapshot frame (big-endian u32 item count + u32-length-prefixed items) followed by optional delta frames (1-byte opcode + item bytes). A configurable delta_ratio and MAX_DELTA_FRAMES cap determine when a new snapshot group is forced. Both packages expose Producer/Consumer APIs with deduplication keyed on encoded item bytes. The Rust crate also adds a Sizer utility (BufMut byte counter) and re-exports moq-json as the json feature. Both packages are registered in their respective workspace manifests, and a cross-package sync note is added to CLAUDE.md.

🚥 Pre-merge checks | ✅ 4 | ❌ 1

❌ Failed checks (1 warning)

Check name	Status	Explanation	Resolution
Docstring Coverage	⚠️ Warning	Docstring coverage is 48.00% which is insufficient. The required threshold is 80.00%.	Write docstrings for the functions missing them to satisfy the coverage threshold.

✅ Passed checks (4 passed)

Check name	Status	Explanation
Title check	✅ Passed	The title 'feat(moq-data): add metadata-over-MoQ helpers (set + json)' directly and clearly summarizes the main change: adding new metadata-over-MoQ helpers with set and json modules.
Description check	✅ Passed	The description comprehensively covers the changeset, explaining the new moq-data crate/package, the set and json modules, wire format design, test coverage, and public API surface.
Linked Issues check	✅ Passed	Check skipped because no linked issues were found for this pull request.
Out of Scope Changes check	✅ Passed	Check skipped because no linked issues were found for this pull request.

_{✏️ Tip: You can configure your own custom pre-merge checks in the settings.}

✨ Finishing Touches

✨ Simplify code

• Create PR with simplified code
• Commit simplified code in branch claude/unruffled-wing-ad0814

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

[coderabbitai]

Actionable comments posted: 4

🧹 Nitpick comments (5)

js/data/src/set/set.test.ts (2)

69-82: ⚡ Quick win

Consider calling finish() for resource cleanup.

While the test correctly verifies live consumption, it doesn't close the track or producer. For completeness and to avoid potential resource leaks, consider adding:

 	producer.remove("video");
 	expect(await consumer.next()).toEqual(set("audio"));
+
+	producer.finish();
 });

This makes the cleanup explicit and matches the pattern in other tests.

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@js/data/src/set/set.test.ts` around lines 69 - 82, The test "live consumer
sees each change" creates Track and Producer objects but does not clean up these
resources after the test completes, which can lead to resource leaks. Add a call
to finish() at the end of the test after the last assertion to explicitly close
and clean up the track and producer resources. Follow the same cleanup pattern
used in other tests in this file to ensure consistency.

---

35-44: ⚡ Quick win

Test name implies structure verification but only checks final state.

The test name "deltas off: a snapshot group per change" suggests verifying that each change creates a separate snapshot group, but the assertion only checks the final reconstructed set. Consider adding a structure assertion to match the test name:

 	producer.insert("audio");
 	producer.finish();
 
+	expect(await structure(track)).toEqual([1, 1]); // two groups, one frame each
-	expect((await drain(track)).at(-1)).toEqual(set("video", "audio"));

Alternatively, rename the test to focus on the end state rather than the internal structure.

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@js/data/src/set/set.test.ts` around lines 35 - 44, The test "deltas off: a
snapshot group per change" has a name that implies verification of internal
snapshot group structure, but the assertion only checks the final reconstructed
set. Either add an assertion to verify that drain(track) contains multiple
snapshot groups (one per insert operation) to match the test name, or rename the
test to accurately reflect that it only verifies the final state after inserts
(for example, "deltas off: inserts produce correct final set").

js/data/src/json.ts (1)

1-2: 💤 Low value

Rephrase comment to focus on current state, not future plans.

The comment "will migrate here over time" describes future intent rather than current state. Based on learnings, comments should reflect the current state of the code, not its history or future. Consider rephrasing to explain why the re-export exists today without forecasting future changes.

📝 Suggested rephrase

-// Snapshot/delta JSON publishing, re-exported from `@moq/json`. JSON lives in its own package today
-// and will migrate here over time.
+// Snapshot/delta JSON publishing, re-exported from `@moq/json`.
 export * from "`@moq/json`";

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@js/data/src/json.ts` around lines 1 - 2, The comment at the beginning of the
json.ts file contains forward-looking language about future migration ("will
migrate here over time") instead of explaining the current state. Rephrase the
comment to describe why the re-export from `@moq/json` exists in its current state
today, removing the future-oriented language. Focus on the present reason for
the re-export structure rather than forecasting planned changes.

Source: Learnings

js/data/src/set/consumer.ts (1)

14-14: ⚡ Quick win

Consider importing Codec directly for clarity.

Using Config<T>["codec"] creates an indirect dependency on the shape of Config. If Config changes or is renamed, this type reference breaks. Importing Codec<T> directly from ./codec.ts makes the dependency explicit and the code more resilient to refactoring.

♻️ Suggested refactor

-import type { Config } from "./producer.ts";
+import type { Codec } from "./codec.ts";
 import { decodeDelta, decodeSnapshot, INSERT, keyOf, REMOVE } from "./wire.ts";

 export class Consumer<T> {
 	`#track`: Moq.Track;
-	`#codec`: Config<T>["codec"];
+	`#codec`: Codec<T>;
 
 	`#group`?: Moq.Group;
 	// Keyed by encoded bytes so items dedupe by value, not reference.
 	`#current` = new Map<string, T>();
 	`#framesRead` = 0;
 
-	constructor(track: Moq.Track, config: Config<T>) {
+	constructor(track: Moq.Track, config: { codec: Codec<T> }) {
 		this.#track = track;
 		this.#codec = config.codec;
 	}

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@js/data/src/set/consumer.ts` at line 14, The `#codec` field currently uses
Config<T>["codec"] as its type, creating an indirect dependency on the Config
type that makes the code fragile to refactoring. Replace this indexed type
reference by directly importing Codec<T> from ./codec.ts and using it as the
type annotation for the `#codec` field instead of Config<T>["codec"]. This makes
the dependency explicit and resilient to Config changes or renames.

js/data/src/set/producer.ts (1)

99-103: ⚡ Quick win

Consider caching encoded bytes to avoid re-encoding in snapshots.

#snapshot() re-encodes every value on each call, even though the map is already keyed by the encoded bytes (via keyOf(bytes)). Storing the Uint8Array alongside each T would eliminate redundant encoding when rolling snapshots, especially for large sets or expensive codecs.

⚡ Suggested optimization

Change #items to store { value: T, bytes: Uint8Array }:

-	`#items` = new Map<string, T>();
+	`#items` = new Map<string, { value: T; bytes: Uint8Array }>();

Then update insert, remove, has, values, and #snapshot:

 	insert(value: T): boolean {
 		const bytes = this.#codec.encode(value);
 		const key = keyOf(bytes);
 		if (this.#items.has(key)) return false;

-		this.#items.set(key, value);
+		this.#items.set(key, { value, bytes });
 		this.#publish(INSERT, bytes);
 		return true;
 	}

 	remove(value: T): boolean {
 		const bytes = this.#codec.encode(value);
 		const key = keyOf(bytes);
 		if (!this.#items.has(key)) return false;

 		this.#items.delete(key);
 		this.#publish(REMOVE, bytes);
 		return true;
 	}

 	values(): IterableIterator<T> {
-		return this.#items.values();
+		return (function* (items) {
+			for (const { value } of items.values()) yield value;
+		})(this.#items);
 	}

 	`#snapshot`(): Uint8Array {
-		const items: Uint8Array[] = [];
-		for (const value of this.#items.values()) items.push(this.#codec.encode(value));
-		return encodeSnapshot(items);
+		const items = Array.from(this.#items.values(), ({ bytes }) => bytes);
+		return encodeSnapshot(items);
 	}

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@js/data/src/set/producer.ts` around lines 99 - 103, The `#snapshot()` method
re-encodes every value each time it is called, which is inefficient since
encoding has already been performed. Refactor the `#items` data structure to
store both the original value and its pre-encoded bytes as an object `{ value:
T, bytes: Uint8Array }` instead of just the value `T`. Then update the
`#snapshot()` method to use the cached `bytes` field directly instead of calling
`this.#codec.encode(value)` in the loop. Additionally, update the `insert`,
`remove`, `has`, and `values` methods to work with this new structure by
accessing the value property where needed and storing the encoded bytes when
inserting new items.

🤖 Prompt for all review comments with AI agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Inline comments:
In `@js/data/src/set/producer.ts`:
- Around line 6-7: Replace the em dash ("—") used in the comment block with
appropriate punctuation following style guidelines. Either use a period to
separate the two related thoughts into distinct sentences, or use a
comma/parenthesis to connect them as a single continuous thought. The comment
describes Maximum frames behavior and should be reworded to remove the em dash
while maintaining clarity.

In `@js/data/src/set/wire.ts`:
- Around line 45-55: The snapshot decoders currently accept trailing bytes after
processing all declared items, which can allow malformed snapshots to parse
successfully and silently drop intended state. In js/data/src/set/wire.ts (lines
45-55), after the decode loop completes, add a check that requires offset to
equal frame.length, throwing an error if there are extra bytes remaining. In
rs/moq-data/src/set.rs (lines 379-390), after the decode loop completes, add a
check that requires the frame has no remaining data using frame.has_remaining(),
returning Error::Malformed if there are trailing bytes. Both files must validate
that all bytes in the snapshot have been consumed by the declared items.

In `@rs/moq-data/src/set.rs`:
- Around line 16-19: The snapshot format documentation in the comment block
describes the encoding using varint notation (varint(count) and varint(len)),
but the actual implementation uses big-endian u32 encoding. Update lines 16-17
in the snapshot documentation to accurately reflect that the format uses
big-endian u32 for both the count and length fields instead of variadic integer
encoding, ensuring the docs match the actual wire contract implemented in the
module.
- Around line 376-379: The code preallocates a HashSet using capacity from
untrusted wire data (the count obtained from frame.get_u32()) without first
validating it, which can allow an attacker to cause memory exhaustion. Remove
the capacity preallocation by replacing HashSet::with_capacity(count as usize)
with HashSet::new(), and let the HashSet grow naturally as items are inserted.
This ensures structural validation of the frame happens before any resource
allocation based on untrusted data.

---

Nitpick comments:
In `@js/data/src/json.ts`:
- Around line 1-2: The comment at the beginning of the json.ts file contains
forward-looking language about future migration ("will migrate here over time")
instead of explaining the current state. Rephrase the comment to describe why
the re-export from `@moq/json` exists in its current state today, removing the
future-oriented language. Focus on the present reason for the re-export
structure rather than forecasting planned changes.

In `@js/data/src/set/consumer.ts`:
- Line 14: The `#codec` field currently uses Config<T>["codec"] as its type,
creating an indirect dependency on the Config type that makes the code fragile
to refactoring. Replace this indexed type reference by directly importing
Codec<T> from ./codec.ts and using it as the type annotation for the `#codec`
field instead of Config<T>["codec"]. This makes the dependency explicit and
resilient to Config changes or renames.

In `@js/data/src/set/producer.ts`:
- Around line 99-103: The `#snapshot()` method re-encodes every value each time
it is called, which is inefficient since encoding has already been performed.
Refactor the `#items` data structure to store both the original value and its
pre-encoded bytes as an object `{ value: T, bytes: Uint8Array }` instead of just
the value `T`. Then update the `#snapshot()` method to use the cached `bytes`
field directly instead of calling `this.#codec.encode(value)` in the loop.
Additionally, update the `insert`, `remove`, `has`, and `values` methods to work
with this new structure by accessing the value property where needed and storing
the encoded bytes when inserting new items.

In `@js/data/src/set/set.test.ts`:
- Around line 69-82: The test "live consumer sees each change" creates Track and
Producer objects but does not clean up these resources after the test completes,
which can lead to resource leaks. Add a call to finish() at the end of the test
after the last assertion to explicitly close and clean up the track and producer
resources. Follow the same cleanup pattern used in other tests in this file to
ensure consistency.
- Around line 35-44: The test "deltas off: a snapshot group per change" has a
name that implies verification of internal snapshot group structure, but the
assertion only checks the final reconstructed set. Either add an assertion to
verify that drain(track) contains multiple snapshot groups (one per insert
operation) to match the test name, or rename the test to accurately reflect that
it only verifies the final state after inserts (for example, "deltas off:
inserts produce correct final set").

🪄 Autofix (Beta)

Fix all unresolved CodeRabbit comments on this PR:

• Push a commit to this branch (recommended)
• Create a new PR with the fixes

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro

Run ID: bad80947-5197-4023-b7c0-82e37a99ed9e

📥 Commits

Reviewing files that changed from the base of the PR and between ae51aac and a1d8d7b.

⛔ Files ignored due to path filters (2)

• Cargo.lock is excluded by !**/*.lock
• bun.lock is excluded by !**/*.lock

📒 Files selected for processing (18)

• CLAUDE.md
• Cargo.toml
• js/data/README.md
• js/data/package.json
• js/data/src/index.ts
• js/data/src/json.ts
• js/data/src/set/codec.ts
• js/data/src/set/consumer.ts
• js/data/src/set/index.ts
• js/data/src/set/producer.ts
• js/data/src/set/set.test.ts
• js/data/src/set/wire.ts
• js/data/tsconfig.json
• package.json
• rs/moq-data/Cargo.toml
• rs/moq-data/README.md
• rs/moq-data/src/lib.rs
• rs/moq-data/src/set.rs

[coderabbitai]

Actionable comments posted: 1

Caution

Some comments are outside the diff and can’t be posted inline due to platform limitations.

⚠️ Outside diff range comments (1)

rs/moq-data/src/set.rs (1)

190-220: ⚠️ Potential issue | 🟠 Major | ⚡ Quick win

Preserve self.current on publish failures.

Line 191 and Line 208 commit local state before write_snapshot or write_delta succeeds. If publishing fails, retries become no-ops while the track may never have observed the change.

🐛 Proposed fix

@@
-		if self.should_snapshot(delta_size, snapshot_size) {
-			self.current.insert(item);
-			self.write_snapshot()?;
+		if self.should_snapshot(delta_size, snapshot_size) {
+			self.current.insert(item.clone());
+			if let Err(err) = self.write_snapshot() {
+				self.current.remove(&item);
+				return Err(err);
+			}
 		} else {
 			// Write the delta straight from the reference, then move the item into the set.
 			self.write_delta(INSERT, &item)?;
 			self.current.insert(item);
 		}
@@
-		if self.should_snapshot(delta_size, snapshot_size) {
-			self.write_snapshot()?;
-		} else {
-			self.write_delta(REMOVE, &removed)?;
-		}
+		let publish = if self.should_snapshot(delta_size, snapshot_size) {
+			self.write_snapshot()
+		} else {
+			self.write_delta(REMOVE, &removed)
+		};
+		if let Err(err) = publish {
+			self.current.insert(removed);
+			return Err(err);
+		}
 		Ok(true)
 	}

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@rs/moq-data/src/set.rs` around lines 190 - 220, The insert and remove methods
modify self.current (local state) before ensuring that the write operations
succeed. In the insert method, move the self.current.insert(item) call to occur
only after both the write_snapshot() and write_delta() calls have completed
successfully. In the remove method, since the item is taken from self.current
before the write attempt, restore the removed item back into self.current if
write_snapshot() or write_delta() fails, so that retries will attempt to publish
the change again rather than becoming no-ops.

🤖 Prompt for all review comments with AI agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Inline comments:
In `@rs/moq-data/src/set.rs`:
- Around line 262-263: Update the documentation comment to reference the correct
method name. The doc comment currently references a non-existent trait method
`Item::size`, which should be changed to `Item::encode_size`. Locate the doc
comment describing the byte size computation and replace the outdated method
reference to point to the correct method that is currently used for sizing
calculations.

---

Outside diff comments:
In `@rs/moq-data/src/set.rs`:
- Around line 190-220: The insert and remove methods modify self.current (local
state) before ensuring that the write operations succeed. In the insert method,
move the self.current.insert(item) call to occur only after both the
write_snapshot() and write_delta() calls have completed successfully. In the
remove method, since the item is taken from self.current before the write
attempt, restore the removed item back into self.current if write_snapshot() or
write_delta() fails, so that retries will attempt to publish the change again
rather than becoming no-ops.

🪄 Autofix (Beta)

Fix all unresolved CodeRabbit comments on this PR:

• Push a commit to this branch (recommended)
• Create a new PR with the fixes

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro

Run ID: ba4419ef-89a3-4a83-8ab4-624fc7f8b1cd

📥 Commits

Reviewing files that changed from the base of the PR and between ca5fc62 and 38c7140.

📒 Files selected for processing (3)

• rs/moq-data/README.md
• rs/moq-data/src/set.rs
• rs/moq-net/src/lib.rs

✅ Files skipped from review due to trivial changes (1)

• rs/moq-data/README.md