fix(js/net): serialize subscription updates with group pops - #2820
Conversation
25e11af to
5156c25
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: ef79ad8d2a
ℹ️ 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".
Co-Authored-By: GPT-5 <noreply@openai.com>
Review follow-ups on the control/pop serialization, all internal to `@moq/net`. No wire or public API change. `tryRecvGroup` returned `GroupConsumer | Error | null | undefined`, which callers had to test in one exact order, and which collapsed "idle" and "finished but holding groups above the cap" into the same `undefined`. The publisher recovered the difference by re-reading `track.closed`, duplicating a check the read had already made. Return the four outcomes as a tagged `Recv` instead, mirroring the Rust publisher's `Recv::Group` / `Recv::Boundary` / `Recv::Finished`, so the ordering rule disappears and the serving loop reads the state once. Decode SUBSCRIBE_UPDATE into a single-message slot rather than an unbounded array. The decoder parks until the serving loop takes what it decoded, so a peer flooding updates while the loop is blocked in a control-stream write backs up in QUIC flow control instead of on our heap. This is also one message per turn, like the Rust publisher's poll. Teardown releases a decoder parked on a full slot through `writer.closed`, which both the clean and the error path settle. Fold the writer's terminal state and the peer's FIN into one `done` control: the serving loop already treated them identically, and neither is an update. Add the peer-FIN test the previous change left implicit. Returning on FIN without draining the buffer is deliberate and matches the Rust publisher's `TrackEnd::PeerFin`, which drops in-flight group machines rather than completing them for a subscriber that has left. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
The previous commit decoded SUBSCRIBE_UPDATE into a single-message slot to bound memory. That capped the control drain at one message per data step: `take()` empties the slot and wakes the decoder through a signal, but the serving loop continues synchronously, so the decoder never gets a turn to decode an already-buffered second message before `tryRecvGroup` runs. Two updates arriving while the loop was parked in a write applied one before the pop and one after, and the group went out under the superseded range, putting frames on the wire that the peer's current subscription excluded. Update-then-FIN had the same shape: groups opened for a subscriber that had already left. Read ahead into a queue again. The loop's drain is synchronous and cannot yield to the decoder mid-burst, so anything not already decoded when the loop resumes lands after the next pop instead of before it. The Rust publisher gets this from `poll_decode_maybe`, which decodes out of the reader's buffer synchronously; nothing on this side can, so it reads ahead instead. Found by an adversarial review. Regression test queues two conflicting updates behind a blocked write with a group ready, and fails without this. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Co-Authored-By: GPT-5 <noreply@openai.com>
ef79ad8 to
e2b22d9
Compare
There was a problem hiding this comment.
💡 Codex Review
moq/js/net/src/lite/publisher.ts
Line 661 in e2b22d9
Yield before popping the next buffered group
When a group backlog is readable and another SUBSCRIBE_UPDATE is buffered behind the update just taken, this fire-and-forget call leaves no await on the normal group path, so the loop synchronously pops the entire backlog. The decoder necessarily yields at each asynchronous Reader operation, allowing every group to be positioned under the first update before the newer update or FIN becomes visible, which can send groups or frames the latest bounds exclude. The burst test does not cover this ordering because it holds SUBSCRIBE_START until both updates have decoded; yield and recheck controls before the next pop, with a regression where serving wakes after only the first of two buffered updates. (Written by GPT-5.6 Sol)
AGENTS.md reference: AGENTS.md:L115-L121
ℹ️ 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".
Co-Authored-By: GPT-5 <noreply@openai.com>
There was a problem hiding this comment.
💡 Codex Review
moq/js/net/src/lite/publisher.ts
Line 665 in 55fc179
Race blocked response writes against peer departure
When the SUBSCRIBE_START write is flow-control blocked and the peer FINs its read half, the decoder records done, but this await cannot observe it. A FIN on the peer's independent send half does not unblock this writer, and runSubscribe cannot call stream.close() until #runTrack returns, so the subscription remains attached and retains its queued groups until the entire connection closes. Race response writes against the control termination signal, or actively cancel the writer when decoding observes FIN. (Written by GPT-5.6 Sol)
AGENTS.md reference: AGENTS.md:L98-L98
ℹ️ 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".
Co-Authored-By: GPT-5 <noreply@openai.com>
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 08831e50ca
ℹ️ 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".
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: a980337ae8
ℹ️ 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".
| return; | ||
| } | ||
|
|
||
| void this.#runGroup({ |
There was a problem hiding this comment.
When multiple groups are already buffered, this fire-and-forget dispatch lets the for loop synchronously pop the entire backlog without yielding. SubscriptionControls.#decode must cross at least one await even when a SUBSCRIBE_UPDATE is already buffered in the same reader, so a pipelined update that raises the floor or tightens the cap cannot populate #update before all cached groups are assigned their old ranges. Yield to or otherwise poll the decoder between buffered group pops so the advertised control-first ordering also holds outside response-write backpressure.
Useful? React with 👍 / 👎.
1 participant
Summary
SUBSCRIBE_UPDATErange handling and cap-aware group reads under one publisher state ownerSUBSCRIBE_STARTandSUBSCRIBE_ENDwrites against control termination so peer FIN detaches a blocked subscriptionRoot cause
recvGroup()removed a buffered group synchronously, but the publisher snapshotted mutable frame bounds in a later promise job. A bufferedSUBSCRIBE_UPDATEcould run between those steps and serve the group under bounds it was never taken under.The independent decoder also retained every decoded update while the serving loop was blocked on a write. Since each update restates the full subscription, retaining only the newest pending local range state preserves control-first serving while bounding memory.
Applying one local range update could still wake the serving loop and synchronously drain another buffered group before the decoder completed its next already-delivered framed message. A task boundary on the update path gives the decoder time to publish newer control state before the next pop without adding scheduling cost to normal group delivery.
Response backpressure exposed three further ownership issues. Delaying the whole update behind a blocked response prevented the priority listener from re-ranking group streams when congestion made it most important. Separating immediate full subscription publication from serialized local cursor and frame-bound mutation preserves both behaviors. A peer FIN also could not unblock the independent response half, so response writes now race the decoder's sticky terminal state and release the subscription as soon as the subscriber leaves. Because
Promise.racedoes not cancel its losing promise, termination also resets the writable half so the blocked encode and transport stream cannot remain retained.Public API changes
None. The synchronous group receive, readiness hooks, and control coordination are package-internal.
Test plan
nix develop --command just checknix develop --command just testSUBSCRIBE_START, and reset of the abandoned response writeCloses #2808.
(Written by GPT-5)