feat(json): a sliding-window mode, appended to the back and dropped from the front - #3168
feat(json): a sliding-window mode, appended to the back and dropped from the front#3168kixelated wants to merge 1 commit into
Conversation
…rom the front Neither existing mode fits a bounded, joinable run of records. `stream` keeps a log forever in one group, so a long-running publisher outgrows the group cache budget and every new consumer fails with `Lagged`. `snapshot` keeps only the latest value. The obvious fix -- roll the log's group and re-seed the new one with the records still retained -- breaks the reader. Re-seeded records are indistinguishable from new ones on the wire, so a consumer that was keeping up receives them twice. Add a mode where the restatement is explicit instead, so a reader can tell "you already have these" from "here is another one". Every frame is a tagged op. Each group opens with a `reset` naming the retained records and the absolute offset of the first, followed by `push` and `pop`. Only the reset carries an index; a push takes the next one and a pop drops from the front, both positional against the reset, which is sound because the group-scoped DEFLATE window already makes a mid-group join undecodable. Trimming is therefore an op, not a group boundary: dropping a record costs one small frame inside the shared compression window rather than a roll that would throw that window away. The producer rolls when a group's ops outgrow `op_ratio` times the reset that opened it, exactly as `snapshot` rolls on its delta budget. That is purely a compression decision, and the consumer never surfaces it: a reset restating records already delivered yields nothing, so however often the publisher rolls, a reader sees one continuous stream of events. Every index is reported exactly once -- `Push` on arrival, `Pop` on leaving, and `Skip` for a record that existed but was dropped before this reader saw it. A reader that joins late or falls a group behind learns from the reset's offset what it will never receive, rather than silently missing it. Records are stored pre-encoded, so a record's bytes are identical whether a reader gets it as a push or in a later reset. A failed write does not roll the window back: the edit really happened, only the consumer's knowledge of it is lost, and the next reset repairs that. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
![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: 88fb95fed3
ℹ️ 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".
| for index in delivered..offset { | ||
| self.events.push_back(Event::Skip(index)); |
There was a problem hiding this comment.
When a decoder has processed one valid reset, a later tiny reset can set offset arbitrarily far ahead and this loop eagerly allocates one Event::Skip per intervening index before decode returns. A remote publisher can therefore hang or exhaust the consumer with two small frames, such as offsets 0 and u64::MAX; the matching TypeScript loop in js/json/src/window/decoder.ts also blocks the browser main thread. Represent the gap compactly or enforce a bounded advance instead of materializing an untrusted range.
Useful? React with 👍 / 👎.
| ratio != 0 | ||
| && self.group_frames > 0 | ||
| && self.group_frames < MAX_GROUP_FRAMES | ||
| && self.op_bytes <= ratio * self.reset_len |
There was a problem hiding this comment.
For large records, this permits a group to exceed moq-net's 32 MiB byte cache even though it stays below the 256-frame and ratio limits. For example, a 16 MiB reset followed by a 15 MiB push, a pop of the original record, and another 15 MiB push leaves a restatable 30 MiB window but a roughly 46 MiB group, so rs/moq-net/src/model/group.rs::MAX_GROUP_CACHE evicts the reset and a late reader receives Lagged instead of joining. Track cumulative group bytes, including the pending op, and roll before the reset can be evicted; the matching TypeScript budget has the same defect.
Useful? React with 👍 / 👎.
| pub fn reset(&mut self) { | ||
| self.flate = None; | ||
| } |
There was a problem hiding this comment.
After the decoder has consumed its first group, this method leaves delivered populated, so an uncompressed push or pop as the first frame of a later group passes the MissingReset check and is silently applied against the previous group's position. Consumer::poll_next calls this method at every group boundary, so a malformed or incorrectly routed group produces corrupt indices rather than the documented error. Preserve the cross-group delivery cursor, but separately clear and check whether the current group has received its reset.
Useful? React with 👍 / 👎.
| const dropped = Math.min(count, this.#window.length); | ||
| if (dropped <= 0) return undefined; | ||
|
|
||
| this.#window.splice(0, dropped); | ||
| this.#offset += dropped; |
There was a problem hiding this comment.
When callers pass a fractional or non-finite JavaScript number, Math.min preserves it while splice coerces it differently. For example, pop(0.5) removes no array element but advances the offset by 0.5 and emits {"pop":0.5}, immediately desynchronizing the producer and consumer; NaN additionally creates an invalid reset containing NaN. Validate that count is a nonnegative safe integer, or expose an integer-constrained type, before mutating state.
AGENTS.md reference: AGENTS.md:L165-L168
Useful? React with 👍 / 👎.
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. |
1 participant
Summary
Neither existing mode fits a bounded, joinable run of records.
streamkeeps a log forever in one group, so a long-running publisher outgrows the group cache budget and every new consumer fails withLagged.snapshotkeeps only the latest value.The obvious fix — roll the log's group and re-seed the new one with the records still retained — breaks the reader. A
streamconsumer yields every frame in order across group boundaries, so re-seeded records are indistinguishable from new ones and a consumer that was keeping up receives them twice. This adds a mode where the restatement is explicit, so a reader can tell "you already have these" from "here is another one".Wire
Every frame is a tagged op:
{"reset":{"offset":N,"records":[...]}}opens every group, naming the retained records and the absolute index of the first.{"push":REC}appends to the back.{"pop":N}drops from the front.Only the reset carries an index. A push takes the next one and a pop drops from the front, both positional against the reset — sound because the group-scoped DEFLATE window already makes a mid-group join undecodable, so a reader never sees a subset of a group's frames.
Trimming is therefore an op, not a group boundary. Dropping a record costs one small frame inside the shared compression window, rather than a roll that would throw that window away.
Rolling is invisible
The producer rolls when a group's ops outgrow
op_ratiotimes the reset that opened it, exactly assnapshotrolls on its delta budget, with the same 256-frame cap. That is purely a compression decision: no caller-driven cut, no age bound, and the consumer never surfaces it. A reset restating records already delivered yields nothing, so however often the publisher rolls, a reader sees one continuous stream of events.rolling_is_invisible_to_the_consumerpins this by running the same edits atop_ratio1000 and 0 and comparing the event streams.What a reader is told
Every index is reported exactly once:
Pushon arrival,Popon leaving,Skipfor a record that existed but was dropped before this reader saw it. A reader that joins late, or falls a group behind, learns from the reset's offset what it will never receive rather than silently missing it. A fresh reader adopts the first reset's offset instead of skipping all of history.Two decisions worth review
Valuein Rust, JSON text in TS), and a reset splices them in rather than re-serializing. So a record's bytes are identical whether a reader gets it as a push or in a later reset.Public API changes
Purely additive; no existing item renamed, removed, or signature-changed.
moq_json::windowmodule:Producer,Consumer,Encoder,Decoder,Event,Encoded,Pending,ProducerConfig,ConsumerConfig.moq_json::Error::MissingResetvariant (the enum is#[non_exhaustive]).@moq/jsonWindowexport with the matching surface.Wire behavior changes
None to any existing mode. This is a new track format that nothing in-tree publishes or consumes yet; the broadcast timeline moves onto it separately, which is where the hang draft's Track Framing gets updated.
Test plan
just checkandjust test(1949 tests, 1 skipped) green.op_ratio: 0(where every edit restates the window), rolling being invisible, compressed round-trip across many rolls, a fresh consumer adopting the offset without spurious skips, and a lagging consumer being told what it missed with no gaps in the reported indices.rs/moq-json/tests/window-vectors.jsonpins the exact frame bytes for all six op shapes, asserted from both suites, following the existing differ-vector precedent.(written by claude-opus-5)