feat(net): reset revoked lite streams with UNAUTHORIZED

doc/concept/moq-lite.md

@@ -43,7 +43,9 @@ certificate, or nothing), so a publisher learns before anyone subscribes
 whether its broadcasts can reach the peer. More tokens can be presented later
 without reconnecting; the session's scope is the union of every open token's
 grant, and withdrawing, revoking, or narrowing one withdraws only what it alone
-covered.
+covered. A subscription or fetch that loses access resets with the
+`UNAUTHORIZED` stream code, so the peer can tell it apart from the session
+closing.
 
 A client that publishes a broadcast outside its grant closes the session with
 `UNAUTHORIZED` and names the path in the close reason, rather than waiting

doc/lib/js/net.md

@@ -54,7 +54,7 @@ for (;;) {
 - **Subscriptions** carry a priority, a `Time.Milli` max age, and optional `groups` bounds. Groups arrive out of order and are read frame by frame, with `Error.TooFarBehind` when a reader asks for a frame the group never held and `Error.GroupTooLarge` when a write exceeds the cache budget and aborts the group.
 - **Track ends**: `close()` ends a track at its live edge, while `finishAt(n)` declares the exclusive end ahead of it and still accepts the groups below. A subscriber reads the end with `final()` or awaits `finished()`. A remote track ends only once every group below its end has arrived or was dropped; one reset before its header arrived is skipped after the subscription's max age on moq-lite (one second without one), or after one second on IETF.
 - **Datagrams** on moq-lite 05+ and fetch-by-sequence for history.
-- **Authorization** on moq-lite 06, and on moq-transport draft-17+ when both sides negotiate the [MoQ Auth extension](/draft/moq-auth): an established session's `auth.grant` is a `Getter<Auth.Grant | undefined>` with what the relay lets this side publish and subscribe to, learned right after setup. `auth.add(token)` presents another token without reconnecting and resolves with an `Auth.Token` to `close()` later; it rejects with `Auth.Unsupported` when the peer takes no tokens in band. A connection that publishes a broadcast outside its grant closes with `SessionCode.Unauthorized`, naming the path in the reason.
+- **Authorization** on moq-lite 06, and on moq-transport draft-17+ when both sides negotiate the [MoQ Auth extension](/draft/moq-auth): an established session's `auth.grant` is a `Getter<Auth.Grant | undefined>` with what the relay lets this side publish and subscribe to, learned right after setup. `auth.add(token)` presents another token without reconnecting and resolves with an `Auth.Token` to `close()` later; it rejects with `Auth.Unsupported` when the peer takes no tokens in band. A connection that publishes a broadcast outside its grant closes with `SessionCode.Unauthorized`, naming the path in the reason. On moq-lite, a subscription the grant stops covering resets with `StreamCode.Unauthorized` and the session stays up.
 - **Errors** live under one namespace: a stream reset throws `Error.Stream` with a `StreamCode`, while a session close gives `Error.Session` with a `SessionCode`. The registries are disjoint, so the same number means different things in each, and 64+ is yours. Named conditions such as `Error.TooFarBehind`, `Error.FrameTooLarge`, and `Error.GroupTooLarge` subclass `Error.Stream`, so one `code` check handles a condition raised here or reported by the peer. IETF streams use their own mapping: cancellation sends CANCELLED, other local failures send INTERNAL\_ERROR, and received codes remain opaque.
 - **Paths** with `Path.relative` for the cross-broadcast catalog references hang uses. Path patterns (`Path.Pattern`, `Path.Patterns`) are re-exported from [`@moq/pattern`](https://www.npmjs.com/package/@moq/pattern). Literal `Path` stays a coordinate.
 

doc/lib/rs/moq-net.md

@@ -99,7 +99,8 @@ and the presenter sees `Unsupported`; after a grant, such an update revokes it.
 A client whose origin publishes a broadcast outside its grant closes the
 session with `Unauthorized`, naming the path in the close reason. A grant that
 shrinks withdraws the announcements and cancels the subscriptions it no longer
-covers, and leaves the session up.
+covers, and leaves the session up. On moq-lite each cancelled stream resets with
+`StreamError::Unauthorized`.
 
 ## Patterns
 

drafts/draft-lcurley-moq-lite.md

@@ -325,6 +325,8 @@ Sent when resetting a stream (RESET_STREAM), or when refusing to receive one (ST
 | ------- | ------------- | ----------- |
 |  0x39  | TIMESTAMP_MISMATCH | A frame's timestamp does not match its track's timescale. |
 | ------- | ------------- | ----------- |
+|  0x3A  | UNAUTHORIZED | The [scope](#auth-stream) does not cover this request, or no longer does. The session stays up. |
+| ------- | ------------- | ----------- |
 
 Note that CANCELLED is 0x1, not 0x0: a stream reset with 0x0 is an INTERNAL_ERROR, not a routine cancellation.
 An endpoint terminating a stream because the session is ending SHOULD use SESSION_CLOSED rather than the session's own code, since the two spaces are disjoint.
@@ -512,6 +514,7 @@ The opener withdraws a token by closing or resetting its side of the stream, and
 A grant names the paths the opener may publish to the acceptor and subscribe to from it, relative to the session.
 An endpoint's scope is the union of the grants of its open Auth Streams, and a stream that ends removes its grant from the union.
 When the union shrinks, an endpoint SHOULD withdraw its announcements and cancel its subscriptions that the union no longer covers, keeping the session and everything still covered.
+An endpoint that cancels a subscription, or refuses or stops serving a request, because the union does not cover it resets the stream with UNAUTHORIZED.
 An endpoint that announces a broadcast outside the union once its own setup tokens are answered SHOULD close the session with UNAUTHORIZED, rather than wait for a subscription that will never come.
 
 An acceptor that does not verify a token in band, or cannot express its grant in AUTH_OK, resets the stream, the same as a peer that does not support the Auth Stream (see [STREAM_TYPE](#stream_type)).
@@ -1426,6 +1429,7 @@ The `Message Length` describes the payload size on the wire.
 - Split the reserved stream error range: 32 through 47 stays reserved, and 48 through 63 is moq-lite's own, assigned by the tables and mapped rather than forwarded across a bridge. Assigned 0x30 NO_CAPACITY there: it permits one re-resolution within the tier excluding the refusing advertiser, and a receiver that has spent or lacks that retry resets downstream with another code. Assigned 0x32 GROUP_TOO_LARGE: a group that grew past the publisher's cache budget is aborted. Every other code is terminal.
 - Assigned 0x33 NOT_FOUND, 0x34 OLD, and 0x35 EVICTED in the stream error table: a group the publisher cannot serve because it was never here, has been superseded, or was dropped under memory pressure.
 - Assigned 0x36 UNROUTABLE, 0x37 WRONG_SIZE, 0x38 FRAME_TOO_LARGE, and 0x39 TIMESTAMP_MISMATCH in the stream error table, moving them out of the reserved 32 through 47 range, which no longer carries provisional placeholders.
+- Assigned 0x3A UNAUTHORIZED in the stream error table: a request reset because the scope does not cover it, or no longer does, distinct from SESSION_CLOSED.
 - Assigned 0x31 CONTROL_TIMEOUT in the stream error table: a request stream torn down because the peer never answered, which DELIVERY_TIMEOUT described as late content. It has no moq-transport value and bridges to INTERNAL_ERROR.
 - A disallowed stream type, a role mismatch, or a missing extension is a PROTOCOL_VIOLATION; the session table gains no code for them, so nothing is sent from the reserved 32 through 47 range in either registry.
 - Added implicit Announce IDs: each ANNOUNCE_START assigns the next per-stream ordinal.

js/net/src/auth.test.ts

@@ -1,13 +1,16 @@
-import { describe, expect, test } from "bun:test";
+import { describe, expect, spyOn, test } from "bun:test";
 import type { Getter } from "@moq/signals";
 import { type Grant, type Issued, Unsupported } from "./auth.ts";
 import { accept as acceptSession, connect as connectSession, type Established } from "./connection/index.ts";
-import { SessionCode, SessionError } from "./error.ts";
+import { SessionCode, SessionError, StreamCode, toStreamCode } from "./error.ts";
 import * as Ietf from "./ietf/index.ts";
 import * as Lite from "./lite/index.ts";
 import { createMockTransportPair, type MockTransport } from "./mock.ts";
 import { Producer as OriginProducer } from "./origin.ts";
 import * as Path from "./path.ts";
+import { Writer } from "./stream.ts";
+import { Timestamp } from "./time.ts";
+import { wireOf } from "./wire.ts";
 
 const url = new URL("https://localhost:4443/test");
 
@@ -231,3 +234,67 @@ test.each([Lite.ALPN_05, Ietf.ALPN.DRAFT_16])("%s has no grant", async (protocol
 	client.close();
 	server.close();
 });
+
+// moq-transport has no stream code for it, so only moq-lite resets with UNAUTHORIZED.
+test("a revoked grant resets its subscriptions with UNAUTHORIZED", async () => {
+	const clientOrigin = new OriginProducer();
+	const up = clientOrigin.createBroadcast(Path.from("up/y"));
+	up.announce();
+	const upTrack = up.createTrack("video");
+
+	const serverOrigin = new OriginProducer();
+	const down = serverOrigin.createBroadcast(Path.from("room/x"));
+	down.announce();
+	const downTrack = down.createTrack("video");
+
+	const { client, server } = await connect({
+		publish: clientOrigin,
+		serverPublish: serverOrigin,
+		protocol: Lite.ALPN_06,
+	});
+	const requests = server.auth.requests();
+	const issued: Issued[] = [];
+	void (async () => {
+		for (;;) {
+			const request = await requests.next();
+			if (!request) break;
+			issued.push(request.accept(grant(["up"], ["room"])));
+		}
+	})();
+	await waitFor(client.auth.grant, (g) => g !== undefined && g.subscribe.size > 0);
+
+	const frame = { payload: new Uint8Array([1]), timestamp: Timestamp.fromMillis(0) };
+	downTrack.appendGroup().writeFrame(frame);
+	upTrack.appendGroup().writeFrame(frame);
+
+	// The client subscribes to the server, and the server to the client.
+	const downSub = wireOf(client).consume(Path.from("room/x")).track("video").subscribe().ordered();
+	const upSub = wireOf(server).consume(Path.from("up/y")).track("video").subscribe().ordered();
+	expect(await downSub.nextGroup()).toBeDefined();
+	expect(await upSub.nextGroup()).toBeDefined();
+
+	const resets = spyOn(Writer.prototype, "reset");
+	try {
+		issued[0]?.revoke(SessionCode.Unauthorized, "expired");
+		const drained = async (track: typeof downSub) => {
+			for (;;) if (!(await track.nextGroup())) break;
+		};
+		await Promise.all([drained(downSub).catch(() => void 0), drained(upSub).catch(() => void 0)]);
+
+		const reasons = resets.mock.calls.map(([reason]) => reason);
+		// Both sides of the revocation: the subscription the client cancels, and the one it
+		// stops serving. Nothing reads as the session closing.
+		const messages = reasons.map((reason) => (reason as Error).message);
+		expect(messages).toContain("unauthorized: room/x");
+		expect(messages).toContain("unauthorized: up/y");
+		for (const reason of reasons) {
+			expect([StreamCode.Unauthorized, StreamCode.Cancel]).toContain(toStreamCode(reason));
+		}
+	} finally {
+		resets.mockRestore();
+		downSub.close();
+		upSub.close();
+		client.close();
+		server.close();
+	}
+});

js/net/src/error.test.ts

@@ -228,6 +228,7 @@ test("the code tables match the spec", () => {
 		StreamCode.Old,
 		StreamCode.Evicted,
 		StreamCode.FrameTooLarge,
+		StreamCode.Unauthorized,
 	];
 	for (const code of Object.values(StreamCode)) {
 		if (assignedLite.includes(code)) {
@@ -245,6 +246,7 @@ test("the code tables match the spec", () => {
 	expect(Number(StreamCode.Old)).toBe(0x34);
 	expect(Number(StreamCode.Evicted)).toBe(0x35);
 	expect(Number(StreamCode.FrameTooLarge)).toBe(0x38);
+	expect(Number(StreamCode.Unauthorized)).toBe(0x3a);
 
 	// The spaces are disjoint: 0 ends a session cleanly but fails a stream.
 	expect(Number(SessionCode.Cancel)).not.toBe(Number(StreamCode.Cancel));

js/net/src/error.ts

@@ -92,6 +92,8 @@ export const StreamCode = Object.freeze(
 		Evicted: 0x35 as StreamCode,
 		/** A frame declared a payload larger than the receiver accepts. */
 		FrameTooLarge: 0x38 as StreamCode,
+		/** The grant does not cover this request, or no longer does. The session stays up. */
+		Unauthorized: 0x3a as StreamCode,
 		/** The publisher could serve this request but has no capacity for it now. */
 		NoCapacity: 0x30 as StreamCode,
 		/** A group grew past its cache budget and was aborted. */
@@ -332,6 +334,16 @@ export function controlTimeout(cause: unknown): Stream {
 	return new Stream(StreamCode.ControlTimeout, { cause, message });
 }
 
+/**
+ * The {@link StreamCode.Unauthorized} error for a request the grant does not cover, naming
+ * the broadcast. Unlike {@link SessionCode.Unauthorized}, it ends only this stream.
+ *
+ * @internal
+ */
+export function unauthorized(broadcast: string): Stream {
+	return new Stream(StreamCode.Unauthorized, { message: `unauthorized: ${broadcast}` });
+}
+
 /**
  * Decode a transport failure into a {@link Stream} when it carries a stream reset code,
  * otherwise pass it through.

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

@@ -182,6 +182,7 @@ test("stream reset codes are shared only where the draft agrees", () => {
 			StreamCode.Evicted,
 			StreamCode.FrameTooLarge,
 			StreamCode.GroupTooLarge,
+			StreamCode.Unauthorized,
 		]) {
 			expect(sharedStreamCode(code, version)).toBe(false);
 		}

js/net/src/ietf/subscriber.ts

@@ -3,7 +3,7 @@ import * as announce from "../announced.ts";
 import type { Grant } from "../auth.ts";
 import * as broadcast from "../broadcast.ts";
 import { BroadcastCache } from "../consume.ts";
-import { controlTimeout, error, ProtocolViolation, reason, SessionCode, SessionError } from "../error.ts";
+import { controlTimeout, error, ProtocolViolation, reason, unauthorized } from "../error.ts";
 import * as netGroup from "../group.ts";
 import { Cost, type Route, routesEqual, UNKNOWN_HOP } from "../hop.ts";
 import { hiddenBelow, hooks, scopeCaptures, scopeHead, scopeOverlaps } from "../internal.ts";
@@ -495,9 +495,9 @@ export class Subscriber {
 	}
 
 	async #runSubscribe(broadcast: Path.Valid, request: track.Request) {
-		const unauthorized = new SessionError(SessionCode.Unauthorized, { reason: broadcast });
+		const refused = unauthorized(broadcast);
 		if (this.#denied(broadcast)) {
-			request.reject(unauthorized);
+			request.reject(refused);
 			return;
 		}
 
@@ -646,7 +646,7 @@ export class Subscriber {
 			// straight out of its loop for the same reason.
 			if (terminal === revokedEnded) {
 				console.info(`subscription no longer authorized: broadcast=${broadcast} track=${request.name}`);
-				producer.close(unauthorized);
+				producer.close(refused);
 			} else {
 				producer.close();
 			}

js/net/src/lite/publisher.ts

@@ -2,7 +2,7 @@ import { type Dispose, type Getter, race, Signal } from "@moq/signals";
 import type { Grant } from "../auth.ts";
 import { enforceGrant } from "../auth_session.ts";
 import type * as broadcast from "../broadcast.ts";
-import { error, NotFound, reason, SessionCode, SessionError, StreamCode, StreamError } from "../error.ts";
+import { error, NotFound, reason, StreamCode, StreamError, unauthorized } from "../error.ts";
 import type * as group from "../group.ts";
 import { type Hop, type Route, routesEqual } from "../hop.ts";
 import { hiddenBelow, hooks } from "../internal.ts";
@@ -587,7 +587,7 @@ export class Publisher {
 		// Checked before resolving, so a denied request never reaches the origin.
 		const denied = () => this.#denied(msg.broadcast);
 		if (denied()) {
-			stream.writer.reset(new SessionError(SessionCode.Unauthorized, { reason: msg.broadcast }));
+			stream.writer.reset(unauthorized(msg.broadcast));
 			return;
 		}
 
@@ -624,7 +624,7 @@ export class Publisher {
 		let datagrams = Promise.resolve();
 		let controls: SubscriptionControls | undefined;
 
-		const revoked = new SessionError(SessionCode.Unauthorized, { reason: msg.broadcast });
+		const revoked = unauthorized(msg.broadcast);
 		const disposeGrant = this.#grant?.subscribe(() => {
 			if (!denied()) return;
 			console.debug(`publish revoked: broadcast=${msg.broadcast} track=${track.name}`);
@@ -721,7 +721,7 @@ export class Publisher {
 			return;
 		}
 		if (this.#denied(msg.broadcast)) {
-			stream.writer.reset(new SessionError(SessionCode.Unauthorized, { reason: msg.broadcast }));
+			stream.writer.reset(unauthorized(msg.broadcast));
 			return;
 		}
 
@@ -946,7 +946,7 @@ export class Publisher {
 	 */
 	async runTrackInfo(msg: TrackMessage, stream: Stream) {
 		if (this.#denied(msg.broadcast)) {
-			stream.writer.reset(new SessionError(SessionCode.Unauthorized, { reason: msg.broadcast }));
+			stream.writer.reset(unauthorized(msg.broadcast));
 			return;
 		}
 		try {

js/net/src/lite/subscriber.ts

@@ -10,10 +10,9 @@ import {
 	error,
 	ProtocolViolation,
 	reason,
-	SessionCode,
-	SessionError,
 	StreamCode,
 	StreamError,
+	unauthorized,
 } from "../error.ts";
 import * as netGroup from "../group.ts";
 import { Cost, type Hop, MAX_HOPS, type Route, routesEqual, UNKNOWN_HOP } from "../hop.ts";
@@ -527,9 +526,9 @@ export class Subscriber {
 			request.reject(new Error(EMPTY_RANGE));
 			return;
 		}
-		const unauthorized = new SessionError(SessionCode.Unauthorized, { reason: broadcast });
+		const refused = unauthorized(broadcast);
 		if (this.#denied(broadcast)) {
-			request.reject(unauthorized);
+			request.reject(refused);
 			return;
 		}
 
@@ -586,8 +585,8 @@ export class Subscriber {
 		const disposeGrant = this.#grant?.subscribe(() => {
 			if (!this.#denied(broadcast)) return;
 			console.debug(`subscribe revoked: id=${id} broadcast=${broadcast} track=${request.name}`);
-			producer.close(unauthorized);
-			stream.abort(unauthorized);
+			producer.close(refused);
+			stream.abort(refused);
 		});
 		try {
 			// Watch for subscription changes and send SUBSCRIBE_UPDATE. Lite01/Lite02

quest/m1/auth/README.md

@@ -90,8 +90,6 @@ existing lite-06 ALPN.
 
 ## Quests
 
-- [Unauthorized reset](/quest/m1/auth/unauthorized.md) - a subscription that
-  loses access resets with a dedicated UNAUTHORIZED stream code
 - [Path patterns](/quest/m1/auth/patterns.md) - one matcher for every path
   predicate, and AUTH_OK carries pattern grants from AUTH's first release
 - [Relay tokens](/quest/m1/auth/relay-refresh.md) - the relay verifies tokens

quest/m1/auth/relay-refresh.md

@@ -84,5 +84,3 @@ Additive.
 
 - [Origin narrowing](/quest/m1/origin-narrowing.md) - the live re-scope a shrinking token union needs, so no temporary close-on-shrink policy ships
 - [Pattern interest](/quest/m1/auth/patterns.md) - AUTH can represent the complete grants relay revalidation returns
-- [Unauthorized reset](/quest/m1/auth/unauthorized.md) - the code this
-  relay's revocations reset with

rs/libmoq/src/error.rs

@@ -400,6 +400,7 @@ fn stream_kind(err: &moq_net::StreamError) -> moq_protocol_kind {
 		moq_net::StreamError::WrongSize => MOQ_PROTOCOL_KIND_WRONG_SIZE,
 		moq_net::StreamError::FrameTooLarge => MOQ_PROTOCOL_KIND_FRAME_TOO_LARGE,
 		moq_net::StreamError::TimestampMismatch => MOQ_PROTOCOL_KIND_TIMESTAMP_MISMATCH,
+		moq_net::StreamError::Unauthorized => MOQ_PROTOCOL_KIND_UNAUTHORIZED,
 		moq_net::StreamError::App(_) => MOQ_PROTOCOL_KIND_APP,
 		moq_net::StreamError::Unknown(_) => MOQ_PROTOCOL_KIND_UNKNOWN,
 		_ => MOQ_PROTOCOL_KIND_UNKNOWN,