feat(js/net): carry datagrams over moq-transport as OBJECT_DATAGRAM

doc/concept/standard.md

@@ -31,7 +31,7 @@ subset drops. What a peer actually observes against this implementation:
 - **Strict SETUP and parameters.** Repeated unknown `SETUP` options, GREASE included, are accepted if well-formed; a repeated known option is rejected. On draft 17 and later, a `GROUP_ORDER` other than Ascending (1) or Descending (2) closes the session.
 - **One credential per session**, carried in `SETUP` and forwarded to the [auth server](/bin/relay/auth#the-contract) unverified. A token attached to an individual request is ignored. An alias reference, a second token, or a parameter the negotiated draft does not define closes the session.
 - **Refused, not fatal.** A legal request this stack does not serve is rejected on its own and the session stays up: `FORWARD=0`, range filters, `TRACK_STATUS`, `SUBSCRIBE_TRACKS`, and the fetch forms above. On draft 19 and later, and in Rust on drafts 14 through 16, a subscription update may change only priority; any other update ends that subscription.
-- **Datagrams** are a single normal object at object 0, forwarded without renumbering. Anything else is dropped. JavaScript does not carry datagrams on moq-transport.
+- **Datagrams** are a single normal object at object 0, forwarded without renumbering. Anything else is dropped, and a malformed one closes the session. Rust and JavaScript both carry them on every draft.
 - **Priority.** Higher is served first. The IETF default of 128 is this stack's 127, and a track that never sets one is 127.
 - **Size.** An object extension block larger than 64 KiB ends that subgroup stream. The session stays up. This cap is ours, not the draft's.
 

js/net/src/ietf/aliases.ts

@@ -118,6 +118,11 @@ export class TrackAliases<T> {
 		}
 	}
 
+	/** The alias's binding right now, without waiting for one. */
+	peek(alias: bigint): T | undefined {
+		return this.#active.get(alias)?.value;
+	}
+
 	/**
 	 * Establishes an alias and releases any data streams waiting for it.
 	 *

js/net/src/ietf/connection.ts

@@ -213,6 +213,7 @@ export class Connection implements Established {
 		if (this.#closed) return;
 
 		this.#closed = true;
+		this.#publisher.close();
 
 		// Before the session, whose own close would send a clean code first.
 		try {
@@ -232,11 +233,16 @@ export class Connection implements Established {
 
 	async #run(early: Reader[]): Promise<void> {
 		try {
-			// Both run together. runPublishNamespaces is a no-op when the peer asked to be
+			// All run together. runPublishNamespaces is a no-op when the peer asked to be
 			// solicited; otherwise it pushes PUBLISH_NAMESPACE. On draft-16 and later a
 			// SUBSCRIBE_NAMESPACE stream is filled either way. A NAMESPACE is discovery,
 			// not a second route.
-			await Promise.all([this.#runBidis(), this.#runUnis(early), this.#publisher.runPublishNamespaces()]);
+			await Promise.all([
+				this.#runBidis(),
+				this.#runUnis(early),
+				this.#runDatagrams(),
+				this.#publisher.runPublishNamespaces(),
+			]);
 		} catch (err) {
 			if (!this.#closed) {
 				console.error("fatal error running connection", err);
@@ -389,6 +395,17 @@ export class Connection implements Established {
 		}
 	}
 
+	// A malformed OBJECT_DATAGRAM is the peer breaking the protocol, like an invalid stream type.
+	async #runDatagrams() {
+		try {
+			await this.#subscriber.runDatagrams();
+		} catch (err: unknown) {
+			if (!(err instanceof ProtocolViolation)) throw err;
+			console.warn("malformed datagram", err);
+			this.#violated(err);
+		}
+	}
+
 	/**
 	 * Handles unidirectional streams for media delivery (groups).
 	 */

js/net/src/ietf/datagram.test.ts

