fix(net): consumers see the source's real error, never Dropped

doc/bin/cli.md

@@ -432,7 +432,9 @@ continuity error, and the rest carry on. It takes neither `--linger`,
 
 A stdout export ends with the broadcast. `export ts --linger 10s` waits that
 long for the same publisher instance to come back, and carries on with the same
-stream. A replacement (a restarted publisher, under a new
+stream. A return that cannot be read yet, such as one still routed through a
+dead session, keeps it waiting for the next announcement; if the linger runs
+out first, it exits 1 with that error. A replacement (a restarted publisher, under a new
 [epoch](/concept/moq-lite#publisher-epochs)) exits 1 unless `--stitch` follows
 it as a full program switch: a new PMT from its catalog, with every PID flagging
 the break. `--linger` also bounds a mid-stream stitch from the replacement

drafts/draft-lcurley-moq-lite.md

@@ -351,7 +351,7 @@ Sent when resetting a stream (RESET_STREAM), or when refusing to receive one (ST
 | ------- | ------------- | ----------- |
 |  0x35  | EVICTED | The group was dropped under memory pressure. Unlike OLD it was still current, so it can be re-fetched. |
 | ------- | ------------- | ----------- |
-|  0x36  | UNROUTABLE | The broadcast is neither announced nor served, so there is no route to it. |
+|  0x36  | UNROUTABLE | The broadcast is neither announced nor served, so there is no route to it, or the route serving it was lost, such as when a relay's upstream session ends. |
 | ------- | ------------- | ----------- |
 |  0x37  | WRONG_SIZE | A frame's payload length disagreed with its declared size. |
 | ------- | ------------- | ----------- |
@@ -366,6 +366,7 @@ Sent when resetting a stream (RESET_STREAM), or when refusing to receive one (ST
 
 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.
+A relay terminating a stream because the upstream session serving it ended SHOULD use UNROUTABLE, since the downstream session is not ending.
 
 ## Handshake
 See the [Session](#session) section for ALPN negotiation and session activation details.
@@ -1600,6 +1601,7 @@ The `Message Length` describes the payload size on the wire.
 - A subscriber MAY cancel by FIN on its Subscribe Stream, and the publisher then FINs without SUBSCRIBE_END.
 - Removed the NO_CAPACITY stream error and its one re-resolution. Every refusal is terminal, including at another route of the same prefix.
 - Assigned 0x3B UNAUTHORIZED in the stream error table: a request reset because the endpoint does not authorize it, or no longer does, distinct from SESSION_CLOSED.
+- A relay resets with UNROUTABLE, rather than INTERNAL_ERROR, a stream served through an upstream session that ended.
 - A GOAWAY recipient MAY keep opening requests on the current session until it has moved to a replacement session; previously it MUST NOT open new streams. The sender SHOULD keep answering them and MAY reset one with GOING_AWAY, which the recipient SHOULD retry on another route.
 - Replaced Warm and Cold Route Cost with one static Route Cost in moq-lite-07-wip; lite-06 retains both fields, reading Warm and writing Cold at saturation.
 - Made TRACK_INFO Publisher Max Age optional, encoded as milliseconds plus one with zero meaning no limit.

go/wrapper/errors.go

@@ -101,7 +101,8 @@ func IsShutdown(err error) bool {
 }
 
 // IsAuthError reports whether err is an authentication/authorization failure
-// (HTTP 401/403, an expired token, or a protocol Unauthorized or Expired session close).
+// (HTTP 401/403, an expired token, a protocol Unauthorized or Expired session
+// close, or a track revocation).
 func IsAuthError(err error) bool {
 	if errors.Is(err, ErrUnauthorized) || errors.Is(err, ErrForbidden) || errors.Is(err, ErrExpired) {
 		return true

go/wrapper/errors_test.go

@@ -131,3 +131,12 @@ func TestExpiredIsAuthError(t *testing.T) {
 		t.Fatal("an expired request token is not an auth error")
 	}
 }
+
+func TestTrackRevocationIsAuth(t *testing.T) {
+	err := ffi.NewMoqErrorProtocol(ffi.MoqProtocolError{
+		Scope: moq.ErrorScopeStream, Code: 0x3b, Kind: moq.ProtocolKindUnauthorized, Message: "unauthorized",
+	})
+	if !moq.IsAuthError(err) {
+		t.Fatalf("a track revocation is not an auth error: %v", err)
+	}
+}

js/net/src/error.test.ts

@@ -334,10 +334,17 @@ test("toStreamCode: a local condition maps to the code the peer can act on", ()
 	expect(toStreamCode(new StreamError(StreamCode.GoingAway))).toBe(StreamCode.GoingAway);
 	expect(toStreamCode(new StreamError(StreamCode(70)))).toBe(StreamCode(70));
 
-	// Anything else already means 0, and so does a session code: the registries are disjoint, so
-	// forwarding one onto a stream would mistranslate it (session 0x4 is a stream GOING_AWAY).
+	// Anything else already means 0.
 	expect(toStreamCode(new Error("boom"))).toBe(StreamCode.Internal);
-	expect(toStreamCode(new SessionError(SessionCode.GoawayTimeout))).toBe(StreamCode.Internal);
+	// A session's close reaching a stream came from the route it was served through. Its code is
+	// not forwarded: the registries are disjoint (session 0x4 is a stream GOING_AWAY).
+	expect(toStreamCode(new SessionError(SessionCode.GoawayTimeout))).toBe(StreamCode.Unroutable);
+	// An application code means the same in both registries, so the source's own error survives.
+	expect(toStreamCode(new SessionError(SessionCode(70)))).toBe(StreamCode(70));
+	// So is a session that failed at the transport, with no close code: its route is gone.
+	expect(toStreamCode(Object.assign(new Error("connection lost"), { source: "session" }))).toBe(
+		StreamCode.Unroutable,
+	);
 });
 
 // A control request that outlived its deadline is not late content. It carries its own code so a

js/net/src/error.ts

@@ -354,8 +354,11 @@ interface TransportErrorOptions {
  *
  * Lossy on purpose. An error describes what went wrong here, while the registry is what the peer
  * can act on, so anything without a code of its own degrades to {@link StreamCode.Internal}
- * rather than inventing one. A {@link Session} lands there too: the two registries are
- * disjoint, so forwarding its code onto a stream would mistranslate it.
+ * rather than inventing one. A {@link Session} is {@link StreamCode.Unroutable}: a stream that
+ * can still be reset outlived the session that failed, so that was the route it was served
+ * through, and forwarding the session's code would mistranslate it across disjoint registries.
+ * An application code is the exception, since both registries share the 64+ range. A session
+ * that failed at the transport, with no code at all, is {@link StreamCode.Unroutable} too.
  *
  * On an IETF stream the code has to be one the negotiated draft assigns the same meaning to,
  * so a condition it does not register degrades to INTERNAL_ERROR as well. See
@@ -375,6 +378,11 @@ function localStreamCode(err: unknown): StreamCode {
 	if (err instanceof TimeoutError) return StreamCode.DeliveryTimeout;
 	// Session-scoped: the peer learns which rule it broke from the session close, not from here.
 	if (err instanceof ProtocolViolation) return StreamCode.SessionClosed;
+	// App codes share the 64+ range in both registries.
+	if (err instanceof Session) return err.code >= 64 ? StreamCode(err.code) : StreamCode.Unroutable;
+	// A session that failed underneath without a close code, as on a network or idle failure.
+	const source = typeof err === "object" && err !== null ? (err as { source?: unknown }).source : undefined;
+	if (source === "session") return StreamCode.Unroutable;
 	return StreamCode.Internal;
 }
 

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

@@ -4,7 +4,7 @@ import { Signal } from "@moq/signals";
 import type { Auth, Grant } from "../auth.ts";
 import { Producer as BroadcastProducer } from "../broadcast.ts";
 import * as Epoch from "../epoch.ts";
-import { error, ProtocolViolation, StreamCode } from "../error.ts";
+import { error, ProtocolViolation, SessionCode, SessionError, StreamCode, unauthorized } from "../error.ts";
 import { Producer as GroupProducer, MAX_GROUP_FRAMES } from "../group.ts";
 import { type Hop, HopSchema } from "../hop.ts";
 import { createMockTransportPair } from "../mock.ts";
@@ -1801,7 +1801,13 @@ test("subscription completion sends PUBLISH_DONE on every supported draft", asyn
 	] as const;
 
 	for (const version of versions) {
-		for (const abort of [undefined, new Error("failed")]) {
+		// A revocation bridged from upstream keeps its meaning: UNAUTHORIZED, not INTERNAL_ERROR.
+		for (const [abort, status] of [
+			[undefined, 0x2],
+			[new Error("failed"), 0x0],
+			[unauthorized("test"), 0x1],
+			[new SessionError(SessionCode.Expired), 0x1],
+		] as const) {
 			const pair = createMockTransportPair(ALPN.DRAFT_19);
 			const session = new NativeSession(pair.server, version, true);
 			const path = Path.from("test");
@@ -1826,7 +1832,7 @@ test("subscription completion sends PUBLISH_DONE on every supported draft", asyn
 			expect(await client.reader.u53()).toBe(PublishDone.id);
 			const done = await PublishDone.decode(client.reader, version);
 			expect(done.requestId).toBe(version <= Version.DRAFT_16 ? requestId : undefined);
-			expect(done.statusCode).toBe(abort ? 0x0 : 0x2);
+			expect(done.statusCode).toBe(status);
 
 			await running;
 			client.close();

js/net/src/ietf/publisher.ts

@@ -13,6 +13,8 @@ import {
 	NotFound,
 	ProtocolViolation,
 	reason,
+	SessionCode,
+	SessionError,
 	StreamCode,
 	StreamError,
 	unauthorized,
@@ -787,6 +789,17 @@ export class Publisher {
 				console.warn(`publish error: broadcast=${name} track=${track.name} error=${reason(publishError)}`);
 			}
 
+			// A revocation bridged from upstream, such as a moq-lite UNAUTHORIZED reset, is one here too.
+			// PUBLISH_DONE's EXPIRED means the subscription's own lifetime ran out, so an expired
+			// token is reported as no longer granted.
+			if (
+				(publishError instanceof StreamError && publishError.code === StreamCode.Unauthorized) ||
+				(publishError instanceof SessionError &&
+					(publishError.code === SessionCode.Unauthorized || publishError.code === SessionCode.Expired))
+			) {
+				unauthorized = true;
+			}
+
 			// PUBLISH_DONE is required for every supported draft. The peer may have already
 			// closed its side to unsubscribe, in which case there is nowhere to send it.
 			// Draft 14 registers no UPDATE_FAILED, so a failed update ends there as any error does.

js/net/src/ietf/subscriber.ts

@@ -43,7 +43,7 @@ import {
 	ObjectIdGap,
 } from "./object.ts";
 import { fromWire, toWire } from "./priority.ts";
-import { type Publish, PublishDone, PublishError, publishDoneClean } from "./publish.ts";
+import { type Publish, PublishDone, PublishDoneStatus, PublishError, publishDoneClean } from "./publish.ts";
 import {
 	type PublishNamespace,
 	PublishNamespaceCancel,
@@ -823,7 +823,7 @@ export class Subscriber {
 			// unsubscribe, a revoked grant); race them once so the demand loop doesn't
 			// re-subscribe each pass.
 			const done = race([
-				this.#runPublishDone(stream, subscription).then(() => publisherEnded),
+				this.#runPublishDone(stream, subscription, broadcast).then(() => publisherEnded),
 				producer.closed.then(() => localEnded),
 				revoked,
 			]);
@@ -881,14 +881,16 @@ export class Subscriber {
 	 * is only a hint: a peer may send 0 regardless, so 0 waits out the grace. A request
 	 * stream that FINs without PUBLISH_DONE is a protocol violation.
 	 */
-	async #runPublishDone(stream: Stream, subscription: Subscription): Promise<void> {
+	async #runPublishDone(stream: Stream, subscription: Subscription, broadcast: Path.Valid): Promise<void> {
 		const version = this.#session.version;
 		if (await stream.reader.done()) throw new ProtocolViolation("subscribe stream ended without PUBLISH_DONE");
 		const typeId = await stream.reader.u53();
 		if (typeId !== PublishDone.id) {
 			throw new ProtocolViolation(`unexpected message on a subscription: 0x${typeId.toString(16)}`);
 		}
 		const done = await PublishDone.decode(stream.reader, version);
+		// A revocation reads the same as one from a moq-lite peer.
+		if (done.statusCode === PublishDoneStatus.UNAUTHORIZED) throw unauthorized(broadcast);
 		if (!publishDoneClean(done.statusCode, version)) {
 			throw new Error(`publish done: status=0x${done.statusCode.toString(16)} reason=${done.reasonPhrase}`);
 		}

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

@@ -1,6 +1,6 @@
 import { expect, test } from "bun:test";
 import { Milli } from "@moq/clock";
-import { ProtocolViolation } from "../error.ts";
+import { ProtocolViolation, StreamCode, StreamError } from "../error.ts";
 import type { Consumer as GroupConsumer } from "../group.ts";
 import { createMockTransportPair } from "../mock.ts";
 import * as Path from "../path.ts";
@@ -17,6 +17,7 @@ const VERSION = Version.DRAFT_19;
 const ALIAS = 9n;
 const TRACK_ENDED = 0x2;
 const INTERNAL_ERROR = 0x0;
+const UNAUTHORIZED = 0x1;
 
 // A plain subgroup stream: no extensions, no subgroup id, end of group on FIN.
 const FLAGS: GroupFlags = {
@@ -147,6 +148,14 @@ test("a PUBLISH_DONE with an error status aborts the track", async () => {
 	expect(await reader.closed).toBeInstanceOf(Error);
 });
 
+test("an UNAUTHORIZED PUBLISH_DONE reads as the revocation a moq-lite peer sends", async () => {
+	const { reader, done } = await subscribed();
+	await done(UNAUTHORIZED, 0n);
+	const closed = await reader.closed;
+	expect(closed).toBeInstanceOf(StreamError);
+	expect((closed as StreamError).code).toBe(StreamCode.Unauthorized);
+});
+
 test("END_OF_TRACK after a group's last object ends the track after that group", async () => {
 	const { subscriber, reader, done } = await subscribed();
 	const group = groupStream(subscriber, 4);

js/net/src/stream.test.ts

@@ -518,7 +518,9 @@ test("Writer reset forwards only a stream error code", async () => {
 		() => undefined,
 		(err: unknown) => err,
 	);
-	expect(sessionResult).toBe(session);
+	// The session code means nothing on a stream; the stream only lost its route.
+	expect(sessionResult).toBeInstanceOf(StreamError);
+	expect((sessionResult as StreamError).code).toBe(StreamCode.Unroutable);
 
 	const stream = new StreamError(StreamCode.DeliveryTimeout);
 	const streamWriter = new Writer(new WritableStream<Uint8Array>(), QUIC);

kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Errors.kt

@@ -13,8 +13,8 @@ val MoqException.isShutdown: Boolean
     get() = this is MoqException.Cancelled || this is MoqException.Closed
 
 /**
- * True for HTTP 401/403, an expired token, and a protocol Unauthorized or Expired session close. Unlike a
- * transport failure, retrying without new credentials won't help, so callers
+ * True for HTTP 401/403, an expired token, a protocol Unauthorized or Expired session close,
+ * and a track revocation. Unlike a transport failure, retrying without new credentials won't help, so callers
  * should surface these rather than reconnect.
  */
 val MoqException.isAuth: Boolean

py/moq-rs/README.md

@@ -264,7 +264,7 @@ Every handle whose cleanup is `cancel()` is an async context manager, so exiting
 - **`log_level(level="info")`**. Initialize logging for the underlying Rust layer (`"error"`, `"warn"`, `"info"`, `"debug"`, `"trace"`). Call once per process.
 - **`Error`**. The exception raised by all operations. Catch a specific case via its variants, e.g. `except moq.Error.AlreadyResponded:`, `except moq.Error.Cancelled:`, or `except moq.Error.Busy:` when a setter races an in-flight connect/listen/accept.
 - **`is_shutdown(err)`**. True for `Cancelled` and `Closed`, which arise from graceful shutdown rather than an actual failure. Use it to break out of an `async for` without treating the expected end-of-stream error as a problem.
-- **`is_auth(err)`**. True for `Unauthorized` (HTTP 401), `Forbidden` (HTTP 403), and `Expired` (a request token expired), and for a protocol Unauthorized or Expired session close. Retrying without new credentials won't help, so surface these rather than reconnect.
+- **`is_auth(err)`**. True for `Unauthorized` (HTTP 401), `Forbidden` (HTTP 403), and `Expired` (a request token expired), and for a protocol Unauthorized or Expired session close or track revocation. Retrying without new credentials won't help, so surface these rather than reconnect.
 - **`protocol_error(err)`**. The structured protocol failure (session or stream scope, verbatim wire code, known kind) when the peer sent one.
 
 ## See Also

py/moq-rs/moq/errors.py

@@ -38,7 +38,8 @@ def is_shutdown(err: BaseException) -> bool:
 
 
 def is_auth(err: BaseException) -> bool:
-    """True for HTTP 401/403, an expired token, and a protocol Unauthorized or Expired session close.
+    """True for HTTP 401/403, an expired token, a protocol Unauthorized or Expired
+    session close, and a track revocation.
 
     Unlike a transport failure, retrying without new credentials won't help, so
     callers should surface these rather than reconnect.

py/moq-rs/tests/test_local.py

@@ -125,6 +125,7 @@ def test_protocol_error_helper_covers_known_app_and_unknown():
     cases = [
         (moq.ErrorScope.SESSION, 0x2, moq.ProtocolKind.UNAUTHORIZED, True),
         (moq.ErrorScope.SESSION, 0x18, moq.ProtocolKind.EXPIRED, True),
+        (moq.ErrorScope.STREAM, 0x3B, moq.ProtocolKind.UNAUTHORIZED, True),
         (moq.ErrorScope.SESSION, 0x7, moq.ProtocolKind.TOO_MANY_REQUESTS, False),
         (moq.ErrorScope.SESSION, 64 + 404, moq.ProtocolKind.APP, False),
         (moq.ErrorScope.SESSION, 0x1F, moq.ProtocolKind.UNKNOWN, False),

quest/m1/README.md

@@ -88,7 +88,6 @@ lands before the cut.
 - [Announcement takeover](/quest/m1/announce-takeover.md) - a new independent route explicitly continues instances served through its predecessor, with shared Rust/JS semantics and lite-07 wire
 - [IETF takeover](/quest/m1/ietf-takeover.md) - carry the shared takeover contract through the negotiated IETF epoch extension
 - [Identity and recovery guide](/quest/m1/identity-guide.md) - explain pinned catalogs, lazy jobs, takeover, restart and version compatibility
-- [Dropped sources](/quest/m1/dropped-sources.md) - track consumers see the producer's real error on every end path, never `Dropped`; a dead route ends Unroutable, and the TS follower lingers on announcements instead of matching errors
 - [A partial group is never cached as whole](/quest/m1/partial-group-reset.md) - every reader, local or served, gets `Old` when an unfinished subscription group expires, so no hop caches the prefix as complete
 - [Generated C bindings](/quest/m1/c/README.md) - C generated from moq-ffi ships as `moq-c` 0.8.0 and replaces the hand-written libmoq
 - [No rustls panic on an http dial](/quest/m1/http-dial-crypto-provider.md) - the `http://` certificate fetch installs a crypto provider first

quest/m1/decode-errors.md

@@ -32,4 +32,3 @@ truncate. Wire: none.
 ## Related
 
 - [Container reads drop backfill](/quest/m1/read-backfill.md) - the other Rust/JS container-consumer parity gap
-- [Dropped sources](/quest/m1/dropped-sources.md) - consumers see the producer's real error on every end