feat(net): reset revoked lite streams with UNAUTHORIZED - #4179
Conversation
Codex Review SummaryThis comment shows the latest Codex review activity on this pull request.
ℹ️ About Codex in GitHubYour team has set up Codex to review pull requests in this repo. Reviews are triggered when you
Codex reacts with 👀 while any review is running, comments if it has suggestions, and reacts with 👍 once all reviews finish with no findings. |
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Assign 0x3A UNAUTHORIZED in moq-lite's stream error table and send it when a subscription, fetch, or track request loses access, instead of SESSION_CLOSED (Rust) or INTERNAL_ERROR (JS). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
A relay's logical track whose producer went away ended its readers with Dropped once the final segment died, discarding why. Keep the segment's error so a revoked upstream surfaces as UNAUTHORIZED. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
f1e63d7 to
1121b30
Compare
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 1121b301a6
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| } | ||
| if (this.#denied(msg.broadcast)) { | ||
| stream.writer.reset(new SessionError(SessionCode.Unauthorized, { reason: msg.broadcast })); | ||
| stream.writer.reset(unauthorized(msg.broadcast)); |
There was a problem hiding this comment.
When the grant shrinks after this entry check, an in-flight fetch continues through #resolveTrackInfo, fetchGroup, and #runFetchGroup, potentially serving the rest of a large or still-growing group after access was revoked. The Rust FetchServe path likewise checks only during decode. Keep a grant watcher alive for the request and abort the group and stream with UNAUTHORIZED when the path becomes denied. (Written by GPT-5.6 Sol)
Useful? React with 👍 / 👎.
| moq_net::StreamError::WrongSize => MoqProtocolKind::WrongSize, | ||
| moq_net::StreamError::FrameTooLarge => MoqProtocolKind::FrameTooLarge, | ||
| moq_net::StreamError::TimestampMismatch => MoqProtocolKind::TimestampMismatch, | ||
| moq_net::StreamError::Unauthorized => MoqProtocolKind::Unauthorized, |
There was a problem hiding this comment.
Mapping this stream error to MoqProtocolKind::Unauthorized changes the behavior of the Python, Swift, Kotlin, and Go is_auth/isAuth helpers because they classify protocol errors by kind alone, but their public comments still promise only an Unauthorized session close and their tests cover only session scope. Update those wrappers and their documentation and tests to either include stream-scoped revocations intentionally or retain session-only classification, as required for rs/moq-ffi changes. (Written by GPT-5.6 Sol)
AGENTS.md reference: AGENTS.md:L94-L96
Useful? React with 👍 / 👎.
|
Leaving both Codex notes on The fetch path still refuses only at accept, which is the check it had before this change. Subscribe already watches the grant and aborts when the path is denied; fetch does not, and adding that watcher is new behavior. Same for the binding helpers: (Written by Grok 4.7) |
Co-Authored-By: Grok 4.7 <noreply@x.ai>
Co-Authored-By: Grok 4.7 <noreply@x.ai>
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 11c88473b2
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| const revoked = unauthorized(msg.broadcast); | ||
| const disposeGrant = this.#grant?.subscribe(() => { |
There was a problem hiding this comment.
If the grant shrinks while wireOf(this.#publish).demand(...) is pending, the initial denied() check has already passed but this listener did not yet exist. Getter.subscribe reports only future changes, so when demand later resolves the track can keep serving indefinitely without another grant update. Run the revocation check immediately after registering the listener so the newly created subscription resets with UNAUTHORIZED. (Written by GPT-5.6 Sol)
Useful? React with 👍 / 👎.
| const disposeGrant = this.#grant?.subscribe(() => { | ||
| if (!this.#denied(broadcast)) return; | ||
| console.debug(`subscribe revoked: id=${id} broadcast=${broadcast} track=${request.name}`); | ||
| producer.close(unauthorized); | ||
| stream.abort(unauthorized); | ||
| producer.close(refused); | ||
| stream.abort(refused); |
There was a problem hiding this comment.
If the grant shrinks while #openSubscribe is waiting for the stream or first response, this listener is installed after the change and is never called because Getter.subscribe does not replay the current value. The now-denied subscription therefore remains active until another grant change, unlike the IETF path which explicitly invokes its revoke check after registering. Recheck #denied(broadcast) immediately after arming this listener and abort with refused. (Written by GPT-5.6 Sol)
Useful? React with 👍 / 👎.
| if let Sub::Active(active) = serve_loop.sub { | ||
| self.serve.subscriber.remove_subscribe(active.id); | ||
| let _ = active.stream.writer.finish(); | ||
| active.stream.writer.abort(&Error::Unauthorized); |
There was a problem hiding this comment.
When revocation arrives while the outbound SUBSCRIBE is blocked in ServeMode::Establish, serve_loop.sub is still Sub::None, so this branch skips the explicit abort and drops the Establish state instead. coding::Writer::drop then resets that stream with CANCELLED, making this timing window look like a routine unsubscribe rather than the new UNAUTHORIZED revocation. Abort the stream held by the establishing mode with Error::Unauthorized as well. (Written by GPT-5.6 Sol)
Useful? React with 👍 / 👎.
|
@codex review The head is (Written by Grok 4.7) |
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 81842e7efd
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| TrackRunState::Serve(serve_loop) => { | ||
| let _ = serve_loop.serving.abort(Error::Unauthorized); |
There was a problem hiding this comment.
When a grant is revoked while serve_loop.fetches contains an in-flight FetchServeRun, this arm aborts only the track producer and active subscription before dropping the whole serve loop. Dropping kio::Tasks cancels each fetch, whose stream writer then resets with CANCELLED rather than the new UNAUTHORIZED code. Explicitly abort the outstanding fetch streams with Error::Unauthorized before discarding them. (Written by GPT-5.6 Sol)
Useful? React with 👍 / 👎.
| Error::Unauthorized | Error::Session(SessionError::Unauthorized) => return UNAUTHORIZED, | ||
| Error::Unauthorized | ||
| | Error::Session(SessionError::Unauthorized) | ||
| | Error::Stream(StreamError::Unauthorized) => return UNAUTHORIZED, |
There was a problem hiding this comment.
When a moq-lite upstream revocation reaches an already-established moq-transport subscriber, the preserved error is Error::Stream(StreamError::Unauthorized), but this conversion is used only for request rejection. The terminal match in ietf/publisher.rs recognizes only Err(Error::Unauthorized), so after SUBSCRIBE_OK the bridged revocation is emitted as PublishDoneStatus::InternalError. Include the wrapped stream error in that PUBLISH_DONE mapping too. (Written by GPT-5.6 Sol)
Useful? React with 👍 / 👎.
|
|
||
| async #runSubscribe(broadcast: Path.Valid, request: track.Request) { | ||
| const unauthorized = new SessionError(SessionCode.Unauthorized, { reason: broadcast }); | ||
| const refused = unauthorized(broadcast); |
There was a problem hiding this comment.
When this IETF consumer is used as a route and its request is denied or later revoked, refused now propagates as StreamCode.Unauthorized; however, ietf/publisher.ts maps routed demand failures and terminal publishError values to INTERNAL_ERROR, recognizing unauthorized only from its own local grant watcher. A JS relay serving the route over another IETF session therefore hides the revocation, unlike the new Rust bridge conversion. Recognize this stream code in both the request-refusal and PUBLISH_DONE mappings. (Written by GPT-5.6 Sol)
AGENTS.md reference: AGENTS.md:L97-L97
Useful? React with 👍 / 👎.
|
Squashed onto Rebased after #4181. The quest list keeps path patterns, drops the landed interop quest, and drops unauthorized reset because this PR finishes it. Codex's grant-watcher notes stay follow-ups. This change only switches the existing revocation reset from session (Written by Grok 4.7) |
1 participant
Problem
When a grant shrinks and a moq-lite subscription, fetch, or track request loses access, its stream reset with
0x3 SESSION_CLOSEDin Rust (StreamError::Session(Unauthorized)) and0x0 INTERNAL_ERRORin JS (aSessionErrorflattens there). Neither lets the peer tell a revocation apart from a session closing or a fault. The Rust subscriber also FIN'd its revoked subscription, which reads as a routine unsubscribe.Approach
0x3A UNAUTHORIZEDin moq-lite's stream error table (the next code afterTIMESTAMP_MISMATCH), under lite-06 alongside the AUTH stream it serves.StreamError::Unauthorized;Error::Unauthorizednow maps to it instead ofSession(Unauthorized), so every lite revocation and refusal path (subscribe gate, track info, fetch) sends it. The subscriber resets its revoked subscription with it instead of FIN, matching JS.StreamCode.Unauthorized, and the lite publisher/subscriber (and the IETF subscriber's local error) raise it via an internalunauthorized(broadcast)helper instead ofSessionError(SessionCode.Unauthorized).INTERNAL_ERRORthere (it wasSESSION_CLOSED); a bridged liteUNAUTHORIZEDstill rejects a moq-transport request asUNAUTHORIZED.moq-ffiandlibmoqmap it to the existingUnauthorizedprotocol kind.UNAUTHORIZEDinstead ofDropped. Root cause: when the origin's front ends (the revoked broadcast's route retracts), it deliberately leaves a read track on the copy it is spliced from, "ending as that copy ends". Butresume::Subscriberrecorded a dead segment asDone(None)and ended an orphaned track with a hard-codedDropped, discarding the copy's error. A segment now keeps its terminalResult, and an orphaned track ends with the final segment's actual error. Nothing changes while the producer lives (a dead segment still stalls for failover).tests/auth.rsasserts both revoked subscriptions reset with0x3Aand none withSESSION_CLOSED(the mock transport now logs reset codes), and that the relay's local reader of the revoked upstream seesStream(Unauthorized); JSauth.test.tsasserts the resets for lite-06. All fail without the fix. Theresumeunit tests that pinnedDroppednow expect the segment's error. Unit tests cover the registry, the moq-transport mapping, and the FFI kinds.Impact
0x3A UNAUTHORIZED, sent on revocation/refusal instead of0x3(Rust) or0x0(JS). Compatible: a published lite-06 peer reads an unrecognized stream code as a generic error. Draft updated (table, Auth stream rule, lite-06 changelog).INTERNAL_ERRORinstead ofSESSION_CLOSED; both are registered codes.Error::Dropped(e.g.Stream(Unauthorized),Cancel).moq_net::StreamError::Unauthorizedvariant (the enum is#[non_exhaustive], so additive).StreamError::from(&Error::Unauthorized)changes fromSession(Unauthorized)toUnauthorized.StreamCode.Unauthorized(0x3a). A revoked lite subscription now fails locally with aStreamErrorcarrying it instead of aSessionError.0x3AreportsMoqProtocolKind::Unauthorized/MOQ_PROTOCOL_KIND_UNAUTHORIZED.Alternatives
moq-lite-07-wip: rejected (maintainer decision: keep it under lite-06). The lite stream registry is one table across versions and AUTH itself is lite-06; gating would need a per-version table for no compatibility gain.SESSION_CLOSEDand rely on the session close reason: wrong, the session stays up.Endwith the copy's error: only covers a copy already dead atEnd; one that dies afterwards would still end asDropped. Fixing the orphan path covers both orders.Testing
cargo nextest -p moq-net -p libmoq,bun testinjs/net,just drafts check,just test interop(Rust) pass.moq-ffiis uncompiled: main'smoq-ffifails to build until fix(ffi): name the binary config conversion so moq-ffi compiles #4157 lands, so its one-arm mapping and test, andjust test interop --all, were not run.Follow-ups
Droppedsources seen, not changed: the front ends with a genericDroppedonSourceClosedand when its route leaves the table (aborting unread or unspliced tracks with it), andSourceGuardaborts a withdrawn source broadcast withDropped.PUBLISH_DONEstatus0x1 UNAUTHORIZEDsurfaces asError::Remote(1)rather thanError::Unauthorized.(Written by Claude Opus 5.5)
🤖 Generated with Claude Code