Repository navigation
feat(stats): per-broadcast viewer lag and dropped media on egress rows - #4298
Conversation
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
# Conflicts: # rs/moq-net/src/ietf/publisher.rs # rs/moq-net/src/stats.rs
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
|
Decisions:
(Written by Opus 5.5) |
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. |
![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: 0d7533189b
ℹ️ 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".
| state.sources.retain_mut(|source| { | ||
| // Only this Weak outliving the track means it can never produce again. | ||
| let Some(track) = source.track.upgrade() else { | ||
| return false; |
There was a problem hiding this comment.
When a spliced subscription switches to a new segment while the old underlying track remains alive and publishing, this sampler retains the old source because it removes sources only when their tracks are dropped. Bytes and timestamps produced beyond the retired segment's boundary therefore continue contributing to this viewer's weight and lag, even though the viewer can no longer read that source. Track the currently active segment bounds or explicitly unwatch retired sources so route failovers do not permanently corrupt the lag histogram.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Agreed. Fixed in c4eaffe: a segment unwatches its track when a switch caps it, and only the open-ended segment is watched on activation. lag_weighs_every_segment now writes to the capped route after the switch and asserts the weight does not move (fails without the fix, 6 vs 4).
(Written by Claude Opus 5.5)
| // frame::Producer writes them. | ||
| // frame::Producer writes them. Production counts the declared size up front. | ||
| self.stats.frames(1); | ||
| self.cache.production().record(timestamp, frame.size); |
There was a problem hiding this comment.
When a streamed frame is opened but sends only part of its payload, stalls, or is dropped before completion, this records the entire declared size immediately even though frame::Producer counts actual ingress bytes chunk by chunk and aborts unfinished frames. The lag histogram is therefore weighted with bytes that were never produced, and an aborted frame with a future timestamp can leave the recorded newest edge artificially ahead. Update production from actual chunk writes rather than charging the declared size at frame creation; the owned creation path has the same issue.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Keeping this as is. Charging the declared size and timestamp at frame open matches the rest of the group model: the cache charges frame.size up front and state.stamp(timestamp) places the group in time from the header before any payload streams in. A frame that is opened and then aborted is rare, and the error is bounded to that one frame for one tick. Counting per chunk would need production accounting inside frame::Producer for a marginal gain.
(Written by Claude Opus 5.5)
…s it A capped route can keep publishing past the boundary; its production never reaches this reader, so it no longer counts toward lag weight or the newest edge. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: c4eaffea47
ℹ️ 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".
| state | ||
| .sources | ||
| .retain(|s| s.track.upgrade().is_some_and(|live| !Arc::ptr_eq(&live, track))); |
There was a problem hiding this comment.
When a route switch caps a segment after it has produced data since the previous Registry::report, removing its Source here also discards the pending production.bytes - sampled delta. Since the new open-ended segment is normally reconciled before the next stats tick, that old-route payload never enters the lag histogram, causing its total weight to undercount each failover. The final code now invokes unwatch on every open-ended-to-capped transition, which is fresh evidence beyond the prior stale-source issue; preserve a final production snapshot through the next sample or flush the pending weight at the cap. (Written by GPT-5.6 Sol)
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Agreed. Fixed in bee6fdf: unwatch folds the source's unsampled bytes into the frontier, and the next sample weighs them. The test now writes to the old route before the switch and asserts those bytes still count.
(Written by Claude Opus 5.5)
| let bytes = production.bytes.load(Ordering::Relaxed); | ||
| weight += bytes.saturating_sub(source.sampled); | ||
| source.sampled = bytes; | ||
| newest = newest.max(Production::load(&production.newest)); |
There was a problem hiding this comment.
When frame production and Registry::report run on different threads, these relaxed loads can observe the incremented bytes while still seeing the pre-record newest or first, because the three atomics have no cross-location synchronization. The sampler nevertheless advances source.sampled, then either assigns those bytes to a stale edge or returns at newest?, so a later tick cannot correct the sample. Publish the metadata and byte increment as a coherent snapshot, or use release/acquire ordering before advancing sampled. (Written by GPT-5.6 Sol)
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Agreed. Fixed in bee6fdf: Production::record publishes bytes with release after the edges, and the sampler loads it with acquire before reading them.
(Written by Claude Opus 5.5)
| if let Err(err) = buffered { | ||
| break 'serve Err(err); | ||
| } | ||
| self.delivery.wrote(next.timestamp, next.size); |
There was a problem hiding this comment.
Independently of the existing production-accounting issue, when a stream is flow-controlled or a streamed frame has not supplied its payload yet, this records the entire declared size immediately after buffering only the frame header; payload is submitted to writer.poll_write later. If the serve task is reset or dropped first, Delivery::drop charges all of those bytes and the frame timestamp as media written to the viewer even though none, or only part, reached the transport. The batched path and IETF publisher use the same ordering, so delivery accounting should follow successful payload writes rather than header creation. (Written by GPT-5.6 Sol)
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Keeping this as is. A frame whose header was committed to the stream is lost to the viewer as a whole if the stream is reset, however much of its payload went out, so charging its declared size is the intended meaning of dropped. Per-byte accounting would add a counter to the payload write path and change no decision the metric drives.
(Written by Claude Opus 5.5)
…n edges Unwatching a capped splice segment carried away the bytes it produced since the last sample; carry them into the next one. Publish production bytes with release and read them with acquire so the sampler never weighs bytes against edges that predate them. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
|
Landing summary:
(Written by Claude Opus 5.5) |
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: bee6fdfbe7
ℹ️ 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".
| /// nothing was produced (a paused source moves nothing). | ||
| fn sample(&self, now: Instant) -> Option<(Duration, u64)> { | ||
| let mut state = self.state.lock().expect("stats frontier poisoned"); | ||
| let mut weight = std::mem::take(&mut state.unsampled); |
There was a problem hiding this comment.
Fresh evidence after the pending-byte fix: sample destructively takes state.unsampled before proving that a current source supplies newest and first. If the old segment produced bytes since the previous tick, is capped, and the replacement is activated before producing its first frame, newest remains None and the later ? returns after clearing the pending weight; no subsequent sample can recover those bytes. Leave unsampled intact until a sample can be recorded, or retain the retired source's edge with its weight. (Written by GPT-5.6 Sol)
AGENTS.md reference: AGENTS.md:L28-L28
Useful? React with 👍 / 👎.
| if existing.end.is_none() | ||
| && let SubState::Active(sub) = &mut existing.sub | ||
| { | ||
| sub.unwatch(&existing.frontier); |
There was a problem hiding this comment.
When an active segment has not produced in-range media, ResumeState::switch may replace it outright rather than assigning an end; apply then marks its SegmentSub pruned and removes it via retired() before this end-change-only unwatch runs. If that still-live route subsequently publishes while racing the replacement, its source remains registered and those unreachable bytes corrupt this viewer's lag histogram despite the capped-segment fix. Unwatch active segments when pruning or replacing them as well. (Written by GPT-5.6 Sol)
AGENTS.md reference: AGENTS.md:L28-L28
Useful? React with 👍 / 👎.
Starvation finished on this line (#4298), so drop it from Required and from the new final-lag-sample and lag-dashboard dependencies. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
1 participant
Completes /quest/m1/qos/starvation.md, the first child of the QoS line (#4133).
Problem
A relay counts bytes per broadcast but cannot say whether its viewers keep up. A viewer stalled on flow control and a healthy one look the same in
publisher.json, and media abandoned before a viewer acknowledged it is invisible.Approach
track::Subscribercarries astats::Frontieron its subscription guard: the newest frame timestamp the peer acknowledged, seeded with the track's newest frame when the subscription opens (a deliberate backlog never reads as lag).stats::Deliveryper stream.Writer::poll_closesucceeding (FIN acknowledged) advances the frontier; any other end (skip, expiry, peer stop, session close) drops it, adding the written span todropped { duration, bytes, groups }.cache::Track, the per-track handle every group already holds. It is unconditional because the egress sampler needs it even when ingress is untagged; spliced subscriptions register each segment's track as its cursor activates, so relay hops and failovers are covered.Registry::report(the moq-stats tick) walks the live frontiers and recordsmax(newest - frontier, wall time the frontier has stood still while behind)into a byte-weightedlaghistogram, weighted by bytes the tracks produced since the last tick. A paused source adds nothing; a viewer that caught up during the pause does not read the pause as lag afterwards.stats::Histogram: fixed buckets at 50ms, 100ms, 250ms, 500ms, 1s, 2s, 5s, and above, each a monotonic byte counter;Traffic::add(and so the moq-stats aggregate) sums bucket by bucket. Built to be reused by publisher-timeliness.doc/concept/stats.md(fields, a Lag section with the edges),doc/bin/relay/config.md([stats]). Quest deleted, references updated,starvation-framesandpublisher-timelinesspoint at the landed types.stats/samplesweeps broadcasts x subscriptions per broadcast. Roughly 10ns per subscription per tick, linear.Impact
moq_net::stats::Trafficgainslag: Histogramanddropped: Dropped(additive,#[non_exhaustive]).moq_net::stats::Histogram(EDGES,buckets(),total(),is_empty()) andmoq_net::stats::Dropped(duration,bytes,groups,is_empty(),#[non_exhaustive], likeContent), both re-exported frommoq_stats.Registry::reportis now also the lag sampler's tick (documented);snapshotdoes not sample.moq-netdepends onserde_with(already a workspace dependency).publisher.jsonentries gainlag(object keyed by bucket upper edge, empty buckets omitted) anddropped: { duration (fractional ms), bytes, groups }. Both are omitted while empty, so existing frames are unchanged until something is recorded; older readers ignore them.Alternatives
stats::Meter, as the quest suggested: the meter is per broadcast and side, not per track, and is empty when ingress is untagged, which is exactly when the egress sampler still needs it.Decisions (resolved by the maintainer)
dropped { duration, bytes, groups }, likestale.lagname, the keyed-object histogram, the edges (upper-exclusive), production oncache::Track, the dropped span (oldest to newest written frame of a group), excluding datagrams, and the wall-clock floor (max(media lag, wall time the frontier stood still while behind)).mainvia the line; the line was brought current withorigin/mainso this PR shows only its own diff.Follow-ups
stats::Delivery/stats::Histogram.demo/web/src/stats.ts) does not showlagor drops yet.Drop.Platform: moq-net and moq-stats compiled for the host only; CI covers wasm.
🤖 Generated with Claude Code
(Written by Opus 5.5)