fix(stream): detect a rolled log without waiting for the group to end

[kixelated]

Closes #3260.

Summary

• Root cause. A stream track is one group; a second group means records are missing, which all four implementations report as Rolled / Stream.Rolled. But each only asked the track for another group once the held one ended (poll_recv_group / recvGroup gated on group.is_none()). A publisher that opens a second group and leaves the first open (itself the same violation) therefore parked the consumer indefinitely, on a track that already carried the condition the contract says to fail on.
• The consumers now consult the track whenever the held group's read would block. A second group there fails immediately instead of waiting for a group a broken publisher may never finish. A finished track does not truncate the group in hand: its frames may still arrive, so that case keeps waiting as before.
• Drain, then fail (the open question in the issue). Records already received from the first group are still yielded, since they precede the gap; only a read that would actually block consults the track. That keeps the boundary-triggered behavior the existing tests pin, and keeps the four implementations agreeing on what a reader sees. Rust reads the frame first and polls the track only on Pending; TS drains with tryReadFrame (guarded by skipped, which is the evicted prefix readFrame reports as lagged) before racing the blocking read.
• The failure is now sticky. Previously a second poll_next after Rolled could return Ok(None), which reads as a cleanly completed log.
• TS builds the raced recvGroup promise once per consumer, tagged for the race. recvGroup delivers each group exactly once, so a second call would take a second group off the track and lose the first; and tagging it inside the race would chain a fresh reaction onto that one pending promise for every frame the log delivers. That cached promise is also what makes the failure sticky there: a second group leaves it resolved, so dropping the held group sends every later read back to the same throw, with no flag to carry. Rust polls rather than awaits, so it keeps an explicit rolled flag instead.
• A concurrent next() is refused, the way js/json/src/window/consumer.ts already does it. Sharing the one recvGroup promise means two concurrent calls resolve to the same group, and the second would take the first's group for a rolled log and fail a valid one. Rust gets this from &mut self.
• The failure closes both groups: the one it abandons mid-read and the second one it rejects. The read that loses the race stays registered on the held group's signals otherwise, which keeps the group consumer, and through its expiry closure the track subscription, reachable after the caller drops the stream consumer. Closing settles that read; the mirrors are per-subscriber, so this touches neither the producer nor another subscriber.

Public API changes

None. No pub / exported item added, renamed, or re-signed; Error::Rolled and Rolled already existed. Behavior only, which is why this could target main, except the code being changed exists only on dev.

Wire behavior changes

None on the wire. What changes is when a consumer gives up on a non-conforming publisher: previously never, while the first group stayed open; now as soon as the second group is observed and the first has nothing more in hand. drafts/draft-lcurley-moq-hang.md is updated to require that, and to say explicitly that already-received payloads may be yielded first.

Test plan

• One regression test per implementation (a_second_group_is_reported_while_the_first_is_open / "a second group is reported while the first is still open"): two groups opened, neither finished. Without the fix the Rust reads return Pending forever and the TS reads never settle. Each also asserts the failure is sticky, and the TS ones assert both group mirrors are released (dropping the close() calls turns them red).
• A concurrent-read test per TS implementation: the second call is refused rather than served the first's group.
• cargo test -p moq-binary -p moq-json (30 tests) and cargo clippy -p moq-binary -p moq-json --all-targets -- -D warnings: clean.
• bun test js/binary/src/stream js/json/src/stream (26 tests), bun biome check, and bun run --filter=@moq/binary --filter=@moq/json check (tsc): clean.
• Cross-Package Sync: all four implementations plus the draft are in this PR.
• Reviewed adversarially by Codex; both findings (the concurrent read, the abandoned read) are fixed above.

(Written by claude-opus-5)

[chatgpt-codex-connector]

Codex Review Summary

This comment shows the latest Codex review activity on this pull request.

Review	Status	Commit	Review trigger
📝 Code Review	✅ Completed 2026-09-01T21:30:48.426874Z	9b6c866	New commits

ℹ️ 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" or "@codex security review".

Codex reacts with 👀 while any review is running, comments if it has suggestions, and reacts with 👍 once all reviews finish with no findings.

[chatgpt-codex-connector]

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: 0cc66c68b2

ℹ️ 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".

[chatgpt-codex-connector]

Cancel the losing frame read

When the second group wins this race while the first group remains open, the losing readFrame() cannot be cancelled and stays subscribed to the first group's signals indefinitely. That pending async operation retains the Group.Consumer, whose expiry closure retains the track subscriber, so the exact broken-publisher scenario handled here can keep the subscription and its state alive even after the caller discards the stream consumer. Race group.readable() or another cancellable readiness primitive and only start readFrame() after readiness wins; the identical JSON implementation needs the same correction. (Written by GPT-5.6 Sol)

Useful? React with 👍 / 👎.