@@ -0,0 +1,112 @@
+import { expect, test } from "bun:test";
+import { ProtocolViolation } from "../error.ts";
+import { encodeVarint } from "../stream.ts";
+import { ObjectDatagram } from "./datagram.ts";
+import { type IetfVersion, Version } from "./version.ts";
+
+const ALL: IetfVersion[] = [
+	Version.DRAFT_14,
+	Version.DRAFT_15,
+	Version.DRAFT_16,
+	Version.DRAFT_17,
+	Version.DRAFT_18,
+	Version.DRAFT_19,
+	Version.DRAFT_20,
+	Version.DRAFT_21,
+	Version.DRAFT_22,
+];
+
+const LEGACY: IetfVersion[] = [Version.DRAFT_14, Version.DRAFT_15, Version.DRAFT_16];
+
+function single(publisherPriority?: number): ObjectDatagram {
+	return new ObjectDatagram({
+		trackAlias: 3n,
+		groupId: 42,
+		publisherPriority,
+		endOfGroup: true,
+		properties: Uint8Array.of(0x10, 0x05),
+		body: { payload: new TextEncoder().encode("hello") },
+	});
+}
+
+async function rejects(bytes: Uint8Array, version: IetfVersion) {
+	expect(await ObjectDatagram.decode(bytes, version).catch((err: unknown) => err)).toBeInstanceOf(ProtocolViolation);
+}
+
+test("roundtrips on every draft", async () => {
+	for (const version of ALL) {
+		const datagram = single(7);
+		const bytes = datagram.encode(version);
+		expect(bytes[0]).toBe(0x07); // properties, end of group, object 0
+		expect(await ObjectDatagram.decode(bytes, version)).toEqual(datagram);
+	}
+});
+
+test("a default priority needs draft-15", async () => {
+	expect(() => single().encode(Version.DRAFT_14)).toThrow();
+	await rejects(Uint8Array.of(0x0c, 0x01, 0x02), Version.DRAFT_14);
+
+	const bytes = single().encode(Version.DRAFT_16);
+	expect(bytes[0]).toBe(0x0f);
+	expect(await ObjectDatagram.decode(bytes, Version.DRAFT_16)).toEqual(single());
+});
+
+test("an explicit object id and a status", async () => {
+	const datagram = new ObjectDatagram({
+		trackAlias: 1n,
+		groupId: 2,
+		objectId: 0,
+		publisherPriority: 128,
+		endOfGroup: false,
+		body: { status: 0 },
+	});
+	for (const version of ALL) {
+		const bytes = datagram.encode(version);
+		expect(bytes[0]).toBe(0x20);
+		expect(await ObjectDatagram.decode(bytes, version)).toEqual(datagram);
+	}
+});
+
+test("rejects invalid types", async () => {
+	for (const version of ALL) {
+		// A status that ends the group, the reserved bit, a bit past the defined ones, and a
+		// Type too wide for any of them.
+		for (const kind of [0x22, 0x10, 0x40, 2 ** 40]) {
+			const type = encodeVarint(kind, version);
+			const bytes = new Uint8Array(type.byteLength + 4);
+			bytes.set(type);
+			bytes.set([0x01, 0x02, 0x03, 0x04], type.byteLength);
+			await rejects(bytes, version);
+		}
+	}
+});
+
+test("a status with properties needs a Normal Object past draft-16", async () => {
+	const datagram = new ObjectDatagram({
+		trackAlias: 1n,
+		groupId: 2,
+		objectId: 0,
+		publisherPriority: 0,
+		endOfGroup: false,
+		properties: Uint8Array.of(0x10, 0x05),
+		body: { status: 3 },
+	});
+	for (const version of ALL) {
+		if (LEGACY.includes(version)) {
+			expect(await ObjectDatagram.decode(datagram.encode(version), version)).toEqual(datagram);
+		} else {
+			expect(() => datagram.encode(version)).toThrow();
+			// The same bytes a legacy draft writes, which these drafts refuse.
+			await rejects(datagram.encode(Version.DRAFT_16), version);
+		}
+	}
+});
+
+test("rejects an empty properties block, a truncated header, and bytes after a status", async () => {
+	// PROPERTIES and ZERO_OBJECT_ID, alias 1, group 2, priority 0, empty block.
+	await rejects(Uint8Array.of(0x05, 0x01, 0x02, 0x00, 0x00, 0x78), Version.DRAFT_16);
+	// The group ID is missing.
+	await rejects(Uint8Array.of(0x04, 0x01), Version.DRAFT_16);
+	// A Normal status, then a stray byte.
+	await rejects(Uint8Array.of(0x24, 0x01, 0x02, 0x00, 0x00, 0x00), Version.DRAFT_16);
+});

js/net/src/ietf/datagram.ts

