From 65b055b2660f5188774995c851f94e7c57910c43 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 16 Oct 2026 08:21:13 -0700 Subject: [PATCH 01/11] quest: claim quest/m1/dropped-sources Co-Authored-By: Claude Opus 5.5 From c5fef4690511b25af1527540fb597706c66a39c4 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 16 Oct 2026 09:14:49 -0700 Subject: [PATCH 02/11] fix(net): consumers see the source's real error, never Dropped A front no longer invents Dropped: a lost route ends Unroutable, a retired or closed origin ends Cancel or Closed. The wire conversions lose their catch-alls; an upstream session or transport failure under a stream goes out as UNROUTABLE, and Dropped stays Internal to mark a real bug. ts::Follower stops matching errors: announcements decide whether a lingering broadcast returned, and the linger expiry reports the last resolve error, so `moq export ts --stitch --linger` no longer exits on a catalog request routed through a dying session. IETF PUBLISH_DONE carries a bridged revocation as UNAUTHORIZED in Rust and JS, and the JS subscriber reads it as the shared Unauthorized stream error. The bindings document and test is_auth for a stream-scoped revocation. Closes #5347 Co-Authored-By: Claude Opus 5.5 --- doc/bin/cli.md | 4 +- drafts/draft-lcurley-moq-lite.md | 4 +- go/wrapper/errors.go | 3 +- go/wrapper/errors_test.go | 9 + js/net/src/error.test.ts | 7 +- js/net/src/error.ts | 6 +- js/net/src/ietf/publisher.test.ts | 11 +- js/net/src/ietf/publisher.ts | 10 + js/net/src/ietf/subscriber.ts | 8 +- js/net/src/ietf/tail.test.ts | 11 +- js/net/src/stream.test.ts | 4 +- .../kotlin/dev/moq/Errors.kt | 4 +- py/moq-rs/README.md | 2 +- py/moq-rs/moq/errors.py | 2 +- py/moq-rs/tests/test_local.py | 1 + quest/m1/README.md | 1 - quest/m1/dropped-sources.md | 88 -------- rs/moq-mux/src/container/ts/export_test.rs | 86 ++++++++ rs/moq-mux/src/container/ts/follower.rs | 76 ++++--- rs/moq-net/src/error.rs | 55 ++++- rs/moq-net/src/ietf/publisher.rs | 19 +- rs/moq-net/src/model/front.rs | 38 ++-- rs/moq-net/src/model/resume.rs | 9 +- rs/moq-net/tests/integration/auth.rs | 7 +- rs/moq-net/tests/integration/main.rs | 1 + rs/moq-net/tests/integration/source_errors.rs | 198 ++++++++++++++++++ swift/Sources/Moq/Errors.swift | 4 +- 27 files changed, 491 insertions(+), 177 deletions(-) delete mode 100644 quest/m1/dropped-sources.md create mode 100644 rs/moq-net/tests/integration/source_errors.rs diff --git a/doc/bin/cli.md b/doc/bin/cli.md index 6a54f98ae6..805b8f750a 100644 --- a/doc/bin/cli.md +++ b/doc/bin/cli.md @@ -395,7 +395,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 diff --git a/drafts/draft-lcurley-moq-lite.md b/drafts/draft-lcurley-moq-lite.md index a51b44f89f..079cd8fc6f 100644 --- a/drafts/draft-lcurley-moq-lite.md +++ b/drafts/draft-lcurley-moq-lite.md @@ -349,7 +349,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. | | ------- | ------------- | ----------- | @@ -364,6 +364,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. @@ -1597,6 +1598,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 a stream served through an upstream session that ended with UNROUTABLE, rather than INTERNAL_ERROR. - 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. diff --git a/go/wrapper/errors.go b/go/wrapper/errors.go index 20babf9999..ac601c285c 100644 --- a/go/wrapper/errors.go +++ b/go/wrapper/errors.go @@ -99,7 +99,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 diff --git a/go/wrapper/errors_test.go b/go/wrapper/errors_test.go index cecfc0f95c..9e5100f853 100644 --- a/go/wrapper/errors_test.go +++ b/go/wrapper/errors_test.go @@ -130,3 +130,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) + } +} diff --git a/js/net/src/error.test.ts b/js/net/src/error.test.ts index d1a609bcfd..6c634d17fb 100644 --- a/js/net/src/error.test.ts +++ b/js/net/src/error.test.ts @@ -334,10 +334,11 @@ 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); }); // A control request that outlived its deadline is not late content. It carries its own code so a diff --git a/js/net/src/error.ts b/js/net/src/error.ts index d0b1dc5acd..8448a4326e 100644 --- a/js/net/src/error.ts +++ b/js/net/src/error.ts @@ -354,8 +354,9 @@ 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. * * 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 +376,7 @@ 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; + if (err instanceof Session) return StreamCode.Unroutable; return StreamCode.Internal; } diff --git a/js/net/src/ietf/publisher.test.ts b/js/net/src/ietf/publisher.test.ts index 24739e0047..ae773c0c6f 100644 --- a/js/net/src/ietf/publisher.test.ts +++ b/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, 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"; @@ -1918,7 +1918,12 @@ 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], + ] as const) { const pair = createMockTransportPair(ALPN.DRAFT_19); const session = new NativeSession(pair.server, version, true); const path = Path.from("test"); @@ -1943,7 +1948,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(); diff --git a/js/net/src/ietf/publisher.ts b/js/net/src/ietf/publisher.ts index 7e4075d032..9f246e5aa9 100644 --- a/js/net/src/ietf/publisher.ts +++ b/js/net/src/ietf/publisher.ts @@ -11,6 +11,8 @@ import { NotFound, ProtocolViolation, reason, + SessionCode, + SessionError, StreamCode, StreamError, unauthorized, @@ -776,6 +778,14 @@ 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. + if ( + (publishError instanceof StreamError && publishError.code === StreamCode.Unauthorized) || + (publishError instanceof SessionError && publishError.code === SessionCode.Unauthorized) + ) { + 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. try { diff --git a/js/net/src/ietf/subscriber.ts b/js/net/src/ietf/subscriber.ts index ca4827ed8b..981b08bda9 100644 --- a/js/net/src/ietf/subscriber.ts +++ b/js/net/src/ietf/subscriber.ts @@ -44,7 +44,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, @@ -788,7 +788,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, ]); @@ -846,7 +846,7 @@ 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 { + async #runPublishDone(stream: Stream, subscription: Subscription, broadcast: Path.Valid): Promise { 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(); @@ -854,6 +854,8 @@ export class Subscriber { 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}`); } diff --git a/js/net/src/ietf/tail.test.ts b/js/net/src/ietf/tail.test.ts index 5709eafbdc..8a3426258c 100644 --- a/js/net/src/ietf/tail.test.ts +++ b/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); diff --git a/js/net/src/stream.test.ts b/js/net/src/stream.test.ts index b8c95bc29a..d166ac7e6c 100644 --- a/js/net/src/stream.test.ts +++ b/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(), QUIC); diff --git a/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Errors.kt b/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Errors.kt index 12fbb406ce..b7a1adcb0b 100644 --- a/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Errors.kt +++ b/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 diff --git a/py/moq-rs/README.md b/py/moq-rs/README.md index 9b3b9ab764..b3287128a6 100644 --- a/py/moq-rs/README.md +++ b/py/moq-rs/README.md @@ -261,7 +261,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 diff --git a/py/moq-rs/moq/errors.py b/py/moq-rs/moq/errors.py index 4e1aae6bc8..4988098efb 100644 --- a/py/moq-rs/moq/errors.py +++ b/py/moq-rs/moq/errors.py @@ -38,7 +38,7 @@ 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. diff --git a/py/moq-rs/tests/test_local.py b/py/moq-rs/tests/test_local.py index 598d7f463c..a0123d9481 100644 --- a/py/moq-rs/tests/test_local.py +++ b/py/moq-rs/tests/test_local.py @@ -122,6 +122,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), diff --git a/quest/m1/README.md b/quest/m1/README.md index c3b79d29f5..9e2d8da617 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -85,7 +85,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 - [Client settings parity](/quest/m1/obs-client-config.md) - moq-ffi offers moq-c's client knobs, and the OBS Advanced settings get back the ones the C++ migration dropped - [Session report parity](/quest/m1/obs-session-report.md) - a session reports its negotiated draft and reconnect failures, so the OBS dock shows them again - [Generated C++ shape](/quest/m1/cpp-generated-shape.md) - `moq::Client` is the generated type itself and `moq::expected` is one type at every C++ standard diff --git a/quest/m1/dropped-sources.md b/quest/m1/dropped-sources.md deleted file mode 100644 index 82c189daed..0000000000 --- a/quest/m1/dropped-sources.md +++ /dev/null @@ -1,88 +0,0 @@ -# [L] Consumers see the producer's real error, never Dropped - -## Goal - -A track that ends because its source ended reports the source's own error to -every consumer, locally and across a relay. `Dropped` means only that a handle -was dropped without an end, which a correct producer never does. A broadcast -end carries no cause since #4031, so only track errors are in scope. -Consumers never decide behavior by matching an error; errors are diagnostics. - -## Plan - -- Rust already maps IETF `PUBLISH_DONE` Unauthorized to `Error::Unauthorized`, - and a closed source's standing route no longer re-requests it. -- #4179 fixed revoked upstream subscriptions and #4120 preserves session death - and resumed-track errors. Both are on `main`; verify the remaining - track paths (source close, route removal, broadcast withdrawal) locally and - over a mock session, and map JS `PUBLISH_DONE` Unauthorized to #4179's shared - error. Preserve causes at the source rather than remapping `Dropped` at - consumers. -- The same goes for revocation. Leftovers from #4179's review: a bridged - revocation reaches Rust IETF subscribers as `PUBLISH_DONE` InternalError, a - JS relay reports a route revocation as INTERNAL_ERROR, and the bindings - neither document nor test `is_auth` for a stream-scoped Unauthorized. Each - should surface Unauthorized. - -- Decided 2026-10-11 from [#4349](https://github.com/moq-dev/moq/issues/4349), - where a clean publisher close ended an in-flight subscription with - `internal error` and subscribes during cross-node churn answered the same. - `Dropped` has no arm in `From<&Error> for StreamError`, so it falls through - the `_ => Internal` catch-all to 0x0. - - Fronts end with `Error::Dropped` today (`model/front.rs`: `Retired`, - `Closed`, a resolved front whose route leaves the table, `source_closed`). - Requests and tracks a front actually aborts end with `Unroutable` instead, - or `Cancel` on a local origin `Retired` or `Closed`. - - Tracks in flight are not aborted: `Action::End` in `model/origin.rs` - leaves them on their copy, which drains (ANNOUNCE_END does not disturb - in-flight subscriptions) and ends with the source's own error. Trace that - copy's end for the reported case rather than remapping the front. - - Remove the `_ => Internal` catch-alls in both `From<&Error>` conversions - (`StreamError` and `SessionError`); `Dropped` stays `Internal` - deliberately, since it then marks a real bug. - - Regression: a relay test closes a broadcast under an active subscription - whose source copy ends with a specific non-clean error, and expects that - exact error, never `internal error`; a subscribe against a withdrawn - route expects `Unroutable`. - -- Decided 2026-10-12 from [#5347](https://github.com/moq-dev/moq/issues/5347): - `moq export ts --linger --stitch` exited on `internal error` when its - catalog TRACK request landed on a publisher session that was dead but not - yet timed out. On epochless lite-06 the front is offered only its first - route, so when that session timed out, `Front::selected(None)` ended the - resolved front with `Dropped`, `TrackIo::end` rejected the request with it, - and it went out as Internal. A `Dropped` that reaches the wire means some - path never ended with a proper error; diagnose each one at its root rather - than remapping it. - - A route that fails under a request or track (the upstream session dies or - times out, and nothing replaces the copy) ends downstream with - `Unroutable`, for pending requests and for tracks whose copy ends with the - upstream's `Transport` or `Session` error. The downstream session is - healthy; only the path lost its route. This matches - `resume::route_failed`, which already counts those as a route failure. - - Give every variant the catch-alls swallow a real arm, `Transport` - included (also `TooManyRequests`, `Unsupported`, `Duplicate`, - `InvalidRange`, `FrameOpen`, `Encode`, `InvalidJoiningRequestId`). - - Remote errors stay nested (`Error::Stream(..)`), as `Error::stream()` - promises; nothing depends on flat values once the follower stops - matching. Flattening was considered and dropped on #5358's review. - - `ts::Follower` stops matching errors. While lingering, announcements - decide whether the broadcast returned: any resolve error returns it to - waiting for the next announcement, and a new announcement or end cancels - an in-flight resolve instead of sitting out a dead session's idle - timeout. `--linger` is the only exit; at expiry it fails with the last - resolve error if there was one, otherwise with the original end. - - Regressions: a relay test where a TRACK request is in flight to an - upstream session that then times out, and the downstream reset is - Unroutable; a follower test where a resolve error while lingering keeps - lingering until the instance returns, and a refusal surfaces at expiry. - -Public API: none expected; error values consumers observe change. Wire: none. - -## Closes - -- [#5347](https://github.com/moq-dev/moq/issues/5347) - `export ts --stitch --linger` exits on `internal error` when a catalog request lands on a dying publisher - -## Related - -- [#4179](https://github.com/moq-dev/moq/pull/4179) - owns the revoked-upstream path diff --git a/rs/moq-mux/src/container/ts/export_test.rs b/rs/moq-mux/src/container/ts/export_test.rs index 3f7b293e00..0e53aa5001 100644 --- a/rs/moq-mux/src/container/ts/export_test.rs +++ b/rs/moq-mux/src/container/ts/export_test.rs @@ -8869,6 +8869,92 @@ async fn a_follower_keeps_lingering_when_a_return_goes_before_it_resolves() { assert_eq!(start.elapsed(), linger); } +/// Announce `epoch` at `live`, refusing every track request with `NotFound` until the +/// returned task is aborted, which ends the broadcast. +fn publish_refusing(origin: &moq_net::origin::Producer, epoch: &moq_net::Epoch) -> tokio::task::JoinHandle<()> { + let broadcast = origin + .publish("live", moq_net::origin::Route::default().with_epoch(epoch.clone())) + .unwrap(); + let mut dynamic = broadcast.dynamic(); + tokio::spawn(async move { + let _broadcast = broadcast; + while let Ok(request) = dynamic.requested_track().await { + request.reject(moq_net::Error::NotFound); + } + }) +} + +/// A replacement that refuses its catalog while still announced is no return: the follower +/// keeps lingering, and follows the next instance to serve. +#[tokio::test(start_paused = true)] +async fn a_follower_keeps_lingering_through_a_refused_return() { + let origin = crate::source::produce_origin(); + let first = publish_bare(&origin, Some(&moq_net::Epoch::mint())); + let linger = Duration::from_secs(10); + let mut follower = follower(&origin).await.with_linger(linger).with_stitch(true); + let start = tokio::time::Instant::now(); + finish(first); + assert!( + tokio::time::timeout(Duration::from_secs(1), follower.next()) + .await + .is_err(), + "lingering" + ); + + // Replaced, but the replacement refuses its catalog request. + let refusing = publish_refusing(&origin, &moq_net::Epoch::mint()); + assert!( + tokio::time::timeout(Duration::from_secs(2), follower.next()) + .await + .is_err(), + "the refusal keeps it lingering" + ); + refusing.abort(); + + let mut second = publish_bare(&origin, Some(&moq_net::Epoch::mint())); + std::ops::DerefMut::deref_mut(&mut second.1.modify().unwrap()); + assert!( + tokio::time::timeout(Duration::from_secs(1), follower.next()) + .await + .is_err(), + "following the next replacement" + ); + assert!(follower.export().cataloged(), "the replacement served its catalog"); + finish(second); + let end = follower.next().await; + assert!(matches!(end, Ok(None)), "the replacement's clean end: {end:?}"); + assert_eq!(start.elapsed(), Duration::from_secs(4) + linger); +} + +/// A return that stays announced but refuses its catalog fails the follower only once the +/// linger runs out, with that refusal rather than the original clean end. +#[tokio::test(start_paused = true)] +async fn a_follower_reports_a_refused_return_at_the_linger() { + let origin = crate::source::produce_origin(); + let epoch = moq_net::Epoch::mint(); + let first = publish_bare(&origin, Some(&epoch)); + let linger = Duration::from_secs(10); + let mut follower = follower(&origin).await.with_linger(linger); + let start = tokio::time::Instant::now(); + finish(first); + + assert!( + tokio::time::timeout(Duration::from_secs(1), follower.next()) + .await + .is_err(), + "lingering" + ); + let _refusing = publish_refusing(&origin, &epoch); + let end = tokio::time::timeout(linger * 3, follower.next()) + .await + .expect("the linger bounds the wait"); + assert!( + matches!(end, Err(crate::Error::Moq(moq_net::Error::NotFound))), + "the refusal surfaces at the linger: {end:?}" + ); + assert_eq!(start.elapsed(), linger); +} + /// A stitch whose replacement goes before it resolves is no switch yet: the export stays on /// its own instance and follows the next replacement instead. #[tokio::test(start_paused = true)] diff --git a/rs/moq-mux/src/container/ts/follower.rs b/rs/moq-mux/src/container/ts/follower.rs index 8bee334c7f..6147495f34 100644 --- a/rs/moq-mux/src/container/ts/follower.rs +++ b/rs/moq-mux/src/container/ts/follower.rs @@ -83,6 +83,9 @@ enum State { /// replaced publisher that stays up keeps the export until it ends. An export that fails /// while its broadcast stays announced fails here too, without lingering. /// +/// Announcements alone decide whether the broadcast returned: a return that fails to resolve +/// waits for the next one, and the linger running out reports that failure. +/// /// An epochless route has no instance to match, so any return of one is a replacement. pub struct Follower { export: Export>, @@ -98,6 +101,10 @@ pub struct Follower { started: bool, /// The origin closed, so nothing more is announced. closed: bool, + /// Why the last return failed to resolve, reported instead of the end if the linger runs out. + failed: Option, + /// A return failed to resolve: nothing is followed until the next announcement. + refused: bool, linger: Duration, stitch: bool, timer: Option>>, @@ -122,6 +129,8 @@ impl Follower { ended: false, started: false, closed: false, + failed: None, + refused: false, linger: Duration::ZERO, stitch: false, timer: None, @@ -262,6 +271,7 @@ impl Follower { } let follow = match &self.serving { + _ if self.refused => false, Serving::Ours => self.ended, Serving::Other(_) if self.stitch => true, Serving::Other(_) => return Poll::Ready(Err(crate::Error::Replaced(self.path.to_string()))), @@ -283,8 +293,7 @@ impl Follower { continue; } if self.closed || self.elapsed(waiter, deadline) { - tracing::info!(path = %self.path, linger = ?self.linger, "broadcast did not return"); - return Poll::Ready(end.map(|()| None)); + return Poll::Ready(self.gone(end)); } self.state = State::Settling { end, grace, deadline }; return Poll::Pending; @@ -294,26 +303,31 @@ impl Follower { awaiting, fallback, } => { - // Announcements wait until the export is on the instance they describe. + // Whatever the request waits on, the announcements say what serves the path + // now: rather than sit out a route that went, ask afresh from there. A stitch + // stays on the export it has, still awaiting what it was. + if self.poll_announced(waiter) { + tracing::debug!(path = %self.path, "announcement while resolving, asking again"); + self.state = *fallback; + // The fallback's wait predates this poll, so it gets no same-poll grace. + if let Some(expired) = self.expired() { + return Poll::Ready(expired); + } + continue; + } if let Poll::Ready(resolved) = waiter.poll_future(resolving.as_mut()) { let (broadcast, catalog) = match resolved { Ok(resolved) => resolved, - // An instance that went before it resolved is no return or switch yet. - // Until the next announcement says otherwise, nothing serves the path: - // a stitch stays on the export it has, still awaiting what it was, and a - // return keeps lingering. A refusal from an instance still up fails loud - // instead. + // A stitch from a live export fails loud: the replacement would not serve. + Err(err) if matches!(*fallback, State::Running { .. }) => return Poll::Ready(Err(err)), + // While lingering, a return that fails to resolve is no return yet: nothing + // serves the path until the next announcement says otherwise. The error is + // reported only if the linger runs out. Err(err) => { - if !matches!( - err, - crate::Error::Moq(moq_net::Error::Unroutable | moq_net::Error::Dropped) - ) { - return Poll::Ready(Err(err)); - } - tracing::warn!(path = %self.path, %err, "broadcast went before it resolved, still waiting"); - self.serving = Serving::Gone; + tracing::warn!(path = %self.path, %err, "broadcast failed to resolve, still waiting"); + self.failed = Some(err); + self.refused = true; self.state = *fallback; - // The fallback's wait predates this poll, so it gets no same-poll grace. if let Some(expired) = self.expired() { return Poll::Ready(expired); } @@ -321,6 +335,7 @@ impl Follower { } }; self.export.followed(&broadcast, catalog)?; + self.failed = None; self.epoch = self.export.instance().cloned(); self.serving = Serving::Ours; self.ended = false; @@ -348,15 +363,19 @@ impl Follower { } } - /// Apply every announcement on hand. - fn poll_announced(&mut self, waiter: &kio::Waiter) { + /// Apply every announcement on hand, returning whether there was any. + fn poll_announced(&mut self, waiter: &kio::Waiter) -> bool { + let mut any = false; while !self.closed { match self.announced.poll_next(waiter) { Poll::Ready(Some(event)) => self.apply(event), Poll::Ready(None) => self.closed = true, - Poll::Pending => return, + Poll::Pending => break, } + any = true; } + self.refused &= !any; + any } fn apply(&mut self, event: moq_net::announce::Event) { @@ -443,16 +462,23 @@ impl Follower { } /// How a wait for the catalog's first snapshot ends once the linger runs out. - fn expire(&self, awaiting: Awaiting) -> crate::Result> { + fn expire(&mut self, awaiting: Awaiting) -> crate::Result> { match awaiting.missing { - Missing::Return(end) => { - tracing::info!(path = %self.path, linger = ?self.linger, "broadcast did not return"); - end.map(|()| None) - } + Missing::Return(end) => self.gone(end), Missing::Switch => Err(self.missing_catalog()), } } + /// How the export ends once the linger runs out with nothing back: with why the last + /// return failed to resolve, if one did, else as the broadcast ended. + fn gone(&mut self, end: crate::Result<()>) -> crate::Result> { + tracing::info!(path = %self.path, linger = ?self.linger, "broadcast did not return"); + match self.failed.take() { + Some(err) => Err(err), + None => end.map(|()| None), + } + } + fn missing_catalog(&self) -> crate::Error { anyhow::anyhow!( "replacement broadcast `{}` did not serve its catalog within {:?}", diff --git a/rs/moq-net/src/error.rs b/rs/moq-net/src/error.rs index eb86c552f4..033eb92c14 100644 --- a/rs/moq-net/src/error.rs +++ b/rs/moq-net/src/error.rs @@ -645,7 +645,24 @@ impl From<&Error> for SessionError { // (a session 0x4 would become the stream's GOING_AWAY), and the draft already // says an unrecognized code is an unspecified error. Send that instead. Error::Remote(_) => Self::Internal, - _ => Self::Internal, + // Our own transport failed: there is no code to say so. + Error::Transport(_) => Self::Internal, + // Stream-scoped conditions have no session code. + Error::Old + | Error::Evicted + | Error::Lagged + | Error::NotFound + | Error::NotFetchable + | Error::InvalidRange + | Error::InvalidJoiningRequestId + | Error::Unroutable + | Error::FrameTooLarge + | Error::GroupTooLarge + | Error::FrameOpen + | Error::TimestampMismatch + | Error::MalformedTrack => Self::Internal, + // A producer dropped without an end is a bug here, so it is never given a meaning. + Error::Dropped => Self::Internal, } } } @@ -660,8 +677,10 @@ impl From<&Error> for StreamError { Error::Stream(err) => err.clone(), // App codes share the 64+ range in both registries. Error::Session(SessionError::App(app)) => Self::App(*app), - // A session-scoped code has no meaning in this registry; don't forward the number. - Error::Session(_) => Self::Internal, + // A stream that can still be reset outlived the session that failed, so that was + // another one: the route this stream was served through. The path lost its route; + // this session is fine, and the failed session's code means nothing here. + Error::Session(_) | Error::Transport(_) => Self::Unroutable, Error::Cancel | Error::Closed => Self::Cancel, Error::SessionClosed => Self::Session(SessionError::Cancel), Error::Decode(coding::DecodeError::Session(err)) => Self::Session(err.clone()), @@ -695,10 +714,20 @@ impl From<&Error> for StreamError { Error::Version | Error::UnknownAlpn(_) | Error::TooManyParameters + | Error::TooManyRequests + | Error::Duplicate | Error::GoawayTimeout | Error::ProtocolViolation | Error::UnexpectedMessage => Self::Session(SessionError::from(err)), - _ => Self::Internal, + // No stream code says so: an unknown stream type, a request only moq-transport + // can make, or a fault on our own side. + Error::Unsupported + | Error::InvalidRange + | Error::InvalidJoiningRequestId + | Error::FrameOpen + | Error::Encode(_) => Self::Internal, + // A producer dropped without an end is a bug here, so it is never given a meaning. + Error::Dropped => Self::Internal, } } } @@ -889,17 +918,25 @@ mod tests { 64 + 7 ); - // An unregistered code carries no registry, so it must not be re-sent as a number - // that means something here: session 0x4 and 0x5 are GOING_AWAY and TOO_FAR_BEHIND - // on a stream. Downgrade to the unspecified-error code instead. - for code in [0x4, 0x5, 0x1f] { + // An upstream session's close must not be re-sent as a number that means something + // here: session 0x4 and 0x5 are GOING_AWAY and TOO_FAR_BEHIND on a stream. The + // downstream session is healthy; only the path lost its route. + for code in [0x0, 0x4, 0x5, 0x1f] { let crossed = StreamError::from(&Error::from(SessionError::from_code(code))); assert_eq!( crossed, - StreamError::Internal, + StreamError::Unroutable, "session {code:#x} leaked into the stream space" ); } + assert_eq!( + StreamError::from(&Error::Transport("idle timeout".into())), + StreamError::Unroutable + ); + + // A producer dropped without an end is a bug, never given a meaning. + assert_eq!(StreamError::from(&Error::Dropped), StreamError::Internal); + assert_eq!(SessionError::from(&Error::Dropped), SessionError::Internal); } // The registry a code is read with depends on what failed, and picking the wrong one diff --git a/rs/moq-net/src/ietf/publisher.rs b/rs/moq-net/src/ietf/publisher.rs index f86cfcb4e6..d21a6655ea 100644 --- a/rs/moq-net/src/ietf/publisher.rs +++ b/rs/moq-net/src/ietf/publisher.rs @@ -504,9 +504,11 @@ impl Served { // PUBLISH_DONE is the only place the subscriber learns why its renewal failed. Self::UpdateFailed(err) if version == Version::Draft14 => Self::Track(Err(err.clone())).status(version), Self::Track(Ok(())) => (ietf::PublishDoneStatus::TrackEnded, "track ended"), - Self::Track(Err(Error::Unauthorized | Error::Session(crate::SessionError::Unauthorized))) => { - (ietf::PublishDoneStatus::Unauthorized, "not granted") - } + Self::Track(Err( + Error::Unauthorized + | Error::Session(crate::SessionError::Unauthorized) + | Error::Stream(crate::StreamError::Unauthorized), + )) => (ietf::PublishDoneStatus::Unauthorized, "not granted"), // PUBLISH_DONE registers no token expiry, and its EXPIRED means the subscription's // own lifetime ran out, so an expired token is reported as no longer granted. Self::Track(Err(Error::Expired | Error::Session(crate::SessionError::Expired))) => { @@ -6442,6 +6444,17 @@ mod serve_tests { Served::UpdateFailed(Error::Unauthorized).status(Version::Draft14).0, ietf::PublishDoneStatus::Unauthorized ); + // A revocation bridged from upstream, however it was received, is still one. + for err in [ + Error::Unauthorized, + Error::Session(crate::SessionError::Unauthorized), + Error::Stream(crate::StreamError::Unauthorized), + ] { + assert_eq!( + Served::Track(Err(err)).status(Version::Draft16).0, + ietf::PublishDoneStatus::Unauthorized + ); + } } /// Draft-21 ยง9.5: a failed REQUEST_UPDATE is answered with REQUEST_ERROR, then the diff --git a/rs/moq-net/src/model/front.rs b/rs/moq-net/src/model/front.rs index 88da908756..b8118f2619 100644 --- a/rs/moq-net/src/model/front.rs +++ b/rs/moq-net/src/model/front.rs @@ -199,8 +199,6 @@ pub(super) struct Front { upstream: Option, /// Routes excluded from selection: their source ended while still advertised. excluded: HashSet, - /// Why the last candidate fell through, reported if the front ends unresolved. - last_err: Option, /// Whether the parked requesters were resolved (the first source attached). resolved: bool, tracks: BTreeMap, Track>, @@ -229,7 +227,6 @@ impl Front { serving_closing: false, upstream: None, excluded: HashSet::new(), - last_err: None, resolved: false, tracks: BTreeMap::new(), parked: BTreeSet::new(), @@ -331,8 +328,9 @@ impl Front { self.retiring = false; } Event::Unheld => self.held = false, - Event::Retired => self.end(Error::Dropped, &mut actions), - Event::Closed => self.end(Error::Dropped, &mut actions), + // Ended here, not by a source: nothing failed. + Event::Retired => self.end(Error::Cancel, &mut actions), + Event::Closed => self.end(Error::Closed, &mut actions), } if let Some(name) = touched { let after = self.parked_at(&name); @@ -387,11 +385,7 @@ impl Front { // a later announcement gets a fresh front. None => { self.upstream = None; - let err = match self.resolved { - true => Error::Dropped, - false => self.last_err.take().unwrap_or(Error::Unroutable), - }; - self.end(err, actions); + self.end(Error::Unroutable, actions); } } } @@ -414,7 +408,6 @@ impl Front { // is moot, so a front left with nothing reports the path unroutable. Err(Refusal { standing: false, .. }) => { self.upstream = None; - self.last_err = Some(Error::Unroutable); actions.push(Action::Reselect); } // The winning route's answer is final, even while another source serves: @@ -467,7 +460,6 @@ impl Front { self.serving_closing = false; actions.push(Action::Detach { source }); self.drain_copies(); - self.last_err = Some(Error::Dropped); actions.push(Action::Reselect); } @@ -937,7 +929,7 @@ mod tests { best: None, serving_closing: false, }), - &[Action::End { err: Error::Dropped }], + &[Action::End { err: Error::Unroutable }], ); } @@ -950,7 +942,7 @@ mod tests { best: None, serving_closing: false, }), - &[Action::End { err: Error::Dropped }], + &[Action::End { err: Error::Unroutable }], ); } @@ -1117,12 +1109,12 @@ mod tests { track: name("video"), source: 100, closing, - result: Err(Error::Dropped), + result: Err(Error::App(7)), delivered: true, }), &[Action::Abort { track: name("video"), - err: Error::Dropped, + err: Error::App(7), }], ); // A returning reader is not re-spliced either. @@ -1273,7 +1265,7 @@ mod tests { &[Action::Forget { track: name("video") }], ); assert_actions(front.step(Event::Forgotten { track: name("video") }), &[Action::Retire]); - assert_actions(front.step(Event::Retired), &[Action::End { err: Error::Dropped }]); + assert_actions(front.step(Event::Retired), &[Action::End { err: Error::Cancel }]); assert!(front.ended()); } @@ -1351,7 +1343,7 @@ mod tests { assert_actions(front.step(Event::Held), &[]); assert!(!front.ended()); assert_actions(front.step(Event::Unheld), &[Action::Retire]); - assert_actions(front.step(Event::Retired), &[Action::End { err: Error::Dropped }]); + assert_actions(front.step(Event::Retired), &[Action::End { err: Error::Cancel }]); } /// A finished track stays for readers still draining it and for the linger @@ -1555,7 +1547,7 @@ mod tests { #[test] fn teardown_ends_everything() { let mut front = serving(remote(1), 100); - assert_actions(front.step(Event::Closed), &[Action::End { err: Error::Dropped }]); + assert_actions(front.step(Event::Closed), &[Action::End { err: Error::Closed }]); } #[test] @@ -1841,10 +1833,10 @@ mod tests { ); } - /// A front that served ends `Dropped` when its route leaves, not with the retraction - /// of a challenger that never took over. + /// A front that served ends `Unroutable` when its route leaves, after a challenger + /// that never took over was refused. #[test] - fn a_front_that_served_ends_dropped() { + fn a_front_that_served_ends_unroutable() { let mut front = serving(remote(1), 100); front.step(Event::Selected { best: Some(remote(2)), @@ -1866,7 +1858,7 @@ mod tests { best: None, serving_closing: false, }), - &[Action::End { err: Error::Dropped }], + &[Action::End { err: Error::Unroutable }], ); } diff --git a/rs/moq-net/src/model/resume.rs b/rs/moq-net/src/model/resume.rs index 08d72cb316..8b35035df6 100644 --- a/rs/moq-net/src/model/resume.rs +++ b/rs/moq-net/src/model/resume.rs @@ -668,10 +668,11 @@ impl Reader { None | Some(Some(_)) => return Some(Poll::Ready(Ok(None))), Some(None) => {} }, - // No front is left to replace the copy: the track ends as it does. A locally - // closed copy ends it cleanly; a connection ending retains its session error. + // No front is left to replace the copy: the track ends as it does, or unroutable + // when no route served it. A locally closed copy ends it cleanly; a connection + // ending retains its session error. Some(None) => match serving.map(&done) { - None => return Some(Poll::Ready(Err(Error::Dropped))), + None => return Some(Poll::Ready(Err(Error::Unroutable))), Some(Some(Ok(()) | Err(Error::Closed))) => return Some(Poll::Ready(Ok(None))), Some(Some(Err(err))) => return Some(Poll::Ready(Err(err))), Some(None) => {} @@ -769,7 +770,7 @@ impl Reader { _ => {} } } else if concluded.is_some() { - return Poll::Ready(Err(Error::Dropped)); + return Poll::Ready(Err(Error::Unroutable)); } let _ = self.route.poll_changed(serving.unwrap_or_default(), waiter); Poll::Pending diff --git a/rs/moq-net/tests/integration/auth.rs b/rs/moq-net/tests/integration/auth.rs index 4b08a7c529..b18c09d8ec 100644 --- a/rs/moq-net/tests/integration/auth.rs +++ b/rs/moq-net/tests/integration/auth.rs @@ -1554,9 +1554,12 @@ async fn a_narrowing_resets_a_fetch_in_flight(version: &'static str) { } }; // The narrowing also retracts the route, and a fetch-only track whose route - // leaves ends `Dropped` locally, so the reader may see either. The wire carries + // leaves ends `Unroutable` locally, so the reader may see either. The wire carries // the revocation regardless. - assert!(unauthorized(&err) || matches!(err, moq_net::Error::Dropped), "{err:?}"); + assert!( + unauthorized(&err) || matches!(err, moq_net::Error::Unroutable), + "{err:?}" + ); let resets = pair.server_transport.resets(); assert!(resets.contains(&StreamError::Unauthorized.to_code()), "{resets:x?}"); drop(group); diff --git a/rs/moq-net/tests/integration/main.rs b/rs/moq-net/tests/integration/main.rs index d155bb218f..d063caf191 100644 --- a/rs/moq-net/tests/integration/main.rs +++ b/rs/moq-net/tests/integration/main.rs @@ -42,6 +42,7 @@ mod routing_contracts; mod session_close; mod session_panic; mod settled_end_survives_abort; +mod source_errors; mod subscribe_tracks; mod subscription_end_integrity; mod support; diff --git a/rs/moq-net/tests/integration/source_errors.rs b/rs/moq-net/tests/integration/source_errors.rs new file mode 100644 index 0000000000..7d55fb0481 --- /dev/null +++ b/rs/moq-net/tests/integration/source_errors.rs @@ -0,0 +1,198 @@ +#![cfg(feature = "async")] + +//! A track that ends because its source ended reports the source's own error across a +//! relay, and a path that loses its route reports `Unroutable`, never an internal error. +//! +//! Each case runs publisher -> relay -> far over two mock sessions. moq-lite only: +//! moq-transport's PUBLISH_DONE registry has no code for either error. + +use crate::support; + +use std::time::Duration; + +use moq_net::{Error, Hop, StreamError, Timestamp, Version}; +use support::harness::{MockConnectOptions, MockPair, connect_mock}; + +const VERSIONS: [&str; 3] = ["moq-lite-05", "moq-lite-06", "moq-lite-07-wip"]; + +const TIMEOUT: Duration = Duration::from_secs(10); + +fn timed() -> moq_net::track::Info { + moq_net::track::Info::default().with_timescale(moq_net::Timescale::MILLI) +} + +fn produce_origin(hop: u64) -> moq_net::origin::Producer { + let (producer, driver) = moq_net::origin::Producer::new(moq_net::origin::Config::new(Hop::new(hop).unwrap())); + support::harness::spawn(driver); + producer +} + +/// A publisher announcing `bcast`, a relay subscribed to it, and a far origin subscribed +/// to the relay, with `bcast` resolved at the relay and the far end. +struct Relayed { + _publisher: moq_net::origin::Producer, + broadcast: moq_net::broadcast::Producer, + upstream: MockPair, + _downstream: MockPair, + relayed: moq_net::broadcast::Consumer, + remote: moq_net::broadcast::Consumer, +} + +async fn resolve(origin: &moq_net::origin::Producer) -> moq_net::broadcast::Consumer { + let consumer = origin.consume(); + moq_net_sim::timeout(TIMEOUT, consumer.routed("bcast")) + .await + .expect("announce timeout") + .expect("routed"); + moq_net_sim::timeout(TIMEOUT, consumer.request_broadcast("bcast", None)) + .await + .expect("resolve timeout") + .expect("broadcast resolves") +} + +async fn relayed(version: &str) -> Relayed { + let version = version.parse::().unwrap(); + let publisher = produce_origin(1); + let relay = produce_origin(2); + let far = produce_origin(3); + let broadcast = publisher.create_broadcast("bcast").unwrap(); + broadcast.announce(Default::default()).unwrap(); + + let mut options = MockConnectOptions::new(version); + options.server_publish = Some(publisher.consume()); + options.client_subscribe = Some(relay.clone()); + let upstream = connect_mock(options).await; + let mut options = MockConnectOptions::new(version); + options.server_publish = Some(relay.consume()); + options.client_subscribe = Some(far.clone()); + let downstream = connect_mock(options).await; + + Relayed { + _publisher: publisher, + broadcast, + upstream, + _downstream: downstream, + relayed: resolve(&relay).await, + remote: resolve(&far).await, + } +} + +/// Subscribe to `name` and read until the track ends. +async fn read_to_end(broadcast: moq_net::broadcast::Consumer, name: &'static str) -> Result<(), Error> { + let mut sub = broadcast.track(name)?.subscribe(None).await?; + while let Some(mut group) = sub.recv_group().await? { + while group.read_frame().await?.is_some() {} + } + Ok(()) +} + +/// Assert `result` failed with `expected`, whether decided here or received off the wire. +#[track_caller] +fn assert_ended(result: Result<(), Error>, expected: StreamError, version: &str) { + match result { + Err(err) => assert!( + err.stream() == Some(&expected) || (err.stream().is_none() && StreamError::from(&err) == expected), + "{version}: ended with {err:?}, expected {expected:?}" + ), + Ok(()) => panic!("{version}: ended cleanly, expected {expected:?}"), + } +} + +/// #5347: a TRACK request is in flight to an upstream publisher session that dies before +/// answering. The relay's request, and the far end's through it, find the path has no +/// route; neither is an internal error. +#[moq_net_sim::test] +async fn request_in_flight_to_a_dead_upstream_is_unroutable() { + for version in VERSIONS { + let setup = relayed(version).await; + let _track = setup.broadcast.create_track("catalog", timed()).unwrap(); + // The relay's requests never reach the publisher. + setup.upstream.client_transport.hold_bidis(); + let relayed = moq_net_sim::spawn(read_to_end(setup.relayed.clone(), "catalog")); + let remote = moq_net_sim::spawn(read_to_end(setup.remote.clone(), "catalog")); + moq_net_sim::sleep(Duration::from_secs(1)).await; + assert!(!relayed.is_finished(), "{version}: answered through a held stream"); + setup.upstream.server.abort(Error::Timeout); + for reader in [relayed, remote] { + let result = moq_net_sim::timeout(TIMEOUT, reader) + .await + .expect("request hung") + .unwrap(); + assert_ended(result, StreamError::Unroutable, version); + } + } +} + +/// A subscription already delivering when the upstream session dies, with nothing to +/// replace it: the far end learns the path is gone, not that its own session closed. +#[moq_net_sim::test] +async fn track_in_flight_through_a_dead_upstream_is_unroutable() { + for version in VERSIONS { + let setup = relayed(version).await; + let track = setup.broadcast.create_track("video", timed()).unwrap(); + let mut sub = setup.remote.track("video").unwrap().subscribe(None).await.unwrap(); + let mut group = track.append_group().unwrap(); + group.write_frame(Timestamp::ZERO, b"head".as_ref()).unwrap(); + group.finish().unwrap(); + moq_net_sim::timeout(TIMEOUT, sub.recv_group()) + .await + .expect("group timeout") + .unwrap() + .unwrap(); + setup.upstream.server.abort(Error::Timeout); + let result = moq_net_sim::timeout(TIMEOUT, async { + while sub.recv_group().await?.is_some() {} + Ok(()) + }) + .await + .expect("track end hung"); + assert_ended(result, StreamError::Unroutable, version); + } +} + +/// #4349: the broadcast is withdrawn under an active subscription, then its track fails +/// with a specific error. The withdrawal does not disturb the subscription, which ends +/// with that error. +#[moq_net_sim::test] +async fn withdrawn_broadcast_keeps_the_source_error() { + for version in VERSIONS { + let setup = relayed(version).await; + let track = setup.broadcast.create_track("video", timed()).unwrap(); + let mut sub = setup.remote.track("video").unwrap().subscribe(None).await.unwrap(); + let mut group = track.append_group().unwrap(); + group.write_frame(Timestamp::ZERO, b"head".as_ref()).unwrap(); + group.finish().unwrap(); + moq_net_sim::timeout(TIMEOUT, sub.recv_group()) + .await + .expect("group timeout") + .unwrap() + .unwrap(); + setup.broadcast.unannounce(); + moq_net_sim::sleep(Duration::from_secs(1)).await; + track.abort(Error::App(7)).unwrap(); + let result = moq_net_sim::timeout(TIMEOUT, async { + while sub.recv_group().await?.is_some() {} + Ok(()) + }) + .await + .expect("track end hung"); + assert_ended(result, StreamError::App(7), version); + } +} + +/// A subscribe on a broadcast whose route was withdrawn is refused as unroutable. +#[moq_net_sim::test] +async fn subscribe_after_withdrawal_is_unroutable() { + for version in VERSIONS { + let setup = relayed(version).await; + let _track = setup.broadcast.create_track("video", timed()).unwrap(); + setup.broadcast.unannounce(); + moq_net_sim::sleep(Duration::from_secs(1)).await; + for broadcast in [setup.relayed.clone(), setup.remote.clone()] { + let result = moq_net_sim::timeout(TIMEOUT, read_to_end(broadcast, "video")) + .await + .expect("subscribe hung"); + assert_ended(result, StreamError::Unroutable, version); + } + } +} diff --git a/swift/Sources/Moq/Errors.swift b/swift/Sources/Moq/Errors.swift index 7b6037f239..7b8bd03f52 100644 --- a/swift/Sources/Moq/Errors.swift +++ b/swift/Sources/Moq/Errors.swift @@ -11,8 +11,8 @@ extension MoqError { } } - /// 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. public var isAuth: Bool { switch self { From 9f64063f0d6a5c788aa385086d70fea264a0a5da Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 16 Oct 2026 21:11:36 -0700 Subject: [PATCH 03/11] fix(js): report a bridged expired token as PUBLISH_DONE UNAUTHORIZED Matches the Rust IETF publisher, which reports an expired token as no longer granted since PUBLISH_DONE's EXPIRED means the subscription's own lifetime ran out. Co-Authored-By: Claude Opus 5.5 --- js/net/src/ietf/publisher.test.ts | 3 ++- js/net/src/ietf/publisher.ts | 5 ++++- 2 files changed, 6 insertions(+), 2 deletions(-) diff --git a/js/net/src/ietf/publisher.test.ts b/js/net/src/ietf/publisher.test.ts index ae773c0c6f..a7fbe5ee31 100644 --- a/js/net/src/ietf/publisher.test.ts +++ b/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, unauthorized } 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"; @@ -1923,6 +1923,7 @@ test("subscription completion sends PUBLISH_DONE on every supported draft", asyn [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); diff --git a/js/net/src/ietf/publisher.ts b/js/net/src/ietf/publisher.ts index 9f246e5aa9..4cfc30f72f 100644 --- a/js/net/src/ietf/publisher.ts +++ b/js/net/src/ietf/publisher.ts @@ -779,9 +779,12 @@ export class Publisher { } // 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 instanceof SessionError && + (publishError.code === SessionCode.Unauthorized || publishError.code === SessionCode.Expired)) ) { unauthorized = true; } From 5638b4f196a9c217dfefe576b6e1670441a43c0a Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 16 Oct 2026 21:28:32 -0700 Subject: [PATCH 04/11] style(py): wrap the is_auth docstring Co-Authored-By: Claude Opus 5.5 --- py/moq-rs/moq/errors.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/py/moq-rs/moq/errors.py b/py/moq-rs/moq/errors.py index 4988098efb..035d29a49c 100644 --- a/py/moq-rs/moq/errors.py +++ b/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, a protocol Unauthorized or Expired session close, and a track revocation. + """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. From 21a67d77415e58308881a008871c8e7daacd3c0b Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 17 Oct 2026 07:04:00 -0700 Subject: [PATCH 05/11] fix(js): keep a session app code when resetting a stream App codes share the 64+ range in both registries, so a relayed upstream session close with an app code reaches the stream as that code, matching the Rust conversion, rather than UNROUTABLE. Co-Authored-By: Claude Opus 5.5 --- js/net/src/error.test.ts | 2 ++ js/net/src/error.ts | 4 +++- 2 files changed, 5 insertions(+), 1 deletion(-) diff --git a/js/net/src/error.test.ts b/js/net/src/error.test.ts index 6c634d17fb..5160183efd 100644 --- a/js/net/src/error.test.ts +++ b/js/net/src/error.test.ts @@ -339,6 +339,8 @@ test("toStreamCode: a local condition maps to the code the peer can act on", () // 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)); }); // A control request that outlived its deadline is not late content. It carries its own code so a diff --git a/js/net/src/error.ts b/js/net/src/error.ts index 8448a4326e..01cd5ae1f3 100644 --- a/js/net/src/error.ts +++ b/js/net/src/error.ts @@ -357,6 +357,7 @@ interface TransportErrorOptions { * 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. * * 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 @@ -376,7 +377,8 @@ 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; - if (err instanceof Session) return StreamCode.Unroutable; + // App codes share the 64+ range in both registries. + if (err instanceof Session) return err.code >= 64 ? StreamCode(err.code) : StreamCode.Unroutable; return StreamCode.Internal; } From f722d3f428ecabe28b625bd590791790e722cbe1 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 17 Oct 2026 07:15:24 -0700 Subject: [PATCH 06/11] fix(net): give the new IdleTimeout error its wire arms A connection reaching its idle deadline is the same route loss as any other transport failure: Unroutable on a relayed stream, Internal as a session code. Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/error.rs | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/rs/moq-net/src/error.rs b/rs/moq-net/src/error.rs index 033eb92c14..e42aa4395c 100644 --- a/rs/moq-net/src/error.rs +++ b/rs/moq-net/src/error.rs @@ -646,7 +646,7 @@ impl From<&Error> for SessionError { // says an unrecognized code is an unspecified error. Send that instead. Error::Remote(_) => Self::Internal, // Our own transport failed: there is no code to say so. - Error::Transport(_) => Self::Internal, + Error::Transport(_) | Error::IdleTimeout => Self::Internal, // Stream-scoped conditions have no session code. Error::Old | Error::Evicted @@ -680,7 +680,7 @@ impl From<&Error> for StreamError { // A stream that can still be reset outlived the session that failed, so that was // another one: the route this stream was served through. The path lost its route; // this session is fine, and the failed session's code means nothing here. - Error::Session(_) | Error::Transport(_) => Self::Unroutable, + Error::Session(_) | Error::Transport(_) | Error::IdleTimeout => Self::Unroutable, Error::Cancel | Error::Closed => Self::Cancel, Error::SessionClosed => Self::Session(SessionError::Cancel), Error::Decode(coding::DecodeError::Session(err)) => Self::Session(err.clone()), @@ -930,9 +930,10 @@ mod tests { ); } assert_eq!( - StreamError::from(&Error::Transport("idle timeout".into())), + StreamError::from(&Error::Transport("reset".into())), StreamError::Unroutable ); + assert_eq!(StreamError::from(&Error::IdleTimeout), StreamError::Unroutable); // A producer dropped without an end is a bug, never given a meaning. assert_eq!(StreamError::from(&Error::Dropped), StreamError::Internal); From 8eafb5019af8e36896c164ae25135cf4fe3985b9 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 17 Oct 2026 07:19:47 -0700 Subject: [PATCH 07/11] test(net): run source_errors on the moq-clock simulator Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/tests/integration/source_errors.rs | 34 +++++++++---------- 1 file changed, 17 insertions(+), 17 deletions(-) diff --git a/rs/moq-net/tests/integration/source_errors.rs b/rs/moq-net/tests/integration/source_errors.rs index 7d55fb0481..b3662a0849 100644 --- a/rs/moq-net/tests/integration/source_errors.rs +++ b/rs/moq-net/tests/integration/source_errors.rs @@ -40,11 +40,11 @@ struct Relayed { async fn resolve(origin: &moq_net::origin::Producer) -> moq_net::broadcast::Consumer { let consumer = origin.consume(); - moq_net_sim::timeout(TIMEOUT, consumer.routed("bcast")) + moq_clock::timeout(TIMEOUT, consumer.routed("bcast")) .await .expect("announce timeout") .expect("routed"); - moq_net_sim::timeout(TIMEOUT, consumer.request_broadcast("bcast", None)) + moq_clock::timeout(TIMEOUT, consumer.request_broadcast("bcast", None)) .await .expect("resolve timeout") .expect("broadcast resolves") @@ -101,20 +101,20 @@ fn assert_ended(result: Result<(), Error>, expected: StreamError, version: &str) /// #5347: a TRACK request is in flight to an upstream publisher session that dies before /// answering. The relay's request, and the far end's through it, find the path has no /// route; neither is an internal error. -#[moq_net_sim::test] +#[moq_clock::test] async fn request_in_flight_to_a_dead_upstream_is_unroutable() { for version in VERSIONS { let setup = relayed(version).await; let _track = setup.broadcast.create_track("catalog", timed()).unwrap(); // The relay's requests never reach the publisher. setup.upstream.client_transport.hold_bidis(); - let relayed = moq_net_sim::spawn(read_to_end(setup.relayed.clone(), "catalog")); - let remote = moq_net_sim::spawn(read_to_end(setup.remote.clone(), "catalog")); - moq_net_sim::sleep(Duration::from_secs(1)).await; + let relayed = moq_clock::sim::spawn(read_to_end(setup.relayed.clone(), "catalog")); + let remote = moq_clock::sim::spawn(read_to_end(setup.remote.clone(), "catalog")); + moq_clock::sleep(Duration::from_secs(1)).await; assert!(!relayed.is_finished(), "{version}: answered through a held stream"); setup.upstream.server.abort(Error::Timeout); for reader in [relayed, remote] { - let result = moq_net_sim::timeout(TIMEOUT, reader) + let result = moq_clock::timeout(TIMEOUT, reader) .await .expect("request hung") .unwrap(); @@ -125,7 +125,7 @@ async fn request_in_flight_to_a_dead_upstream_is_unroutable() { /// A subscription already delivering when the upstream session dies, with nothing to /// replace it: the far end learns the path is gone, not that its own session closed. -#[moq_net_sim::test] +#[moq_clock::test] async fn track_in_flight_through_a_dead_upstream_is_unroutable() { for version in VERSIONS { let setup = relayed(version).await; @@ -134,13 +134,13 @@ async fn track_in_flight_through_a_dead_upstream_is_unroutable() { let mut group = track.append_group().unwrap(); group.write_frame(Timestamp::ZERO, b"head".as_ref()).unwrap(); group.finish().unwrap(); - moq_net_sim::timeout(TIMEOUT, sub.recv_group()) + moq_clock::timeout(TIMEOUT, sub.recv_group()) .await .expect("group timeout") .unwrap() .unwrap(); setup.upstream.server.abort(Error::Timeout); - let result = moq_net_sim::timeout(TIMEOUT, async { + let result = moq_clock::timeout(TIMEOUT, async { while sub.recv_group().await?.is_some() {} Ok(()) }) @@ -153,7 +153,7 @@ async fn track_in_flight_through_a_dead_upstream_is_unroutable() { /// #4349: the broadcast is withdrawn under an active subscription, then its track fails /// with a specific error. The withdrawal does not disturb the subscription, which ends /// with that error. -#[moq_net_sim::test] +#[moq_clock::test] async fn withdrawn_broadcast_keeps_the_source_error() { for version in VERSIONS { let setup = relayed(version).await; @@ -162,15 +162,15 @@ async fn withdrawn_broadcast_keeps_the_source_error() { let mut group = track.append_group().unwrap(); group.write_frame(Timestamp::ZERO, b"head".as_ref()).unwrap(); group.finish().unwrap(); - moq_net_sim::timeout(TIMEOUT, sub.recv_group()) + moq_clock::timeout(TIMEOUT, sub.recv_group()) .await .expect("group timeout") .unwrap() .unwrap(); setup.broadcast.unannounce(); - moq_net_sim::sleep(Duration::from_secs(1)).await; + moq_clock::sleep(Duration::from_secs(1)).await; track.abort(Error::App(7)).unwrap(); - let result = moq_net_sim::timeout(TIMEOUT, async { + let result = moq_clock::timeout(TIMEOUT, async { while sub.recv_group().await?.is_some() {} Ok(()) }) @@ -181,15 +181,15 @@ async fn withdrawn_broadcast_keeps_the_source_error() { } /// A subscribe on a broadcast whose route was withdrawn is refused as unroutable. -#[moq_net_sim::test] +#[moq_clock::test] async fn subscribe_after_withdrawal_is_unroutable() { for version in VERSIONS { let setup = relayed(version).await; let _track = setup.broadcast.create_track("video", timed()).unwrap(); setup.broadcast.unannounce(); - moq_net_sim::sleep(Duration::from_secs(1)).await; + moq_clock::sleep(Duration::from_secs(1)).await; for broadcast in [setup.relayed.clone(), setup.remote.clone()] { - let result = moq_net_sim::timeout(TIMEOUT, read_to_end(broadcast, "video")) + let result = moq_clock::timeout(TIMEOUT, read_to_end(broadcast, "video")) .await .expect("subscribe hung"); assert_ended(result, StreamError::Unroutable, version); From f19219aa019734c4740cabcd3f1e2977dc2998ca Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 17 Oct 2026 07:59:25 -0700 Subject: [PATCH 08/11] fix(mux): a route update neither restarts a resolve nor its stitch bound An announce Update is the same instance over another route, so it no longer cancels an in-flight resolve or clears a refused return. A stitch that re-resolves after a real announcement keeps its original deadline, so repeated replacements cannot extend the linger bound. Co-Authored-By: Claude Opus 5.5 --- rs/moq-mux/src/container/ts/export_test.rs | 35 ++++++++++++++++++++++ rs/moq-mux/src/container/ts/follower.rs | 31 +++++++++++++------ 2 files changed, 57 insertions(+), 9 deletions(-) diff --git a/rs/moq-mux/src/container/ts/export_test.rs b/rs/moq-mux/src/container/ts/export_test.rs index 0e53aa5001..ede7647eb5 100644 --- a/rs/moq-mux/src/container/ts/export_test.rs +++ b/rs/moq-mux/src/container/ts/export_test.rs @@ -9392,6 +9392,41 @@ async fn a_midstream_stitch_catalog_expires_at_the_linger() { } } +/// Re-pricing a replacement while a stitch resolves it is the same instance: it neither +/// restarts the resolve nor extends the linger bound. +#[tokio::test(start_paused = true)] +async fn a_midstream_stitch_bound_survives_route_updates() { + let origin = crate::source::produce_origin(); + let _first = publish_bare(&origin, Some(&moq_net::Epoch::mint())); + let linger = Duration::from_secs(10); + let mut follower = follower(&origin).await.with_stitch(true).with_linger(linger); + let start = tokio::time::Instant::now(); + let second = origin + .publish( + "live", + moq_net::origin::Route::default().with_epoch(moq_net::Epoch::mint()), + ) + .unwrap(); + let _unanswered = second.dynamic(); + let reprice = async { + for cost in 1u64.. { + tokio::time::sleep(Duration::from_secs(1)).await; + second.announce(second.route().unwrap().with_cost(cost)).unwrap(); + } + }; + let end = tokio::time::timeout(linger * 3, async { + tokio::select! { + end = follower.next() => end, + () = reprice => unreachable!("re-pricing never stops"), + } + }) + .await + .expect("updates extended the stitch bound"); + let err = end.expect_err("the stitch fails at its linger"); + assert!(err.to_string().contains("catalog"), "{err}"); + assert_eq!(start.elapsed(), linger); +} + /// A catalog ready before the bound completes the switch, including zero linger. #[tokio::test(start_paused = true)] async fn a_midstream_stitch_catalog_within_linger_clears_the_bound() { diff --git a/rs/moq-mux/src/container/ts/follower.rs b/rs/moq-mux/src/container/ts/follower.rs index 6147495f34..f20a4f6fb9 100644 --- a/rs/moq-mux/src/container/ts/follower.rs +++ b/rs/moq-mux/src/container/ts/follower.rs @@ -304,10 +304,19 @@ impl Follower { fallback, } => { // Whatever the request waits on, the announcements say what serves the path - // now: rather than sit out a route that went, ask afresh from there. A stitch - // stays on the export it has, still awaiting what it was. + // now: rather than sit out a route that went, ask afresh from there. if self.poll_announced(waiter) { tracing::debug!(path = %self.path, "announcement while resolving, asking again"); + // A stitch still facing a replacement asks that one, within its original bound. + if matches!(*fallback, State::Running { .. }) && matches!(self.serving, Serving::Other(_)) { + self.state = State::Following { + resolving: self.resolve(), + awaiting, + fallback, + }; + continue; + } + // Otherwise it stays on the export it has, still awaiting what it was. self.state = *fallback; // The fallback's wait predates this poll, so it gets no same-poll grace. if let Some(expired) = self.expired() { @@ -363,22 +372,25 @@ impl Follower { } } - /// Apply every announcement on hand, returning whether there was any. + /// Apply every announcement on hand, returning whether any changed what serves the path. fn poll_announced(&mut self, waiter: &kio::Waiter) -> bool { let mut any = false; while !self.closed { - match self.announced.poll_next(waiter) { + any |= match self.announced.poll_next(waiter) { Poll::Ready(Some(event)) => self.apply(event), - Poll::Ready(None) => self.closed = true, + Poll::Ready(None) => { + self.closed = true; + true + } Poll::Pending => break, - } - any = true; + }; } self.refused &= !any; any } - fn apply(&mut self, event: moq_net::announce::Event) { + /// Apply an announcement, returning whether it changed what serves the path. + fn apply(&mut self, event: moq_net::announce::Event) -> bool { let first = !std::mem::replace(&mut self.started, true); self.serving = match event { moq_net::announce::Event::Start(announce) => { @@ -405,8 +417,9 @@ impl Follower { } // The same instance over another route, which subscriptions ride out. A gap that // ended the export's request comes as an end and a start instead. - moq_net::announce::Event::Update(_) => return, + moq_net::announce::Event::Update(_) => return false, }; + true } /// Begin resolving the instance serving the path now. From da07a6dc9d88adf56b9811888b3a377d63dd988c Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 17 Oct 2026 08:01:44 -0700 Subject: [PATCH 09/11] test(mux): count re-prices without an unbounded range Co-Authored-By: Claude Opus 5.5 --- rs/moq-mux/src/container/ts/export_test.rs | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/rs/moq-mux/src/container/ts/export_test.rs b/rs/moq-mux/src/container/ts/export_test.rs index ede7647eb5..ae5e33f606 100644 --- a/rs/moq-mux/src/container/ts/export_test.rs +++ b/rs/moq-mux/src/container/ts/export_test.rs @@ -9409,8 +9409,10 @@ async fn a_midstream_stitch_bound_survives_route_updates() { .unwrap(); let _unanswered = second.dynamic(); let reprice = async { - for cost in 1u64.. { + let mut cost = 0u64; + loop { tokio::time::sleep(Duration::from_secs(1)).await; + cost += 1; second.announce(second.route().unwrap().with_cost(cost)).unwrap(); } }; From d66245ea052fda0b96dc8c3c3446f9ba5776b939 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 17 Oct 2026 08:25:44 -0700 Subject: [PATCH 10/11] fix: a transport-failed session resets JS streams Unroutable A session that failed at the transport carries no close code, so JS saw the raw session-sourced error and reset with INTERNAL_ERROR where Rust's Transport and IdleTimeout reset with UNROUTABLE. The source_errors check no longer lets a far end's own session failure pass as Unroutable, and the draft changelog attaches UNROUTABLE to the reset. Co-Authored-By: Claude Opus 5.5 --- drafts/draft-lcurley-moq-lite.md | 2 +- js/net/src/error.test.ts | 4 ++++ js/net/src/error.ts | 6 +++++- rs/moq-net/tests/integration/source_errors.rs | 6 +++++- 4 files changed, 15 insertions(+), 3 deletions(-) diff --git a/drafts/draft-lcurley-moq-lite.md b/drafts/draft-lcurley-moq-lite.md index 079cd8fc6f..631f725e90 100644 --- a/drafts/draft-lcurley-moq-lite.md +++ b/drafts/draft-lcurley-moq-lite.md @@ -1598,7 +1598,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 a stream served through an upstream session that ended with UNROUTABLE, rather than INTERNAL_ERROR. +- 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. diff --git a/js/net/src/error.test.ts b/js/net/src/error.test.ts index 5160183efd..143870469c 100644 --- a/js/net/src/error.test.ts +++ b/js/net/src/error.test.ts @@ -341,6 +341,10 @@ test("toStreamCode: a local condition maps to the code the peer can act on", () 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 diff --git a/js/net/src/error.ts b/js/net/src/error.ts index 01cd5ae1f3..4f76982764 100644 --- a/js/net/src/error.ts +++ b/js/net/src/error.ts @@ -357,7 +357,8 @@ interface TransportErrorOptions { * 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. + * 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 @@ -379,6 +380,9 @@ function localStreamCode(err: unknown): StreamCode { 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; } diff --git a/rs/moq-net/tests/integration/source_errors.rs b/rs/moq-net/tests/integration/source_errors.rs index b3662a0849..92b81051d7 100644 --- a/rs/moq-net/tests/integration/source_errors.rs +++ b/rs/moq-net/tests/integration/source_errors.rs @@ -90,8 +90,12 @@ async fn read_to_end(broadcast: moq_net::broadcast::Consumer, name: &'static str #[track_caller] fn assert_ended(result: Result<(), Error>, expected: StreamError, version: &str) { match result { + // A session or transport failure converts to Unroutable too, so it must not pass as one. Err(err) => assert!( - err.stream() == Some(&expected) || (err.stream().is_none() && StreamError::from(&err) == expected), + err.stream() == Some(&expected) + || (!matches!(err, Error::Session(_) | Error::Transport(_) | Error::IdleTimeout) + && err.stream().is_none() + && StreamError::from(&err) == expected), "{version}: ended with {err:?}, expected {expected:?}" ), Ok(()) => panic!("{version}: ended cleanly, expected {expected:?}"), From 5bc9594260c6dd268328b9b99aa9d47306d38fa5 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 17 Oct 2026 22:16:12 -0700 Subject: [PATCH 11/11] quest: drop the decode-errors link to the completed dropped-sources quest Co-Authored-By: Claude Opus 5.5 --- quest/m1/decode-errors.md | 1 - 1 file changed, 1 deletion(-) diff --git a/quest/m1/decode-errors.md b/quest/m1/decode-errors.md index 9a54c249f9..0c0b54583f 100644 --- a/quest/m1/decode-errors.md +++ b/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