test: prove unpublished dev API from a moq checkout

.github/workflows/smoke.yml

@@ -274,3 +274,33 @@ jobs:
         # failing the job; remove this once that protocol mismatch is fixed.
         continue-on-error: true
         run: ./moxygen.sh
+
+  from-dev:
+    name: Unpublished dev API
+    runs-on: ubuntu-latest
+    permissions:
+      contents: read
+    timeout-minutes: 45
+
+    steps:
+      - name: Checkout smoke
+        uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6.0.3
+        with:
+          persist-credentials: false
+
+      - name: Checkout moq (dev)
+        uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6.0.3
+        with:
+          repository: moq-dev/moq
+          ref: dev
+          path: moq
+          persist-credentials: false
+
+      - uses: oven-sh/setup-bun@0c5077e51419868618aeaa5fe8019c62421857d6 # v2.2.0
+      - uses: dtolnay/rust-toolchain@29eef336d9b2848a0b548edc03f92a220660cdb8 # stable
+
+      - name: System deps
+        run: sudo apt-get update && sudo apt-get install -y pkg-config cmake g++
+
+      - name: From-dev contract cases
+        run: ./dev.sh --src "$PWD/moq" --timeout 30

README.md

@@ -4,7 +4,7 @@ Cross-language interop smoke test for the **public** [Media over QUIC](https://g
 
 The [moq-dev/moq](https://github.com/moq-dev/moq) monorepo has its own in-tree smoke test, but it builds every client from workspace source. That proves the code in the tree works; it does **not** prove a real user can install the published artifacts and have them talk to each other. A missing wheel, a stale Homebrew formula, a broken `.deb`, an export that didn't survive packaging, a Go module missing its header. none of that shows up until someone installs from a registry.
 
-This repo installs each client straight from its public package registry, stands up a relay, and runs the interop matrix:
+This repo installs each client straight from its public package registry, stands up a relay, and runs the interop matrix. A second **from-dev** channel (`./dev.sh`) installs the unpublished `dev` surface from a moq checkout (path or git `dev`) and runs contract cases the published matrix cannot see yet: reconnecting `Connection` handles, announcements, credential refresh, publication replacement, catalog snapshots-then-deltas, and stats Snapshot versus Window. Embedded relay ownership stays in moq-relay; it is not a smoke client.
 
 - A relay (`moq-relay`) routes broadcasts.
 - For each publisher language, publish an H.264 broadcast.
@@ -84,14 +84,21 @@ RELAY_BIN=/path/to/moq-relay MOQ_BIN=/path/to/moq ./smoke.sh
 
 # prove the harness can fail: no publisher, every subscriber must time out.
 ./smoke.sh --negative --subscribers rust,python
+
+# unpublished dev API (JS packages + relay from a moq checkout, not npm/crates.io):
+just dev --src /path/to/moq
+just dev                 # clones github.com/moq-dev/moq (dev)
 ```
 
 `smoke.sh` installs the language clients (PyPI / Go proxy / npm) into a scratch dir on each run, so you always test the latest published versions. It does **not** install the Rust binaries; that is the channel under test.
 
+`dev.sh` is the other way around: it builds `moq-relay` from `MOQ_SRC` (or clones `dev`) and resolves `@moq/net`, `@moq/hang`, and `@moq/json` from that checkout's `js/` tree so the contract cases exercise the unpublished surface.
+
 ## Layout
 
 ```
-smoke.sh                 orchestrator: relay + media interop matrix
+smoke.sh                 orchestrator: relay + media interop matrix (published packages)
+dev.sh                   orchestrator: unpublished dev API contract cases (path/git `dev`)
 cloudflare.sh            orchestrator: Cloudflare client through both projects' relays
 moxygen.sh               orchestrator: moxygen protocol client through the moq-dev relay
 smoke.toml               relay config (anonymous, self-signed localhost)
@@ -105,6 +112,7 @@ clients/
   kotlin/                 subscribe via dev.moq:moq (Gradle/JVM)
   c/subscribe.c          subscribe via libmoq (prebuilt release)
   js-native/subscribe.ts subscribe via @moq/net + @moq/hang + WebTransport polyfill (node, bun)
+  dev/                   from-dev contract cases: Connection, catalog Snapshot, stats Snapshot vs Window
   (gst)                   subscribe via the moq-gst plugin (moqsrc); no client dir, driven by gst-launch
   docker/                 moq-relay + moq wrappers: docker run the moqdev/* images (the docker channel)
   token/js/              installs @moq/token (npm) for token.sh to drive under node + bun

clients/dev/catalog.ts

@@ -0,0 +1,105 @@
+// Catalog-only reading: Json.Snapshot.Consumer plus the hang catalog schema.
+// A full snapshot is followed by a merge-patch delta. Parsing every frame as a
+// catalog is the old consumer path and must fail on the delta frame.
+import * as Catalog from "@moq/hang/catalog";
+import * as Json from "@moq/json";
+import * as Moq from "@moq/net";
+import { connected, equal, handle, REPLAY_MS, waitActive, waitAnnounce } from "./lib.ts";
+
+const PATH = "dev.catalog";
+
+const SNAPSHOT: Catalog.Root = {
+	json: {
+		tracks: {
+			status: { mode: "snapshot" },
+		},
+	},
+};
+
+const UPDATED: Catalog.Root = {
+	json: {
+		tracks: {
+			extra: { mode: "snapshot" },
+		},
+	},
+};
+
+export async function catalog(url: URL): Promise<void> {
+	const pub = handle(url);
+	const sub = handle(url);
+	try {
+		const pubOrigin = await connected(pub);
+		const subOrigin = await connected(sub);
+		const announced = sub.announced();
+
+		const broadcast = pubOrigin.createBroadcast(Moq.Path.from(PATH));
+		const track = broadcast.createTrack(Catalog.TRACK);
+		const producer = new Json.Snapshot.Producer<Catalog.Root>({
+			track,
+			schema: Catalog.RootSchema,
+			deltaRatio: 100,
+		});
+		broadcast.announce();
+
+		await waitAnnounce(announced, PATH, true);
+		const request = subOrigin.request(Moq.Path.from(PATH));
+		const consumer = await waitActive(request, "catalog broadcast");
+
+		const snapshot = new Json.Snapshot.Consumer<Catalog.Root>({
+			track: consumer.track(Catalog.TRACK).subscribe({
+				priority: Catalog.PRIORITY.catalog,
+				maxAge: REPLAY_MS,
+			}),
+			schema: Catalog.RootSchema,
+		});
+
+		producer.update(SNAPSHOT);
+		const first = await snapshot.next();
+		if (!first || !equal(first, SNAPSHOT)) {
+			throw new Error(`first catalog was ${JSON.stringify(first)}`);
+		}
+
+		producer.update(UPDATED);
+		const second = await snapshot.next();
+		if (!second || !equal(second, UPDATED)) {
+			throw new Error(`delta did not reconstruct ${JSON.stringify(second)}`);
+		}
+
+		producer.finish();
+
+		const raw = consumer
+			.track(Catalog.TRACK)
+			.subscribe({ priority: Catalog.PRIORITY.catalog, maxAge: REPLAY_MS })
+			.ordered();
+		const group = await raw.nextGroup();
+		if (!group) throw new Error("catalog group missing");
+		const frames: Uint8Array[] = [];
+		for (;;) {
+			const frame = await group.readFrame();
+			if (!frame) break;
+			frames.push(frame.payload);
+		}
+		if (frames.length < 2) {
+			throw new Error(`expected a snapshot frame then a delta, got ${frames.length} frame(s)`);
+		}
+
+		const decoder = new TextDecoder();
+		const root = Catalog.RootSchema.parse(JSON.parse(decoder.decode(frames[0])));
+		if (!equal(root, SNAPSHOT)) throw new Error("frame 0 was not the full catalog snapshot");
+
+		let deltaParsedAsCatalog = false;
+		try {
+			Catalog.RootSchema.parse(JSON.parse(decoder.decode(frames[1])));
+			deltaParsedAsCatalog = true;
+		} catch {
+			// The delta is an RFC 7396 merge patch, not a catalog root.
+		}
+		if (deltaParsedAsCatalog) {
+			throw new Error("frame 1 parsed as a full catalog; the consumer is not reconstructing deltas");
+		}
+	} finally {
+		pub.close();
+		sub.close();
+	}
+	console.log("  catalog: ok");
+}

clients/dev/lib.ts

@@ -0,0 +1,75 @@
+// Shared helpers for the from-dev API contract cases.
+import * as Moq from "@moq/net";
+
+export const REPLAY_MS = 30_000;
+
+export function parseUrl(): { url: URL; timeoutMs: number } {
+	const args = process.argv.slice(2);
+	let url: string | undefined;
+	let timeout = "30";
+	for (let i = 0; i < args.length; i++) {
+		if (args[i] === "--url") url = args[++i];
+		else if (args[i] === "--timeout") timeout = args[++i] ?? timeout;
+	}
+	if (!url) {
+		console.error("usage: run.ts --url URL [--timeout S]");
+		process.exit(2);
+	}
+	return { url: new URL(url), timeoutMs: Number.parseFloat(timeout) * 1000 };
+}
+
+export async function waitUntil(pred: () => boolean, label: string, ms = 10_000): Promise<void> {
+	const deadline = Date.now() + ms;
+	for (;;) {
+		if (pred()) return;
+		if (Date.now() > deadline) throw new Error(`timeout waiting for ${label}`);
+		await new Promise((resolve) => setTimeout(resolve, 20));
+	}
+}
+
+export async function connected(conn: Moq.Connection, ms = 10_000): Promise<Moq.Origin.Producer> {
+	await waitUntil(
+		() => conn.status.peek() === "connected" && conn.origin.peek() !== undefined,
+		"connection established",
+		ms,
+	);
+	const origin = conn.origin.peek();
+	if (!origin) throw new Error("connected without an origin");
+	return origin;
+}
+
+export function announcedPath(entry: Moq.Announce.Event): string | undefined {
+	return entry.pattern.isLiteral ? entry.pattern.text : entry.pattern.asPrefix();
+}
+
+export async function waitAnnounce(
+	announced: Moq.Announce.Consumer,
+	path: string,
+	active: boolean,
+): Promise<void> {
+	for (;;) {
+		const entry = await announced.next();
+		if (!entry) throw new Error(`announce stream ended before ${path} ${active ? "appeared" : "retracted"}`);
+		if (announcedPath(entry) === path && entry.active === active) return;
+	}
+}
+
+export async function waitActive(
+	request: Moq.Origin.Request,
+	label: string,
+	ms = 10_000,
+): Promise<Moq.Broadcast.Consumer> {
+	await waitUntil(() => request.active.peek() !== undefined, label, ms);
+	const broadcast = request.active.peek();
+	if (!broadcast) throw new Error(`${label}: request resolved then vanished`);
+	return broadcast;
+}
+
+export function equal(a: unknown, b: unknown): boolean {
+	return JSON.stringify(a) === JSON.stringify(b);
+}
+
+export function handle(url: URL): Moq.Connection {
+	// Private loops so a publisher and a subscriber do not share an origin and skip the relay.
+	return new Moq.Connection({ url, share: false, linger: 0 });
+}

clients/dev/live.ts

@@ -0,0 +1,97 @@
+// Connection, announcements, credential refresh, and publication replacement.
+// Models the moq.pro live session (reconnecting handle, URL swap, announce cursor)
+// without copying that app.
+import * as Moq from "@moq/net";
+import { connected, handle, waitActive, waitAnnounce, waitUntil } from "./lib.ts";
+
+const PATH = "dev.live";
+
+export async function live(url: URL): Promise<void> {
+	await refresh(url);
+	await announceAndReplace(url);
+	console.log("  live: ok");
+}
+
+async function refresh(url: URL): Promise<void> {
+	const first = new URL(url.href);
+	first.searchParams.set("jwt", "first");
+	const second = new URL(url.href);
+	second.searchParams.set("jwt", "second");
+
+	const conn = handle(first);
+	try {
+		await connected(conn);
+		if (conn.closed.peek() !== undefined) {
+			throw new Error("closed settled before the handle was released");
+		}
+
+		conn.url.set(second);
+		await waitUntil(
+			() => conn.url.peek()?.href === second.href && conn.status.peek() === "connected",
+			"connected at the refreshed URL",
+		);
+		if (conn.closed.peek() !== undefined) {
+			throw new Error("closed settled on a URL swap; it is handle disposal, not session end");
+		}
+		if (conn.error.peek() !== undefined) {
+			throw new Error(`error after a recoverable URL swap: ${conn.error.peek()}`);
+		}
+	} finally {
+		conn.close();
+	}
+
+	const closed = await conn.closed;
+	if (closed !== null) throw new Error(`close() settled with ${closed}`);
+}
+
+async function announceAndReplace(url: URL): Promise<void> {
+	const pub = handle(url);
+	const sub = handle(url);
+	try {
+		const pubOrigin = await connected(pub);
+		const subOrigin = await connected(sub);
+		const announced = sub.announced();
+
+		const first = publish(pubOrigin, PATH, "v1");
+		await waitAnnounce(announced, PATH, true);
+
+		const request = subOrigin.request(Moq.Path.from(PATH));
+		const consumer = await waitActive(request, "first publication");
+		await expectFrame(consumer, "v1");
+
+		first.broadcast.close();
+		await waitAnnounce(announced, PATH, false);
+
+		const second = publish(pubOrigin, PATH, "v2");
+		await waitAnnounce(announced, PATH, true);
+		const replaced = await waitActive(request, "replaced publication");
+		await expectFrame(replaced, "v2");
+
+		second.broadcast.close();
+	} finally {
+		pub.close();
+		sub.close();
+	}
+}
+
+function publish(origin: Moq.Origin.Producer, path: string, payload: string) {
+	const broadcast = origin.createBroadcast(Moq.Path.from(path));
+	const track = broadcast.createTrack("messages");
+	const group = track.appendGroup();
+	group.writeString(payload);
+	group.close();
+	broadcast.announce();
+	return { broadcast, track };
+}
+
+async function expectFrame(broadcast: Moq.Broadcast.Consumer, payload: string): Promise<void> {
+	const track = broadcast.track("messages").subscribe({ priority: 0, maxAge: 30_000 });
+	try {
+		const group = await track.recvGroup();
+		if (!group) throw new Error("track ended before a group arrived");
+		const frame = await group.readString();
+		if (frame !== payload) throw new Error(`expected ${payload}, got ${frame}`);
+	} finally {
+		track.close();
+	}
+}

clients/dev/run.ts

@@ -0,0 +1,34 @@
+// From-dev API contract cases. Run against a local relay with JS packages
+// resolved from a moq checkout, not npm latest.
+import { install } from "@moq/web-transport";
+import { catalog } from "./catalog.ts";
+import { parseUrl } from "./lib.ts";
+import { live } from "./live.ts";
+import { stats } from "./stats.ts";
+
+install();
+
+const { url, timeoutMs } = parseUrl();
+
+let timeoutId: ReturnType<typeof setTimeout> | undefined;
+const timeout = new Promise<never>((_, reject) => {
+	timeoutId = setTimeout(() => reject(new Error("timed out waiting for the from-dev cases")), timeoutMs);
+});
+
+try {
+	await Promise.race([
+		(async () => {
+			console.log("from-dev cases against", url.href);
+			await live(url);
+			await catalog(url);
+			await stats(url);
+			console.log("from-dev: ok");
+		})(),
+		timeout,
+	]);
+} catch (err) {
+	console.error(`error: ${err instanceof Error ? err.message : String(err)}`);
+	process.exitCode = 1;
+} finally {
+	if (timeoutId !== undefined) clearTimeout(timeoutId);
+}