From b4a6a414e255aedf818199c25bd2dbd796e192e6 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 25 Sep 2026 11:46:53 -0700 Subject: [PATCH 1/5] quest: claim m1/auth/unauthorized Co-Authored-By: Claude Opus 5.5 From 179de87abf63bf5a89df36aff72479ce36808b0c Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 25 Sep 2026 12:21:13 -0700 Subject: [PATCH 2/5] feat(net): reset revoked lite streams with UNAUTHORIZED Assign 0x3A UNAUTHORIZED in moq-lite's stream error table and send it when a subscription, fetch, or track request loses access, instead of SESSION_CLOSED (Rust) or INTERNAL_ERROR (JS). Co-Authored-By: Claude Opus 5.5 --- doc/concept/moq-lite.md | 4 +- doc/lib/js/net.md | 2 +- doc/lib/rs/moq-net.md | 3 +- drafts/draft-lcurley-moq-lite.md | 4 ++ js/net/src/auth.test.ts | 71 ++++++++++++++++++++++++++++++- js/net/src/error.test.ts | 2 + js/net/src/error.ts | 12 ++++++ js/net/src/ietf/error.test.ts | 1 + js/net/src/ietf/subscriber.ts | 8 ++-- js/net/src/lite/publisher.ts | 10 ++--- js/net/src/lite/subscriber.ts | 11 +++-- quest/m1/auth/README.md | 2 - quest/m1/auth/relay-refresh.md | 2 - quest/m1/auth/unauthorized.md | 21 --------- rs/libmoq/src/error.rs | 1 + rs/libmoq/src/test.rs | 6 +++ rs/moq-ffi/src/error.rs | 16 ++++++- rs/moq-net/src/coding/reader.rs | 7 +-- rs/moq-net/src/error.rs | 16 +++++-- rs/moq-net/src/ietf/error.rs | 23 +++++++++- rs/moq-net/src/lite/publisher.rs | 3 ++ rs/moq-net/src/lite/subscriber.rs | 8 ++-- rs/moq-net/tests/auth.rs | 20 ++++++++- rs/moq-net/tests/support/mock.rs | 33 +++++++++++--- 24 files changed, 220 insertions(+), 66 deletions(-) delete mode 100644 quest/m1/auth/unauthorized.md diff --git a/doc/concept/moq-lite.md b/doc/concept/moq-lite.md index 1fef47c65e..7837d70dc5 100644 --- a/doc/concept/moq-lite.md +++ b/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 diff --git a/doc/lib/js/net.md b/doc/lib/js/net.md index 40d31fe866..e24d431fce 100644 --- a/doc/lib/js/net.md +++ b/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` 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` 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. diff --git a/doc/lib/rs/moq-net.md b/doc/lib/rs/moq-net.md index 36bd679526..be18b198ae 100644 --- a/doc/lib/rs/moq-net.md +++ b/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 diff --git a/drafts/draft-lcurley-moq-lite.md b/drafts/draft-lcurley-moq-lite.md index d08749f81e..41dacc6fc8 100644 --- a/drafts/draft-lcurley-moq-lite.md +++ b/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. diff --git a/js/net/src/auth.test.ts b/js/net/src/auth.test.ts index 2c83af8c6f..208f82186f 100644 --- a/js/net/src/auth.test.ts +++ b/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(); + } +}); diff --git a/js/net/src/error.test.ts b/js/net/src/error.test.ts index 61bb89b056..ec84dd75dc 100644 --- a/js/net/src/error.test.ts +++ b/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)); diff --git a/js/net/src/error.ts b/js/net/src/error.ts index 9a65d22933..fcfe377526 100644 --- a/js/net/src/error.ts +++ b/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. diff --git a/js/net/src/ietf/error.test.ts b/js/net/src/ietf/error.test.ts index dbee25afa3..12d415cbcb 100644 --- a/js/net/src/ietf/error.test.ts +++ b/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); } diff --git a/js/net/src/ietf/subscriber.ts b/js/net/src/ietf/subscriber.ts index 94d31a740c..d6b38911ef 100644 --- a/js/net/src/ietf/subscriber.ts +++ b/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(); } diff --git a/js/net/src/lite/publisher.ts b/js/net/src/lite/publisher.ts index 606b494d84..2aba7ef9a7 100644 --- a/js/net/src/lite/publisher.ts +++ b/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 { diff --git a/js/net/src/lite/subscriber.ts b/js/net/src/lite/subscriber.ts index 06bd57a672..9954f43d10 100644 --- a/js/net/src/lite/subscriber.ts +++ b/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 diff --git a/quest/m1/auth/README.md b/quest/m1/auth/README.md index d9aa7c2573..413a92caf6 100644 --- a/quest/m1/auth/README.md +++ b/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 diff --git a/quest/m1/auth/relay-refresh.md b/quest/m1/auth/relay-refresh.md index 8c319a7883..83bed06a1f 100644 --- a/quest/m1/auth/relay-refresh.md +++ b/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 diff --git a/quest/m1/auth/unauthorized.md b/quest/m1/auth/unauthorized.md deleted file mode 100644 index 7a9a275413..0000000000 --- a/quest/m1/auth/unauthorized.md +++ /dev/null @@ -1,21 +0,0 @@ -# [S] A subscription that loses access resets with UNAUTHORIZED - -## Goal - -When a grant shrinks and a subscription, fetch, or announce loses access, -its stream resets with a dedicated `UNAUTHORIZED` stream code, so the peer -tells revocation apart from a session closing. Today it resets with -`0x3 SESSION_CLOSED` (`StreamError::Session(Unauthorized)`). - -## Plan - -Assign `0x3A UNAUTHORIZED` in the lite draft's stream error table, the next -code in the application block after `TIMESTAMP_MISMATCH`, on lite-06: a -published lite-06 peer already maps an unknown stream code to a generic -error, so the addition is compatible. Add the matching `StreamError` -variant (the enum is non-exhaustive) in Rust and JS, send it from every -revocation path the lite stream added, and cover it in the Rust and JS -tests. Map it to the existing `Unauthorized` protocol kind in -`stream_kind` in `rs/moq-ffi/src/error.rs` and `rs/libmoq/src/error.rs`, -whose wildcard arms would otherwise report `Unknown`, with a test for each. -Update `drafts/draft-lcurley-moq-lite.md` and run `just drafts check`. diff --git a/rs/libmoq/src/error.rs b/rs/libmoq/src/error.rs index ddea6193fa..87d4cb0e76 100644 --- a/rs/libmoq/src/error.rs +++ b/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, diff --git a/rs/libmoq/src/test.rs b/rs/libmoq/src/test.rs index 535d3574d8..bbbfe45c4f 100644 --- a/rs/libmoq/src/test.rs +++ b/rs/libmoq/src/test.rs @@ -333,6 +333,12 @@ fn last_error_protocol_captures_session_known_app_and_unknown() { 0x2, moq_protocol_kind::MOQ_PROTOCOL_KIND_UNAUTHORIZED as u32, ), + ( + moq_net::Error::from(moq_net::StreamError::Unauthorized), + moq_error_scope::MOQ_ERROR_SCOPE_STREAM as u32, + 0x3a, + moq_protocol_kind::MOQ_PROTOCOL_KIND_UNAUTHORIZED as u32, + ), ( moq_net::Error::from(moq_net::SessionError::App(7)), moq_error_scope::MOQ_ERROR_SCOPE_SESSION as u32, diff --git a/rs/moq-ffi/src/error.rs b/rs/moq-ffi/src/error.rs index cce70fb82f..87731e55c3 100644 --- a/rs/moq-ffi/src/error.rs +++ b/rs/moq-ffi/src/error.rs @@ -25,7 +25,7 @@ pub enum MoqProtocolKind { Cancel, /// Something went wrong that isn't worth a dedicated code. Session 1, stream 0. Internal, - /// The peer's token does not grant the requested path or operation. + /// The peer's token does not grant the requested path or operation. Session 2, stream 0x3A. Unauthorized, /// The peer broke a protocol rule; the session is unusable. ProtocolViolation, @@ -143,6 +143,7 @@ fn stream_kind(err: &moq_net::StreamError) -> MoqProtocolKind { moq_net::StreamError::WrongSize => MoqProtocolKind::WrongSize, moq_net::StreamError::FrameTooLarge => MoqProtocolKind::FrameTooLarge, moq_net::StreamError::TimestampMismatch => MoqProtocolKind::TimestampMismatch, + moq_net::StreamError::Unauthorized => MoqProtocolKind::Unauthorized, moq_net::StreamError::App(_) => MoqProtocolKind::App, moq_net::StreamError::Unknown(_) => MoqProtocolKind::Unknown, _ => MoqProtocolKind::Unknown, @@ -467,6 +468,19 @@ mod tests { } } + #[test] + fn stream_unauthorized_keeps_its_kind() { + let err = MoqError::from(moq_net::Error::from(moq_net::StreamError::Unauthorized)); + match err { + MoqError::Protocol { details: protocol } => { + assert_eq!(protocol.scope, MoqErrorScope::Stream); + assert_eq!(protocol.code, 0x3a); + assert_eq!(protocol.kind, MoqProtocolKind::Unauthorized); + } + other => panic!("expected Protocol, got {other}"), + } + } + #[test] fn stream_app_code_is_verbatim() { let err = MoqError::from(moq_net::Error::from(moq_net::StreamError::App(7))); diff --git a/rs/moq-net/src/coding/reader.rs b/rs/moq-net/src/coding/reader.rs index 2005bd11ae..726e83995f 100644 --- a/rs/moq-net/src/coding/reader.rs +++ b/rs/moq-net/src/coding/reader.rs @@ -301,7 +301,7 @@ mod tests { /// STOP_SENDING refuses a stream, so it must carry a stream code. Reusing the session /// table here would have the peer decode it against the wrong registry: `Cancel` would - /// arrive as INTERNAL_ERROR, and `Unauthorized` as KEY_VALUE_FORMATTING_ERROR. + /// arrive as INTERNAL_ERROR, and `Unauthorized` as DELIVERY_TIMEOUT. #[test] fn abort_stops_with_a_stream_code() { const VERSION: crate::lite::Version = crate::lite::Version::Lite05; @@ -309,10 +309,7 @@ mod tests { for (err, expected) in [ (Error::Cancel, StreamError::Cancel.to_code()), (Error::Lagged, StreamError::TooFarBehind.to_code()), - ( - Error::Unauthorized, - StreamError::Session(crate::SessionError::Unauthorized).to_code(), - ), + (Error::Unauthorized, StreamError::Unauthorized.to_code()), ] { let mut reader = Reader::new(StopLog::default(), VERSION); reader.abort(&err); diff --git a/rs/moq-net/src/error.rs b/rs/moq-net/src/error.rs index 148a857f64..b09ef3e5fb 100644 --- a/rs/moq-net/src/error.rs +++ b/rs/moq-net/src/error.rs @@ -175,6 +175,11 @@ pub enum StreamError { #[error("frame timestamp doesn't match track timescale")] TimestampMismatch, + /// The grant does not cover this subscription, fetch, or announcement, or no longer + /// does. Unlike [`SessionError::Unauthorized`], the session stays up. + #[error("unauthorized")] + Unauthorized, + /// An application-chosen code, offset into the 64+ range on the wire. #[error("app code={0}")] App(u16), @@ -207,6 +212,7 @@ impl StreamError { Self::WrongSize => 0x37, Self::FrameTooLarge => 0x38, Self::TimestampMismatch => 0x39, + Self::Unauthorized => 0x3a, Self::App(app) => *app as u32 + 64, Self::Unknown(code) => *code, } @@ -238,6 +244,7 @@ impl StreamError { 0x37 => Self::WrongSize, 0x38 => Self::FrameTooLarge, 0x39 => Self::TimestampMismatch, + 0x3a => Self::Unauthorized, code @ 64.. => match u16::try_from(code - 64) { Ok(app) => Self::App(app), Err(_) => Self::Unknown(code), @@ -573,6 +580,8 @@ impl From<&Error> for StreamError { Error::FrameTooLarge => Self::FrameTooLarge, Error::GroupTooLarge => Self::GroupTooLarge, Error::TimestampMismatch => Self::TimestampMismatch, + // Losing access ends this stream, not the session. + Error::Unauthorized => Self::Unauthorized, Error::Timeout => Self::DeliveryTimeout, Error::GoingAway => Self::GoingAway, // Our own parse failure is, from the peer's side, a malformed track. @@ -587,8 +596,7 @@ impl From<&Error> for StreamError { // A stream refused on its own is not a session failure, so it does not claim // SESSION_CLOSED; there is no stream-scoped PROTOCOL_VIOLATION to send instead. Error::UnexpectedStream => Self::Internal, - Error::Unauthorized - | Error::Version + Error::Version | Error::UnknownAlpn(_) | Error::TooManyParameters | Error::GoawayTimeout @@ -676,6 +684,7 @@ mod tests { StreamError::WrongSize, StreamError::FrameTooLarge, StreamError::TimestampMismatch, + StreamError::Unauthorized, StreamError::App(7), ]; for err in registered { @@ -694,6 +703,7 @@ mod tests { (StreamError::WrongSize, 0x37), (StreamError::FrameTooLarge, 0x38), (StreamError::TimestampMismatch, 0x39), + (StreamError::Unauthorized, 0x3a), ] { assert_eq!(err.to_code(), code, "{err:?} moved off its assigned code"); } @@ -730,7 +740,7 @@ mod tests { // Registered stream codes survive the hop unchanged. for code in [ - 0x0, 0x1, 0x2, 0x4, 0x5, 0x12, 0x31, 0x32, 0x33, 0x34, 0x35, 0x36, 0x37, 0x38, 0x39, + 0x0, 0x1, 0x2, 0x4, 0x5, 0x12, 0x31, 0x32, 0x33, 0x34, 0x35, 0x36, 0x37, 0x38, 0x39, 0x3a, ] { let relayed = StreamError::from(&Error::from(StreamError::from_code(code))); assert_eq!(relayed.to_code(), code, "stream {code:#x} changed across a relay"); diff --git a/rs/moq-net/src/ietf/error.rs b/rs/moq-net/src/ietf/error.rs index 66f1163408..e6358fd60c 100644 --- a/rs/moq-net/src/ietf/error.rs +++ b/rs/moq-net/src/ietf/error.rs @@ -284,7 +284,9 @@ pub(crate) mod request { /// would say something the peer's registry gives a different meaning. pub(crate) fn to_code(err: &Error, kind: Kind, version: Version) -> u64 { let registered = match err { - Error::Unauthorized | Error::Session(SessionError::Unauthorized) => return UNAUTHORIZED, + Error::Unauthorized + | Error::Session(SessionError::Unauthorized) + | Error::Stream(StreamError::Unauthorized) => return UNAUTHORIZED, Error::Timeout | Error::Stream(StreamError::DeliveryTimeout) | Error::Session(SessionError::Timeout) => { return TIMEOUT; } @@ -375,6 +377,20 @@ pub(crate) mod request { Error::Remote(0x30), ]; + /// A relay bridging a moq-lite revocation onto a moq-transport request refuses it as + /// UNAUTHORIZED, the same as a local refusal. + #[test] + fn a_bridged_revocation_is_unauthorized() { + for version in ALL { + for kind in KINDS { + assert_eq!( + to_code(&Error::Stream(StreamError::Unauthorized), kind, version), + UNAUTHORIZED + ); + } + } + } + /// Draft-14 numbers a missing track 0x4 and draft-15 moved it to 0x10, which is /// draft-14's MALFORMED_AUTH_TOKEN. Getting this backwards tells a peer its token is /// broken when the broadcast simply is not here, so it re-authenticates instead of @@ -702,6 +718,7 @@ mod tests { StreamError::FrameTooLarge, StreamError::GroupTooLarge, StreamError::TimestampMismatch, + StreamError::Unauthorized, StreamError::ControlTimeout, StreamError::App(7), StreamError::Unknown(0x1234), @@ -728,6 +745,7 @@ mod tests { 0x37, 0x38, 0x39, + 0x3a, 64 + 7, ] { assert_eq!(from_stream_code(code, Version::Draft20), StreamError::Unknown(code)); @@ -740,7 +758,7 @@ mod tests { /// Every stream error this crate can hold, so the conformance check below covers the /// whole space rather than the variants someone remembered. A new variant belongs here. - const EVERY_ERROR: [StreamError; 18] = [ + const EVERY_ERROR: [StreamError; 19] = [ StreamError::Session(SessionError::Cancel), StreamError::Internal, StreamError::Cancel, @@ -757,6 +775,7 @@ mod tests { StreamError::FrameTooLarge, StreamError::GroupTooLarge, StreamError::TimestampMismatch, + StreamError::Unauthorized, StreamError::App(7), StreamError::Unknown(0x22), ]; diff --git a/rs/moq-net/src/lite/publisher.rs b/rs/moq-net/src/lite/publisher.rs index 12039ab3bf..5d2b167e4b 100644 --- a/rs/moq-net/src/lite/publisher.rs +++ b/rs/moq-net/src/lite/publisher.rs @@ -1391,6 +1391,9 @@ impl SubscribeServe { | Error::Transport(_) => { tracing::info!(id = self.id, broadcast = %self.absolute, track = %self.track, "subscribed cancelled") } + Error::Unauthorized | Error::Stream(crate::StreamError::Unauthorized) => { + tracing::info!(id = self.id, broadcast = %self.absolute, track = %self.track, "subscribed unauthorized") + } err => { tracing::warn!(id = self.id, broadcast = %self.absolute, track = %self.track, %err, "subscribed error") } diff --git a/rs/moq-net/src/lite/subscriber.rs b/rs/moq-net/src/lite/subscriber.rs index 589278c4fd..2ad01b183f 100644 --- a/rs/moq-net/src/lite/subscriber.rs +++ b/rs/moq-net/src/lite/subscriber.rs @@ -3062,11 +3062,13 @@ impl TrackServeRun { request.reject(Error::Unauthorized); } } - TrackRunState::Serve(mut serve_loop) => { + TrackRunState::Serve(serve_loop) => { let _ = serve_loop.serving.abort(Error::Unauthorized); - if let Sub::Active(active) = &mut serve_loop.sub { + // Reset rather than finish, so the publisher reads a revocation instead of + // a routine unsubscribe. + if let Sub::Active(active) = serve_loop.sub { self.serve.subscriber.remove_subscribe(active.id); - let _ = active.stream.writer.finish(); + active.stream.writer.abort(&Error::Unauthorized); } } TrackRunState::Done => {} diff --git a/rs/moq-net/tests/auth.rs b/rs/moq-net/tests/auth.rs index f682dcd2c9..bc82c843d1 100644 --- a/rs/moq-net/tests/auth.rs +++ b/rs/moq-net/tests/auth.rs @@ -7,7 +7,7 @@ mod support; use std::time::Duration; use moq_net::{ - Client, Error, Hop, Pattern, Patterns, Server, Session, SessionError, Version, + Client, Error, Hop, Pattern, Patterns, Server, Session, SessionError, StreamError, Version, auth::{self, Grant}, origin, }; @@ -654,7 +654,12 @@ async fn a_revoked_grant_cancels_its_subscriptions(version: &'static str) { up_sub.recv_group().await.unwrap().unwrap(); setup.revoke(SessionError::Unauthorized, "expired"); - assert!(down_sub.recv_group().await.is_err(), "subscription outlived its grant"); + let err = down_sub + .recv_group() + .await + .err() + .expect("subscription outlived its grant"); + assert!(matches!(err, Error::Unauthorized), "{err:?}"); // The served side ends too: the client stops serving what it may no longer publish. loop { match up_sub.recv_group().await { @@ -664,6 +669,17 @@ async fn a_revoked_grant_cancels_its_subscriptions(version: &'static str) { } } assert_eq!(pair.client_transport.close_reason(), None); + + // Neither stream claims the session closed. moq-lite resets both with UNAUTHORIZED; + // moq-transport has no such stream code and reports it on the request instead. + let resets = pair.client_transport.resets(); + let closed = StreamError::Session(SessionError::Unauthorized).to_code(); + assert!(!resets.contains(&closed), "{resets:x?}"); + if version == LITE_06 { + let unauthorized = StreamError::Unauthorized.to_code(); + let count = resets.iter().filter(|&&code| code == unauthorized).count(); + assert_eq!(count, 2, "both subscriptions reset with UNAUTHORIZED: {resets:x?}"); + } }) .await .expect("timed out"); diff --git a/rs/moq-net/tests/support/mock.rs b/rs/moq-net/tests/support/mock.rs index 610f79b68b..42ac87c78c 100644 --- a/rs/moq-net/tests/support/mock.rs +++ b/rs/moq-net/tests/support/mock.rs @@ -103,6 +103,8 @@ impl ClosedSignal { /// A mock send stream backed by a queue to the peer's reader. pub struct MockSendStream { tx: Option>, + /// Every code the owning side reset a stream with. + resets: Resets, closed: Arc, park: kio::Park, } @@ -131,6 +133,7 @@ impl poll::SendStream for MockSendStream { fn reset(&mut self, code: u32) { if let Some(tx) = self.tx.take() { + self.resets.lock().unwrap().push(code); let _ = tx.try_push(StreamChunk::Reset(code)); } } @@ -254,13 +257,17 @@ impl Drop for MockRecvStream { // ── Stream pair constructor ───────────────────────────────────────── -/// Create a linked (send, recv) stream pair. -fn new_stream_pair() -> (MockSendStream, MockRecvStream) { +/// The stream reset codes one side has sent. +type Resets = Arc>>; + +/// Create a linked (send, recv) stream pair, logging the sender's resets to `resets`. +fn new_stream_pair(resets: &Resets) -> (MockSendStream, MockRecvStream) { let queue = kio::Queue::new(); let closed = Arc::new(ClosedSignal::default()); let send = MockSendStream { tx: Some(queue.clone()), + resets: resets.clone(), closed: closed.clone(), park: kio::Park::default(), }; @@ -306,6 +313,9 @@ struct SessionSide { protocol: Option<&'static str>, /// Connection-level close state shared with the peer. conn: Arc, + /// Stream resets sent by this side, and by the peer. + resets: Resets, + peer_resets: Resets, } /// An in-memory mock WebTransport session. @@ -369,8 +379,8 @@ impl poll::Session for MockSession { _cx: &mut Context<'_>, ) -> Poll> { // Create two stream pairs: one for each direction. - let (our_send, peer_recv) = new_stream_pair(); - let (peer_send, our_recv) = new_stream_pair(); + let (our_send, peer_recv) = new_stream_pair(&self.side.resets); + let (peer_send, our_recv) = new_stream_pair(&self.side.peer_resets); // Deliver (peer_send, peer_recv) to the peer's accept_bi. match self.side.peer_bidi.try_push((peer_send, peer_recv)) { @@ -380,7 +390,7 @@ impl poll::Session for MockSession { } fn poll_open_uni(&mut self, _cx: &mut Context<'_>) -> Poll> { - let (our_send, peer_recv) = new_stream_pair(); + let (our_send, peer_recv) = new_stream_pair(&self.side.resets); // Deliver peer_recv to the peer's accept_uni. match self.side.peer_uni.try_push(peer_recv) { @@ -457,6 +467,13 @@ impl MockSession { self.side.conn.close_state.lock().unwrap().clone() } + /// Every code this side has reset a stream with, in order. + // Only some test binaries inspect the codes. + #[allow(dead_code)] + pub fn resets(&self) -> Vec { + self.side.resets.lock().unwrap().clone() + } + fn close_error(&self) -> MockError { self.side .conn @@ -491,6 +508,8 @@ pub fn create_mock_session_pair(protocol: Option<&'static str>) -> (MockSession, let s2c_uni = kio::Queue::new(); let c2s_datagrams = kio::Queue::new(); let s2c_datagrams = kio::Queue::new(); + let client_resets = Resets::default(); + let server_resets = Resets::default(); let client_side = Arc::new(SessionSide { bidi: s2c_bidi.clone(), @@ -501,6 +520,8 @@ pub fn create_mock_session_pair(protocol: Option<&'static str>) -> (MockSession, peer_datagrams: c2s_datagrams.clone(), protocol, conn: conn.clone(), + resets: client_resets.clone(), + peer_resets: server_resets.clone(), }); let server_side = Arc::new(SessionSide { @@ -512,6 +533,8 @@ pub fn create_mock_session_pair(protocol: Option<&'static str>) -> (MockSession, peer_datagrams: s2c_datagrams, protocol, conn, + resets: server_resets, + peer_resets: client_resets, }); let new = |side| MockSession { From 1121b301a638ffeacf9b798d608e378a98ca781e Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 25 Sep 2026 12:50:25 -0700 Subject: [PATCH 3/5] fix(net): end an orphaned spliced track with its segment's error A relay's logical track whose producer went away ended its readers with Dropped once the final segment died, discarding why. Keep the segment's error so a revoked upstream surfaces as UNAUTHORIZED. Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/model/resume.rs | 81 ++++++++++++++++------------------ rs/moq-net/tests/auth.rs | 9 +++- 2 files changed, 46 insertions(+), 44 deletions(-) diff --git a/rs/moq-net/src/model/resume.rs b/rs/moq-net/src/model/resume.rs index e2227a17f1..572e934ec4 100644 --- a/rs/moq-net/src/model/resume.rs +++ b/rs/moq-net/src/model/resume.rs @@ -1401,8 +1401,8 @@ impl SegmentSub { } /// Move an active cursor into terminal retention and mark the segment done. - fn complete(&mut self, count: Option) { - let previous = std::mem::replace(&mut self.sub, SubState::Done(count)); + fn complete(&mut self, end: Result) { + let previous = std::mem::replace(&mut self.sub, SubState::Done(end)); if let SubState::Active(sub) = previous { self.terminal = Some(*sub); } @@ -1445,11 +1445,11 @@ enum SubState { /// Live cursor over the underlying track. Boxed: the subscriber dwarfs the other /// variants, and every segment holds this enum. Active(Box), - /// The underlying track ended: `Some` with the group count when it finished - /// cleanly, `None` when it aborted or was dropped. An abort is deliberately - /// not surfaced: a dead route stalls the logical track until the next switch - /// replaces it. - Done(Option), + /// The underlying track ended: `Ok` with the group count when it finished + /// cleanly, the error it died with otherwise. An abort is deliberately not + /// surfaced while a switch can still replace the route; it only decides how an + /// orphaned track ends. + Done(Result), } /// A live subscription spliced across every segment of a logical track. @@ -1460,7 +1460,7 @@ enum SubState { /// the producer [`finish`](Producer::finish)es and the final segment completes. /// The producer itself going away without a terminal state ends the track as its /// final segment's track ended, once the remaining segments drain: cleanly if it -/// finished, [`Error::Dropped`] if it died. With nobody left to splice a +/// finished, with its error if it died. With nobody left to splice a /// replacement, a stall would never end. pub struct Subscriber { state: kio::Consumer, @@ -1531,7 +1531,7 @@ impl Subscriber { break; } if seg.pruned { - seg.sub = SubState::Done(None); + seg.sub = SubState::Done(Err(Error::Dropped)); seg.terminal = None; seg.parked.clear(); cut -= 1; @@ -1827,7 +1827,7 @@ impl Subscriber { // The copy was asked for the cache's newest group and started past it: its // source judged that group stale, so the older cache is no use to live reads. if start.is_some_and(|start| edge.is_some_and(|edge| start > edge)) { - seg.complete(None); + seg.complete(Err(Error::Dropped)); return Poll::Ready(()); } } @@ -1848,7 +1848,7 @@ impl Subscriber { seg.sub = SubState::Active(Box::new(sub)); } // The underlying track was rejected or closed: stall, not error. - Err(_) => seg.sub = SubState::Done(None), + Err(err) => seg.sub = SubState::Done(Err(err)), } } Poll::Ready(()) @@ -1877,18 +1877,17 @@ impl Subscriber { return Poll::Ready(Some(group)); } Ok(None) => { - let count = sub.poll_finished(waiter).map(|res| res.ok()); - let count = match count { - Poll::Ready(count) => count, - Poll::Pending => None, + let end = match sub.poll_finished(waiter) { + Poll::Ready(end) => end, + Poll::Pending => Err(Error::Dropped), }; - seg.complete(count); + seg.complete(end); return Poll::Ready(None); } // A dead segment stalls the logical track rather than erroring; // the next switch resumes it. - Err(_) => { - seg.complete(None); + Err(err) => { + seg.complete(Err(err)); return Poll::Ready(None); } }, @@ -1967,15 +1966,15 @@ impl Subscriber { } // The track ran out at or below the floor: the segment drained. Poll::Ready(Ok(None)) => { - let count = match sub.poll_finished(waiter) { - Poll::Ready(count) => count.ok(), - Poll::Pending => None, + let end = match sub.poll_finished(waiter) { + Poll::Ready(end) => end, + Poll::Pending => Err(Error::Dropped), }; - seg.complete(count); + seg.complete(end); } // A dead segment stalls the logical track rather than erroring; // the next switch resumes it. - Poll::Ready(Err(_)) => seg.complete(None), + Poll::Ready(Err(err)) => seg.complete(Err(err)), Poll::Pending => all_done = false, }, SubState::Pending(_) => all_done = false, @@ -2026,7 +2025,7 @@ impl Subscriber { /// Returns `Poll::Ready(Ok(None))` once every segment completed and the /// producer finished, or was dropped with the final segment's track finished. /// Returns `Poll::Ready(Err(_))` if the producer aborted, or was dropped and - /// the final segment's track died ([`Error::Dropped`]). + /// the final segment's track died (with that track's error). pub fn poll_recv_group(&mut self, waiter: &kio::Waiter) -> Poll>> { self.poll_sync(waiter); @@ -2219,10 +2218,7 @@ impl Subscriber { // The producer is gone: no takeover is coming, so the track ends when // its newest segment does, and the way it did. if self.closed && !pending_activation { - return match ready!(self.poll_final(waiter)) { - Some(_) => Poll::Ready(Ok(None)), - None => Poll::Ready(Err(Error::Dropped)), - }; + return Poll::Ready(ready!(self.poll_final(waiter)).map(|_| None)); } Poll::Pending } @@ -2250,35 +2246,35 @@ impl Subscriber { true => Poll::Ready(Ok(end.unwrap_or(0))), // The producer is gone without finishing: the track ends as its final // segment did. - false => Poll::Ready(end.ok_or(Error::Dropped)), + false => Poll::Ready(end), } } /// Wait for the final segment's track to end: its group count when it - /// finished, `None` when it died or there is no segment. Earlier segments don't + /// finished, the error it died with, or [`Error::Dropped`] when there is no segment. Earlier segments don't /// decide the end. Only the subscription is resolved here: consuming groups, or /// completing the segment, would steal them from a `recv_group` caller on the /// same subscriber. - fn poll_final(&mut self, waiter: &kio::Waiter) -> Poll> { + fn poll_final(&mut self, waiter: &kio::Waiter) -> Poll> { let Some(seg) = self.segments.last_mut() else { - return Poll::Ready(None); + return Poll::Ready(Err(Error::Dropped)); }; ready!(Self::poll_activate(seg, &self.last_prefs, self.min_sequence, waiter)); match &mut seg.sub { - SubState::Done(count) => Poll::Ready(*count), + SubState::Done(end) => Poll::Ready(end.clone()), // Observe only: the cursor may still hold groups, so the read path // completes the segment once it drains. - SubState::Active(sub) => Poll::Ready(ready!(sub.poll_finished(waiter)).ok()), + SubState::Active(sub) => Poll::Ready(ready!(sub.poll_finished(waiter))), SubState::Pending(_) => unreachable!("poll_activate resolved above"), } } /// How a logical track whose producer went away without finishing ends, once - /// every segment drained: cleanly if its final segment's track finished, since - /// that is where the content ended, and [`Error::Dropped`] otherwise. + /// every segment drained: as its final segment's track did, since that is where + /// the content ended, or [`Error::Dropped`] when there is none. fn orphan_end(&self) -> Result<()> { match self.segments.last().map(|seg| &seg.sub) { - Some(SubState::Done(Some(_))) => Ok(()), + Some(SubState::Done(end)) => end.clone().map(|_| ()), _ => Err(Error::Dropped), } } @@ -4880,7 +4876,7 @@ mod test { } #[tokio::test] - async fn dropped_producer_errors_once_drained() { + async fn dropped_producer_errors_once_drained_with_the_segment_error() { let (mut track_a, consumer_a) = track_pair("a"); let mut producer = Producer::new(); @@ -4891,12 +4887,13 @@ mod test { assert_eq!(recv(&mut sub), 0); // The segment dies, then the producer goes away without finish/abort: - // no takeover can ever come, so stalling would hang forever. - track_a.abort(Error::Cancel).unwrap(); + // no takeover can ever come, so stalling would hang forever. The reader + // learns why the segment died, not a generic drop. + track_a.abort(Error::Unauthorized).unwrap(); drop(producer); let result = sub.recv_group().now_or_never().expect("must not stall forever"); - assert!(matches!(result, Err(Error::Dropped))); + assert!(matches!(result, Err(Error::Unauthorized)), "{:?}", result.err()); } #[tokio::test] @@ -4944,7 +4941,7 @@ mod test { let result = sub.finished().now_or_never().expect("must not stall forever"); match clean { true => assert!(matches!(result, Ok(0))), - false => assert!(matches!(result, Err(Error::Dropped))), + false => assert!(matches!(result, Err(Error::Cancel)), "{result:?}"), } } } diff --git a/rs/moq-net/tests/auth.rs b/rs/moq-net/tests/auth.rs index bc82c843d1..a7964daab5 100644 --- a/rs/moq-net/tests/auth.rs +++ b/rs/moq-net/tests/auth.rs @@ -661,12 +661,17 @@ async fn a_revoked_grant_cancels_its_subscriptions(version: &'static str) { .expect("subscription outlived its grant"); assert!(matches!(err, Error::Unauthorized), "{err:?}"); // The served side ends too: the client stops serving what it may no longer publish. - loop { + let err = loop { match up_sub.recv_group().await { Ok(Some(_)) => continue, Ok(None) => panic!("served subscription finished instead of ending"), - Err(_) => break, + Err(err) => break err, } + }; + // The relay's own reader learns the peer's code across the origin's splice, not a + // generic drop. moq-transport reports it in PUBLISH_DONE instead. + if version == LITE_06 { + assert!(matches!(err, Error::Stream(StreamError::Unauthorized)), "{err:?}"); } assert_eq!(pair.client_transport.close_reason(), None); From 11c88473b235fc7f520b5b215f7cb1bb68e7e5ff Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 25 Sep 2026 21:25:18 -0700 Subject: [PATCH 4/5] docs(interop): remark-format the interop readme Co-Authored-By: Grok 4.7 --- test/interop/README.md | 14 +++++++------- 1 file changed, 7 insertions(+), 7 deletions(-) diff --git a/test/interop/README.md b/test/interop/README.md index 725d5740d5..cd6e5a6135 100644 --- a/test/interop/README.md +++ b/test/interop/README.md @@ -80,7 +80,7 @@ browser) runs once more with a token that excludes its broadcast. It must fail loud, logging Unauthorized and naming the path, and every subscriber must time out. -Tokens grant subtrees (`name/**`) because moq-lite-06's AUTH_OK carries prefixes: +Tokens grant subtrees (`name/**`) because moq-lite-06's AUTH\_OK carries prefixes: the relay withholds a literal grant it cannot encode, and the client sees none. ## Running locally @@ -129,26 +129,26 @@ samples, not that a machine played them. Each run covers, against a real local relay: -- **capabilities** - probes every platform API the player needs, and fails +* **capabilities** - probes every platform API the player needs, and fails naming what is missing rather than skipping a case. -- **cold start** - the publisher reports when it is announced and encoding, then +* **cold start** - the publisher reports when it is announced and encoding, then a fresh page joins. No reload, unlike the matrix driver: a subscriber that needs a second page load is an initialization bug, not a race. -- **user gesture** - Chromium is launched with +* **user gesture** - Chromium is launched with `--autoplay-policy=document-user-activation-required`, which applies to top-level Web Audio contexts. The fixture and player graphs must be suspended before either page is clicked, then both must carry audio afterwards. Harness state probes use CDP with `userGesture: false`: Playwright's usual page reads themselves grant activation and would invalidate this assertion. -- **capture permission** - the camera case uses a fake device for deterministic +* **capture permission** - the camera case uses a fake device for deterministic input, while Playwright denies and then grants permissions. Both source errors must be visible; neither a full nor a microphone-only denial may announce a broadcast. Granting both permissions must recover and encode without reload. The fake device is not physical hardware, and the headless permission decision is not a person clicking a browser prompt. -- **pause and resume**, **unsubscribe and rejoin**, **detach and reattach**, +* **pause and resume**, **unsubscribe and rejoin**, **detach and reattach**, **publisher stop and same-path republish**, and **late join**. -- **resources return to baseline** - the page wraps `WebTransport`, `WebSocket`, +* **resources return to baseline** - the page wraps `WebTransport`, `WebSocket`, `AudioContext`, and `Worker` to count live instances, so a detach that leaks a session is visible rather than merely invisible. From 81842e7efd2b486f2231a2c4b3b259a0a47351d2 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 25 Sep 2026 21:28:39 -0700 Subject: [PATCH 5/5] docs(interop): keep remark's hyphen list markers Co-Authored-By: Grok 4.7 --- test/interop/README.md | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/test/interop/README.md b/test/interop/README.md index cd6e5a6135..40bd927b03 100644 --- a/test/interop/README.md +++ b/test/interop/README.md @@ -129,26 +129,26 @@ samples, not that a machine played them. Each run covers, against a real local relay: -* **capabilities** - probes every platform API the player needs, and fails +- **capabilities** - probes every platform API the player needs, and fails naming what is missing rather than skipping a case. -* **cold start** - the publisher reports when it is announced and encoding, then +- **cold start** - the publisher reports when it is announced and encoding, then a fresh page joins. No reload, unlike the matrix driver: a subscriber that needs a second page load is an initialization bug, not a race. -* **user gesture** - Chromium is launched with +- **user gesture** - Chromium is launched with `--autoplay-policy=document-user-activation-required`, which applies to top-level Web Audio contexts. The fixture and player graphs must be suspended before either page is clicked, then both must carry audio afterwards. Harness state probes use CDP with `userGesture: false`: Playwright's usual page reads themselves grant activation and would invalidate this assertion. -* **capture permission** - the camera case uses a fake device for deterministic +- **capture permission** - the camera case uses a fake device for deterministic input, while Playwright denies and then grants permissions. Both source errors must be visible; neither a full nor a microphone-only denial may announce a broadcast. Granting both permissions must recover and encode without reload. The fake device is not physical hardware, and the headless permission decision is not a person clicking a browser prompt. -* **pause and resume**, **unsubscribe and rejoin**, **detach and reattach**, +- **pause and resume**, **unsubscribe and rejoin**, **detach and reattach**, **publisher stop and same-path republish**, and **late join**. -* **resources return to baseline** - the page wraps `WebTransport`, `WebSocket`, +- **resources return to baseline** - the page wraps `WebTransport`, `WebSocket`, `AudioContext`, and `Worker` to count live instances, so a detach that leaks a session is visible rather than merely invisible.