fix(moq-net): deliver groups from a takeover source that restarted its numbering - #2534
fix(moq-net): deliver groups from a takeover source that restarted its numbering#2534wrangelvid wants to merge 4 commits into
Conversation
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
There was a problem hiding this comment.
Hey - I've left some high level feedback:
- In
Consumer::poll_latest_changed, theSplicedcase silently returnsPoll::Pending; consider either documenting this explicitly at the call site or making it an invariant (e.g., via a debug assert) that this helper is only used withPlainconsumers to avoid future misuse. - The new
erafield and restart logic inresume::Producer::reanchor_restartedintroduce a second timeline alongsideepoch; consider grouping or encapsulating the epoch/era handling (e.g., via a small helper or struct) so callers and future changes don’t have to remember the subtle interaction between the two counters.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- In `Consumer::poll_latest_changed`, the `Spliced` case silently returns `Poll::Pending`; consider either documenting this explicitly at the call site or making it an invariant (e.g., via a debug assert) that this helper is only used with `Plain` consumers to avoid future misuse.
- The new `era` field and restart logic in `resume::Producer::reanchor_restarted` introduce a second timeline alongside `epoch`; consider grouping or encapsulating the epoch/era handling (e.g., via a small helper or struct) so callers and future changes don’t have to remember the subtle interaction between the two counters.Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
…bering A broadcast in its linger window accepts a replacement source via takeover(), which splices it in at group_start = latest + 1 on the assumption that the replacement resumes the previous sequence space. A real reconnecting publisher (a fresh session with a fresh broadcast::Producer) restarts numbering at zero, so every group it publishes falls below the segment boundary and is filtered on the read side: subscribers stall silently until the restarted source's sequence happens to catch up, swallowing as many groups as the old source ever delivered. For a low-rate event feed (one group per event) this is a lasting outage: a relay consuming an event publisher that restarts loses the next N events after the reconnect, where N is everything delivered before it. The existing test_linger_reconnect_splices constructs its continuation group at the resume boundary explicitly, so this path was uncovered. The new test fails with 'group would have blocked' on main. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
A takeover splices the replacement source at one past the newest delivered group, assuming it resumes the old sequence space. A real reconnecting publisher (fresh session, fresh producer) restarts numbering at zero, so every group it publishes sits below the boundary: the demand floor keeps the wire subscription from ever requesting its groups, and the read cursor filters anything that arrives, starving subscribers silently for as many groups as the old source ever delivered. Takeover splices are now PROVISIONAL: the boundary binds the read cursor but stays out of wire demand until the source's first production shows which case this is. Producing at or past the boundary proves a resume and promotes the floor into demand (subscribers re-slice on the epoch bump); producing sequence zero below a boundary of two or more proves a restart, and the splice resets onto the new source alone so subscribers reconcile onto its own numbering. In between (a resumed source re-delivering history into an unfloored subscription) stays provisional and filtered. Boundary one is the ambiguous corner, resolved as a resume: a restarted source recovers at its next group, which lands on the boundary. The origin dispatcher watches a freshly spliced copy's production edge (a new demand-free track pollable) and resolves the splice; the route tests pin the deferred-floor contract and the restart test delivers a restarted source's first group. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
A restart reanchor re-splices the logical track onto a source with its own numbering, but subscribers created between the takeover and the deciding production carry floors resolved against the OLD numbering (a session's SUBSCRIBE start resolves from the pre-restart latest). Those floors filter everything the restarted source will number for as many groups as the old source ever delivered. The resume state now carries a numbering era, bumped only by a restart reanchor; a subscriber observing an era change resets its monotonic position and start floor before reconciling segments, so the restarted source's groups surface under their own numbering. The late-join-floor test fails without the reset. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
d7023cc to
356e97b
Compare
There was a problem hiding this comment.
Sorry @wrangelvid, you have reached your weekly rate limit of 500000 diff characters.
Please try again later or upgrade to continue using Sourcery
WalkthroughIntroduces production-aware splice settlement and numbering-era tracking. Serving now distinguishes resumed and restarted replacement tracks using route origin identity and first-production observation. Restarted numbering increments a resume era and resets subscriber sequence cursors and floors. Route and linger reconnect tests cover the updated splice and delivery behavior. 🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches✨ Simplify code
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
|
I'll check in a second, but we shouldn't be trying to resume broadcasts with different origin IDs. Like the relay needs soooome indication that a broadcast is new versus continuous, and right now we should be using the first hop as that flag. |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
rs/moq-net/src/model/resume.rs (1)
396-412: 🎯 Functional Correctness | 🟠 Major | ⚡ Quick winSeed
erafrom the current state, not0.
Subscriber::erastarts at0whileResumeState::eramay already be nonzero (any earlier restart reanchor). The firstpoll_syncthen sees an era mismatch that did not happen during this subscriber's lifetime and clearsmin_sequence/next_sequence, silently discarding astart_at()floor set before the first poll. That is the exact late-join pattern the new test drives, so a subscriber attaching after a restart has already settled loses its floor once.🐛 Proposed fix
pub(crate) fn subscribe_shared(&self, prefs: kio::Producer<Subscription>) -> Subscriber { let last_prefs = prefs.read().clone(); + let era = self.state.read().era; Subscriber { state: self.state.clone(), prefs, last_prefs, epoch: 0, - era: 0, + era,🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@rs/moq-net/src/model/resume.rs` around lines 396 - 412, Initialize Subscriber::era in subscribe_shared from the current ResumeState era rather than zero. Preserve the existing state and preference initialization, and ensure a late-joining subscriber retains any start_at() floor when the state has already undergone a restart reanchor.
🧹 Nitpick comments (2)
rs/moq-net/src/model/origin.rs (1)
3740-3797: 🎯 Functional Correctness | 🔵 Trivial | ⚡ Quick winAdd a restarted-source case where the first observed
latestis above zero.Every restart test appends exactly one group before settling, so
reanchor_restartedalways seeslatest == 0, which is the only value its current heuristic treats as a restart. Appending two or three groups back-to-back beforesettle().await(or producing before the splice) exercises the coalesced-observation path that the heuristic misses; see the related comment onrs/moq-net/src/model/resume.rs.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@rs/moq-net/src/model/origin.rs` around lines 3740 - 3797, The test test_linger_reconnect_restarted_sequences_still_deliver currently only observes latest sequence 0 after reconnect. Extend the restarted-source scenario to append at least two groups before settling or splicing, then assert the restarted groups are delivered with the expected sequences, exercising reanchor_restarted when the first observed latest value is above zero.rs/moq-net/src/model/track.rs (1)
1568-1578: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winThe closed arm ignores
known, so it stays ready forever.Once the plain track closes,
state.pollyieldsErr(final_state)on every poll and this returnsReady(closed.max_sequence)even when it equalsknown. Today the caller is saved only by branch ordering:poll_completeis checked before this inserve_track's wait closure and is unconditionally ready on close, so the loop exits viaStep::Failed/Step::Complete. If those branches are ever reordered, an unsettled splice would busy-spin. Cheap to make self-consistent:♻️ Proposed change
- Poll::Ready(Err(closed)) => Poll::Ready(closed.max_sequence), + // A closed track produces nothing further, so only report an actual change. + Poll::Ready(Err(closed)) if closed.max_sequence != known => Poll::Ready(closed.max_sequence), + Poll::Ready(Err(_)) => Poll::Pending,🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@rs/moq-net/src/model/track.rs` around lines 1568 - 1578, Update the closed-state arm of the state.poll match in the track wait logic to compare closed.max_sequence with known and return Poll::Pending when they are equal; only return Poll::Ready when the final sequence advances, preserving the existing behavior for successful polls and pending states.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@rs/moq-net/src/model/resume.rs`:
- Around line 266-280: Update the restart-detection condition in the reanchoring
logic after latest() so any latest value below start is treated as a restarted
source, rather than requiring latest != 0. Retain only the existing start < 2
boundary-of-one ambiguity guard, and preserve the Pending outcome for that
ambiguous case.
---
Outside diff comments:
In `@rs/moq-net/src/model/resume.rs`:
- Around line 396-412: Initialize Subscriber::era in subscribe_shared from the
current ResumeState era rather than zero. Preserve the existing state and
preference initialization, and ensure a late-joining subscriber retains any
start_at() floor when the state has already undergone a restart reanchor.
---
Nitpick comments:
In `@rs/moq-net/src/model/origin.rs`:
- Around line 3740-3797: The test
test_linger_reconnect_restarted_sequences_still_deliver currently only observes
latest sequence 0 after reconnect. Extend the restarted-source scenario to
append at least two groups before settling or splicing, then assert the
restarted groups are delivered with the expected sequences, exercising
reanchor_restarted when the first observed latest value is above zero.
In `@rs/moq-net/src/model/track.rs`:
- Around line 1568-1578: Update the closed-state arm of the state.poll match in
the track wait logic to compare closed.max_sequence with known and return
Poll::Pending when they are equal; only return Poll::Ready when the final
sequence advances, preserving the existing behavior for successful polls and
pending states.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro Plus
Run ID: 199af34e-500e-4913-a2c6-0f02c9506eb0
📒 Files selected for processing (3)
rs/moq-net/src/model/origin.rsrs/moq-net/src/model/resume.rsrs/moq-net/src/model/track.rs
Per review: the first hop is the producer-instance identity, so the takeover no longer infers resumed-vs-restarted from what the source produces. A route whose first hop matches the previously serving one splices at the boundary exactly as before (the demand floor stays eager, and the route tests keep their original assertions, with their synthetic hops adjusted so alternate routes to the same producer share a first hop). A differing first hop is a new broadcast: the logical track resets onto it unbounded and the numbering era bumps, which clears subscriber cursors and late-join floors anchored in the old numbering. This replaces the provisional-splice and production-heuristic machinery from the previous revision; what remains of it is a one-shot warn when a same-identity source produces below the splice boundary (an identity-contract violation that would starve those groups), plus the Origin doc spelling out that ids are minted per producer instance. The restart tests model a reconnect as a fresh session with a fresh identity, which is what real ones are. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Oh I didn't see that. I reworked this by keeping the boundary splice on the same first hop, and a different one resets the logical track onto the new source with an era bump, which should also clear later join floors resolved against the old numbering. |
There was a problem hiding this comment.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (2)
rs/moq-net/src/model/resume.rs (2)
614-620: 🎯 Functional Correctness | 🟠 Major | ⚡ Quick winDrop an in-flight old-era group before returning frames.
poll_read_framechecksself.readingbefore it synchronizes producer state. Ifreset_ontooccurs after an old group entersreading, the next call can return a frame from the replaced broadcast. Synchronize first and clearreadingon an era change.Proposed fix
if era != self.era { self.era = era; self.next_sequence = 0; self.min_sequence = 0; + self.reading = None; } pub fn poll_read_frame(&mut self, waiter: &kio::Waiter) -> Poll<Result<Option<frame::Frame>>> { + self.poll_sync(waiter); loop {As per coding guidelines, add a regression test that selects an old group, resets the producer, then verifies the next frame comes from the new source.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@rs/moq-net/src/model/resume.rs` around lines 614 - 620, Update poll_read_frame to synchronize producer state before checking or returning from self.reading, so an era change detected by reset_onto clears any in-flight old-era group before frames are emitted. Add a regression test covering selection of an old group, producer reset, and verification that the next frame comes from the new source.Source: Coding guidelines
356-356: 🎯 Functional Correctness | 🟠 Major | ⚡ Quick winSeed new subscribers with the current era.
A reset while no subscriber exists leaves
state.era > 0. This subscriber then treats its first snapshot as a restart and clears a caller-setstart_atfloor before its first poll. Initialize fromself.state.read().eraso only resets after subscription creation clear its cursors.Proposed fix
pub(crate) fn subscribe_shared(&self, prefs: kio::Producer<Subscription>) -> Subscriber { let last_prefs = prefs.read().clone(); + let era = self.state.read().era; Subscriber { state: self.state.clone(), prefs, last_prefs, epoch: 0, - era: 0, + era,As per coding guidelines, add a regression test that fails without this fix.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@rs/moq-net/src/model/resume.rs` at line 356, Initialize the new subscriber’s era from self.state.read().era instead of the hardcoded 0 in the subscriber construction path, preserving caller-provided start_at when prior resets occurred before subscription. Add a regression test covering a reset with no subscribers followed by subscription and verify the first poll does not clear its start_at floor.Source: Coding guidelines
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Outside diff comments:
In `@rs/moq-net/src/model/resume.rs`:
- Around line 614-620: Update poll_read_frame to synchronize producer state
before checking or returning from self.reading, so an era change detected by
reset_onto clears any in-flight old-era group before frames are emitted. Add a
regression test covering selection of an old group, producer reset, and
verification that the next frame comes from the new source.
- Line 356: Initialize the new subscriber’s era from self.state.read().era
instead of the hardcoded 0 in the subscriber construction path, preserving
caller-provided start_at when prior resets occurred before subscription. Add a
regression test covering a reset with no subscribers followed by subscription
and verify the first poll does not clear its start_at floor.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro Plus
Run ID: 24bcfd3c-ca2d-4ff5-9fdd-2312f8b80fa4
📒 Files selected for processing (3)
rs/moq-net/src/model/origin.rsrs/moq-net/src/model/resume.rsrs/moq-net/src/model/track.rs
🚧 Files skipped from review as they are similar to previous changes (2)
- rs/moq-net/src/model/track.rs
- rs/moq-net/src/model/origin.rs
2 participants
Summary
latest + 1, assuming it resumes the old sequence space. A reconnecting publisher is a fresh session with a fresh identity and restarted numbering, so the boundary starves it twice: the demand floor asks it for groups it will never number, and late-join floors resolved against the pre-restartlatest()filter anything that does arrive. One-group-per-event feeds lose the next N events after every publisher reconnect.Origindoc now states that ids are minted per producer instance.Public API changes
None. New items are
pub(crate)(resume::Producer::reset_onto,resume::Producer::splice_boundary,track::Consumer::poll_latest_changed). One doc addition onOrigin.Test plan
test_linger_reconnect_restarted_sequences_still_deliverfails on main, passes here.test_linger_restart_resets_late_join_floorsfails without the era cursor reset, passes here.test_linger_reconnect_splicesand the route handover/failover/cost tests pass with their original assertions; 559 moq-net tests green.(Written by Claude Fable 5)