@@ -0,0 +1,189 @@
+/**
+ * OBJECT_DATAGRAM: one Object carried in a QUIC datagram (draft-14 section 10.3.1 through
+ * draft-20 section 11.3.1).
+ *
+ * The model counterpart is {@link Datagram}, a single-frame group, so only an Object at ID 0
+ * maps onto it. The Type is a set of flags on every draft; draft-14 lacks the
+ * DEFAULT_PRIORITY bit and a status with an omitted Object ID. Mirrors the Rust
+ * `ietf::ObjectDatagram`.
+ *
+ * @module
+ */
+import type { Datagram } from "../datagram.ts";
+import { error, ProtocolViolation, reason } from "../error.ts";
+import { type Cursor, encodeVarint, Reader } from "../stream.ts";
+import type { Timescale } from "../time.ts";
+import { encodeObjectExtensions } from "./object.ts";
+import { type IetfVersion, Version } from "./version.ts";
+
+// The bits of an OBJECT_DATAGRAM Type.
+const PROPERTIES = 0x01;
+const END_OF_GROUP = 0x02;
+const ZERO_OBJECT_ID = 0x04;
+const DEFAULT_PRIORITY = 0x08;
+const STATUS = 0x20;
+// Every defined bit. Anything else, including the reserved 0x10, is invalid.
+const ALL = PROPERTIES | END_OF_GROUP | ZERO_OBJECT_ID | DEFAULT_PRIORITY | STATUS;
+
+/** Whether `kind` is a Type this draft defines. */
+function valid(kind: number, version: IetfVersion): boolean {
+	if (version === Version.DRAFT_14) return kind <= 0x07 || kind === 0x20 || kind === 0x21;
+	// A status cannot also end the group.
+	return (kind & ~ALL) === 0 && !((kind & STATUS) !== 0 && (kind & END_OF_GROUP) !== 0);
+}
+
+/** Drafts 14-16 let any status carry Properties; later ones only a Normal Object. */
+function legacy(version: IetfVersion): boolean {
+	return version === Version.DRAFT_14 || version === Version.DRAFT_15 || version === Version.DRAFT_16;
+}
+
+/** What follows an OBJECT_DATAGRAM's header: the payload, or the status of an Object without one. */
+export type DatagramBody = { payload: Uint8Array } | { status: number };
+
+/** A decoded OBJECT_DATAGRAM. */
+export class ObjectDatagram {
+	trackAlias: bigint;
+	groupId: number;
+	/** The Object ID, or `undefined` when the ZERO_OBJECT_ID bit omits it (Object 0). */
+	objectId?: number;
+	/** The Publisher Priority, or `undefined` to inherit the subscription's (draft-15+). */
+	publisherPriority?: number;
+	/** No Object past this one exists in the group. */
+	endOfGroup: boolean;
+	/** The Object Properties block without its length prefix, which carries the Timestamp. */
+	properties?: Uint8Array;
+	body: DatagramBody;
+
+	constructor(props: {
+		trackAlias: bigint;
+		groupId: number;
+		objectId?: number;
+		publisherPriority?: number;
+		endOfGroup: boolean;
+		properties?: Uint8Array;
+		body: DatagramBody;
+	}) {
+		this.trackAlias = props.trackAlias;
+		this.groupId = props.groupId;
+		this.objectId = props.objectId;
+		this.publisherPriority = props.publisherPriority;
+		this.endOfGroup = props.endOfGroup;
+		this.properties = props.properties;
+		this.body = props.body;
+	}
+
+	/** Encode one QUIC datagram's bytes. Throws on a combination this draft cannot carry. */
+	encode(version: IetfVersion): Uint8Array {
+		let kind = 0;
+		if (this.properties !== undefined) kind |= PROPERTIES;
+		if (this.endOfGroup) kind |= END_OF_GROUP;
+		if (this.objectId === undefined) kind |= ZERO_OBJECT_ID;
+		if (this.publisherPriority === undefined) kind |= DEFAULT_PRIORITY;
+		if ("status" in this.body) kind |= STATUS;
+		if (!valid(kind, version)) throw new Error(`OBJECT_DATAGRAM type 0x${kind.toString(16)} is invalid`);
+
+		const parts = [encodeVarint(kind, version), encodeVarint(this.trackAlias, version)];
+		parts.push(encodeVarint(this.groupId, version));
+		if (this.objectId !== undefined) parts.push(encodeVarint(this.objectId, version));
+		if (this.publisherPriority !== undefined) parts.push(Uint8Array.of(this.publisherPriority));
+		if (this.properties !== undefined) {
+			// A present but empty block is a protocol violation for the peer.
+			if (this.properties.byteLength === 0) throw new Error("OBJECT_DATAGRAM properties are empty");
+			parts.push(encodeVarint(this.properties.byteLength, version), this.properties);
+		}
+		if ("status" in this.body) {
+			if (!legacy(version) && this.body.status !== 0 && this.properties !== undefined) {
+				throw new Error("only a Normal OBJECT_DATAGRAM may carry properties");
+			}
+			parts.push(encodeVarint(this.body.status, version));
+		} else {
+			// Runs to the datagram boundary: written raw, no length prefix.
+			parts.push(this.body.payload);
+		}
+
+		const out = new Uint8Array(parts.reduce((total, part) => total + part.byteLength, 0));
+		let offset = 0;
+		for (const part of parts) {
+			out.set(part, offset);
+			offset += part.byteLength;
+		}
+		return out;
+	}
+
+	/**
+	 * Decode one QUIC datagram's bytes. Anything malformed is the peer breaking the protocol,
+	 * so every failure is a {@link ProtocolViolation}.
+	 */
+	static async decode(data: Uint8Array, version: IetfVersion): Promise<ObjectDatagram> {
+		try {
+			return await new Reader(undefined, data, version).decode((c) => decodeFields(c, version));
+		} catch (err: unknown) {
+			if (err instanceof ProtocolViolation) throw err;
+			throw new ProtocolViolation(`malformed OBJECT_DATAGRAM: ${reason(error(err))}`, { cause: err });
+		}
+	}
+}
+
+/**
+ * Encode a datagram the way a publisher sends it: Object 0 of a group that has no other, so it
+ * ends its group, stamped in `timescale` units when the subscription declared one.
+ */
+export async function encodeDatagram(
+	datagram: Datagram,
+	options: { trackAlias: bigint; publisherPriority: number; timescale?: Timescale },
+	version: IetfVersion,
+): Promise<Uint8Array> {
+	const properties =
+		options.timescale !== undefined
+			? await encodeObjectExtensions(datagram.timestamp, options.timescale, version)
+			: undefined;
+	return new ObjectDatagram({
+		trackAlias: options.trackAlias,
+		groupId: datagram.sequence,
+		publisherPriority: options.publisherPriority,
+		endOfGroup: true,
+		properties: properties?.byteLength ? properties : undefined,
+		body: { payload: datagram.payload },
+	}).encode(version);
+}
+
+function decodeFields(c: Cursor, version: IetfVersion): ObjectDatagram {
+	// Full width, so an out-of-range Type is refused rather than overflowing.
+	const raw = c.u62();
+	const kind = raw <= 0xffn ? Number(raw) : -1;
+	if (kind < 0 || !valid(kind, version)) {
+		throw new ProtocolViolation(`invalid OBJECT_DATAGRAM type 0x${raw.toString(16)}`);
+	}
+
+	const trackAlias = c.u62();
+	const groupId = c.u53();
+	const objectId = kind & ZERO_OBJECT_ID ? undefined : c.u53();
+	const publisherPriority = kind & DEFAULT_PRIORITY ? undefined : c.u8();
+	let properties: Uint8Array | undefined;
+	if (kind & PROPERTIES) {
+		properties = c.read(c.u53());
+		if (properties.byteLength === 0) throw new ProtocolViolation("OBJECT_DATAGRAM properties are empty");
+	}
+
+	let body: DatagramBody;
+	if (kind & STATUS) {
+		const status = c.u53();
+		if (!legacy(version) && status !== 0 && properties !== undefined) {
+			throw new ProtocolViolation("only a Normal OBJECT_DATAGRAM may carry properties");
+		}
+		if (c.remaining > 0) throw new ProtocolViolation(`OBJECT_DATAGRAM has ${c.remaining} trailing bytes`);
+		body = { status };
+	} else {
+		body = { payload: c.read(c.remaining) };
+	}
+
+	return new ObjectDatagram({
+		trackAlias,
+		groupId,
+		objectId,
+		publisherPriority,
+		endOfGroup: (kind & END_OF_GROUP) !== 0,
+		properties,
+		body,
+	});
+}

js/net/src/ietf/index.ts

@@ -1,6 +1,7 @@
 export * from "./adapter.ts";
 export * as Cluster from "./cluster.ts";
 export * from "./connection.ts";
+export * from "./datagram.ts";
 export * from "./fetch.ts";
 export * from "./goaway.ts";
 export * from "./hidden.ts";