Conversation
WalkthroughAdds probe stream support across JS/TS and Rust codepaths. JS: routes StreamId.Probe to Publisher.runProbe, adds Publisher.runProbe and Probe.decodeMaybe. Rust: treats Probe control frames as probe listeners, adds recv_probe and run_probe implementations and imports Duration/Stats. Cargo.toml enables tokio "time" feature. Minor cleanup: removed #[allow(...)] attributes around probe re-export and Probe struct. No existing public signatures were removed. 🚥 Pre-merge checks | ✅ 2 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (2 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches
🧪 Generate unit tests (beta)
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 |
![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
🧹 Nitpick comments (3)
js/lite/src/lite/publisher.ts (2)
267-275: Consider callingstream.abort()instead of only resetting the writer.When
getStatsis unsupported, only the writer is reset (line 273), leavingstream.readeropen. Since this is a bidi stream, the reader side will linger until the peer closes. Usingstream.abort(...)would cleanly tear down both sides, consistent with the error path on line 315.Proposed fix
if (!quic.getStats) { - stream.writer.reset(new Error("stats not supported")); + stream.abort(new Error("stats not supported")); return; }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@js/lite/src/lite/publisher.ts` around lines 267 - 275, In runProbe, when quic.getStats is unavailable the code only resets the writer which leaves the bidi stream's reader open; replace the writer-only reset with a full abort by calling stream.abort(new Error("stats not supported")) (or call stream.abort before/after stream.writer.reset) so both reader and writer are torn down consistently with the error path used elsewhere; update the branch that currently calls stream.writer.reset(new Error("stats not supported")) to use stream.abort(...) on the Stream object to ensure the peer side is not left hanging.
280-311: UncancelledsetTimeouton loop exit.When
stream.reader.closedwins the race (line 285-286), the pendingsetTimeoutfrom line 283 is never cleared. While this is functionally harmless (the callback resolves a detached promise), it delays GC and causes a spurious timer fire. Capturing and clearing the timer ID would be cleaner.Proposed fix
try { for (;;) { - const timeout = new Promise<"timeout">((resolve) => - setTimeout(() => resolve("timeout"), PROBE_INTERVAL), - ); + let timer: ReturnType<typeof setTimeout>; + const timeout = new Promise<"timeout">((resolve) => { + timer = setTimeout(() => resolve("timeout"), PROBE_INTERVAL); + }); const result = await Promise.race([timeout, stream.reader.closed]); - if (result !== "timeout") break; + if (result !== "timeout") { + clearTimeout(timer!); + break; + }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@js/lite/src/lite/publisher.ts` around lines 280 - 311, The loop starts a setTimeout Promise but never clears its timer when stream.reader.closed wins the Promise.race, leaving an uncancelled timer; update the loop in publish logic to keep the timer ID returned by setTimeout (for the Promise created at the top of the loop), and call clearTimeout(timerId) when the Promise.race result is not "timeout" (i.e., just before breaking out) so the pending timer is cancelled; make this change where the timeout Promise is created and where the code currently checks result !== "timeout" (references: stream.reader.closed, stream.writer, Probe.encode, quic.getStats, lastSentBitrate, lastSentTime, PROBE_INTERVAL).rs/moq-lite/src/lite/publisher.rs (1)
96-108: AfterPROBE_MAX_AGE, probes are sent unconditionally — even if bitrate is unchanged.When
elapsed >= PROBE_MAX_AGE(10s),thresholdevaluates to0.0. Sincechange(an absolute value ratio) is always≥ 0.0, the conditionchange >= thresholdis alwaystrue, meaning the same bitrate value will be re-sent. This effectively acts as a periodic keepalive.If this is intentional (heartbeat/refresh), a brief comment would help future readers understand the design. If not, consider using
>instead of>=on Line 106 to avoid sending whenchange == 0.0.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@rs/moq-lite/src/lite/publisher.rs` around lines 96 - 108, The probe logic currently treats elapsed >= PROBE_MAX_AGE as threshold == 0.0 so any change >= threshold will re-send even when bitrate is unchanged; update the comparison in the Some((prev, at)) arm (where change and threshold are computed) from >= to > to prevent re-sending when change == 0.0, and add a one-line comment near the PROBE_MAX_AGE/threshold logic explaining that probes beyond PROBE_MAX_AGE should not act as a keepalive (or explicitly note if the periodic refresh is intentional).
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@rs/moq-lite/src/lite/publisher.rs`:
- Around line 79-95: The code references a non-existent Stats type and calls
session.stats().estimated_send_rate() inside run_probe; remove the unused Stats
import and stop calling session.stats().estimated_send_rate(). Modify run_probe
(the async fn run_probe) to accept bitrate information from the caller (e.g.,
add a parameter like estimated_bitrate: Option<u64> or a callback Fn() ->
Option<u64>) and use that parameter in place of
session.stats().estimated_send_rate(); update all call sites to provide the
bitrate or None, or alternatively drop the bitrate-dependent probe logic and
always continue when bitrate is unavailable. Ensure all references to
session.stats() and the Stats import are removed.
---
Nitpick comments:
In `@js/lite/src/lite/publisher.ts`:
- Around line 267-275: In runProbe, when quic.getStats is unavailable the code
only resets the writer which leaves the bidi stream's reader open; replace the
writer-only reset with a full abort by calling stream.abort(new Error("stats not
supported")) (or call stream.abort before/after stream.writer.reset) so both
reader and writer are torn down consistently with the error path used elsewhere;
update the branch that currently calls stream.writer.reset(new Error("stats not
supported")) to use stream.abort(...) on the Stream object to ensure the peer
side is not left hanging.
- Around line 280-311: The loop starts a setTimeout Promise but never clears its
timer when stream.reader.closed wins the Promise.race, leaving an uncancelled
timer; update the loop in publish logic to keep the timer ID returned by
setTimeout (for the Promise created at the top of the loop), and call
clearTimeout(timerId) when the Promise.race result is not "timeout" (i.e., just
before breaking out) so the pending timer is cancelled; make this change where
the timeout Promise is created and where the code currently checks result !==
"timeout" (references: stream.reader.closed, stream.writer, Probe.encode,
quic.getStats, lastSentBitrate, lastSentTime, PROBE_INTERVAL).
In `@rs/moq-lite/src/lite/publisher.rs`:
- Around line 96-108: The probe logic currently treats elapsed >= PROBE_MAX_AGE
as threshold == 0.0 so any change >= threshold will re-send even when bitrate is
unchanged; update the comparison in the Some((prev, at)) arm (where change and
threshold are computed) from >= to > to prevent re-sending when change == 0.0,
and add a one-line comment near the PROBE_MAX_AGE/threshold logic explaining
that probes beyond PROBE_MAX_AGE should not act as a keepalive (or explicitly
note if the periodic refresh is intentional).
ℹ️ Review info
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (7)
js/lite/src/lite/connection.tsjs/lite/src/lite/probe.tsjs/lite/src/lite/publisher.tsrs/moq-lite/Cargo.tomlrs/moq-lite/src/lite/mod.rsrs/moq-lite/src/lite/probe.rsrs/moq-lite/src/lite/publisher.rs
💤 Files with no reviewable changes (2)
- rs/moq-lite/src/lite/mod.rs
- rs/moq-lite/src/lite/probe.rs
| async fn run_probe(session: &S, stream: &mut Stream<S, Version>, _version: Version) -> Result<(), Error> { | ||
| const PROBE_INTERVAL: Duration = Duration::from_millis(100); | ||
| const PROBE_MAX_AGE: Duration = Duration::from_secs(10); | ||
| const PROBE_MAX_DELTA: f64 = 0.25; | ||
|
|
||
| let mut last_sent: Option<(u64, tokio::time::Instant)> = None; | ||
| let mut interval = tokio::time::interval(PROBE_INTERVAL); | ||
|
|
||
| loop { | ||
| tokio::select! { | ||
| res = stream.reader.closed() => return res, | ||
| _ = interval.tick() => {} | ||
| } | ||
|
|
||
| let Some(bitrate) = session.stats().estimated_send_rate() else { | ||
| continue; | ||
| }; |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Check the Stats trait definition for estimated_send_rate
echo "=== Stats trait ==="
ast-grep --pattern 'fn estimated_send_rate($$$) -> $_ { $$$ }'
echo "=== Stats trait (trait def) ==="
rg -n 'estimated_send_rate' --type rust -C 3
echo "=== lite::Probe struct ==="
rg -n 'struct Probe' --type rust -C 5
echo "=== Probe encode impl ==="
rg -n 'impl.*Encode.*Probe\|impl.*Probe' --type rust -C 3Repository: moq-dev/moq
Length of output: 1145
🏁 Script executed:
#!/bin/bash
# Search for Stats trait definition more broadly
echo "=== Stats trait definition ==="
rg -n 'trait Stats' --type rust -C 5
echo "=== estimated_send_rate in trait context ==="
rg -n 'fn estimated_send_rate' --type rust -B 2 -A 2
echo "=== Session type and stats method ==="
rg -n 'fn stats\(&self\)' --type rust -B 2 -A 2Repository: moq-dev/moq
Length of output: 168
🏁 Script executed:
#!/bin/bash
# Look for session.stats() pattern and Stats trait across all files
echo "=== Session trait or type ==="
rg -n 'trait Session|pub trait Session' --type rust -B 1 -A 10
echo "=== All .rs files with Stats ==="
find . -name "*.rs" -type f | xargs rg 'trait Stats|pub trait Stats' -l
echo "=== Files in moq-lite/src/lite ==="
ls -la rs/moq-lite/src/lite/
echo "=== Check lib.rs or mod.rs for Stats export ==="
find rs/moq-lite/src -name "lib.rs" -o -name "mod.rs" | xargs cat -n 2>/dev/nullRepository: moq-dev/moq
Length of output: 6336
🏁 Script executed:
#!/bin/bash
# Read the publisher.rs file to understand the context
echo "=== publisher.rs - full run_probe function and surrounding context ==="
head -n 150 rs/moq-lite/src/lite/publisher.rs | tail -n 100
echo "=== Search for 'stats' in publisher.rs ==="
rg -n 'stats' rs/moq-lite/src/lite/publisher.rs -B 2 -A 2
echo "=== Search for Stats definition across entire repo ==="
rg -n 'struct Stats|enum Stats' --type rust
echo "=== Search for estimated_send_rate definition ==="
rg -n 'estimated_send_rate' --type rust -B 3 -A 3Repository: moq-dev/moq
Length of output: 3832
🏁 Script executed:
#!/bin/bash
# Read the beginning of publisher.rs to see struct and impl definition
echo "=== Beginning of publisher.rs ==="
head -n 80 rs/moq-lite/src/lite/publisher.rs
echo "=== Look for impl blocks and struct definition ==="
rg -n 'pub struct Publisher|impl.*Publisher' rs/moq-lite/src/lite/publisher.rs -B 2 -A 5Repository: moq-dev/moq
Length of output: 2782
🏁 Script executed:
#!/bin/bash
# Check if web_transport_trait is a local crate
echo "=== Looking for web_transport_trait in repo ==="
find . -name "*web_transport*" -type d 2>/dev/null | head -20
echo "=== Check Cargo.toml for web_transport_trait ==="
find . -name "Cargo.toml" -type f | head -5 | xargs grep -l "web_transport_trait" 2>/dev/null
echo "=== Check Cargo.toml in rs/moq-lite ==="
cat rs/moq-lite/Cargo.toml 2>/dev/null | head -40
echo "=== Search for web_transport_trait crate definition ==="
rg -n 'pub trait Session|pub struct Stats' rs/web-transport-trait/ --type rust 2>/dev/null | head -20Repository: moq-dev/moq
Length of output: 1013
🏁 Script executed:
#!/bin/bash
# Look for workspace Cargo.toml
echo "=== Root Cargo.toml or workspace config ==="
cat Cargo.toml 2>/dev/null | grep -A 20 "workspace\|web-transport-trait"
echo "=== Check all Cargo.toml files for web-transport-trait ==="
find . -name "Cargo.toml" -type f -exec grep -l "web-transport-trait" {} \;
echo "=== Search for web-transport-trait in all Cargo files ==="
find . -name "Cargo.toml" -type f -exec grep -B 2 -A 2 "web-transport-trait" {} +Repository: moq-dev/moq
Length of output: 1512
🌐 Web query:
web-transport-trait 0.3.2 Stats estimated_send_rate return type
💡 Result:
In web-transport-trait v0.3.2 there is no Stats type and no estimated_send_rate API (the crate only defines the Session, SendStream, RecvStream, and Error traits and related helpers). [1]
If you instead mean the browser WebTransport getStats() data: the spec defines estimatedSendRate as unsigned long long? (nullable), i.e. “present but may be null”; in Rust this would typically correspond to Option<u64>. [2]
Sources:
[1] web-transport-trait 0.3.2 source (src/lib.rs) https://docs.rs/crate/web-transport-trait/0.3.2/source/src/lib.rs
[2] W3C WebTransport spec (WebTransportConnectionStats.estimatedSendRate) https://www.w3.org/TR/webtransport/
🏁 Script executed:
#!/bin/bash
# Search for where Stats is actually defined
echo "=== Search for Stats definition in local codebase ==="
rg -n '^pub trait Stats|^trait Stats|^pub struct Stats|^struct Stats' --type rust
echo "=== Search for stats() method implementation ==="
rg -n 'fn stats\(&' --type rust -B 2 -A 3
echo "=== Check if Stats is used elsewhere ==="
rg -n 'use.*Stats' --type rust
echo "=== Check if there's a trait extension for Session ==="
rg -n 'impl.*web_transport_trait::Session' --type rust -A 5Repository: moq-dev/moq
Length of output: 3047
Remove the unused Stats import and the invalid session.stats().estimated_send_rate() call.
The Stats type does not exist in web-transport-trait v0.3.2 (which only exports Session, SendStream, RecvStream, and Error traits), and the Session trait has no stats() method. The code at line 93 will not compile:
let Some(bitrate) = session.stats().estimated_send_rate() else {
continue;
};The stats() method and Stats trait need to be defined locally or the probing logic needs a different approach to obtain bitrate information.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@rs/moq-lite/src/lite/publisher.rs` around lines 79 - 95, The code references
a non-existent Stats type and calls session.stats().estimated_send_rate() inside
run_probe; remove the unused Stats import and stop calling
session.stats().estimated_send_rate(). Modify run_probe (the async fn run_probe)
to accept bitrate information from the caller (e.g., add a parameter like
estimated_bitrate: Option<u64> or a callback Fn() -> Option<u64>) and use that
parameter in place of session.stats().estimated_send_rate(); update all call
sites to provide the bitrate or None, or alternatively drop the
bitrate-dependent probe logic and always continue when bitrate is unavailable.
Ensure all references to session.stats() and the Stats import are removed.
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (1)
js/lite/src/lite/publisher.ts (1)
282-286: Clear pending timer when the stream closes early.When
stream.reader.closedwins the race, the timeout still fires later. Clearing it avoids dangling callbacks and keeps shutdown cleaner.Proposed refactor
- const timeout = new Promise<"timeout">((resolve) => - setTimeout(() => resolve("timeout"), PROBE_INTERVAL), - ); - const result = await Promise.race([timeout, stream.reader.closed]); + let timer: ReturnType<typeof setTimeout> | undefined; + const timeout = new Promise<"timeout">((resolve) => { + timer = setTimeout(() => resolve("timeout"), PROBE_INTERVAL); + }); + const result = await Promise.race([timeout, stream.reader.closed]); + if (timer !== undefined) clearTimeout(timer); if (result !== "timeout") break;🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@js/lite/src/lite/publisher.ts` around lines 282 - 286, The timeout created for the Promise.race uses setTimeout but never clears the timer when stream.reader.closed wins; update the code in the loop (where timeout, result and stream.reader.closed are used) to capture the timer id (e.g., let timer = setTimeout(...)), pass that into the timeout Promise, and call clearTimeout(timer) right after awaiting Promise.race when result !== "timeout" (before breaking) so the pending timer is cancelled and no dangling callback fires later.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@js/lite/src/lite/publisher.ts`:
- Around line 288-307: Normalize and clamp the bitrate before using it: when
reading stats.estimatedSendRate from quic.getStats() (used to construct new
Probe(bitrate) and later encoded), defensively convert to a non-negative integer
via truncation (e.g. raw = stats.estimatedSendRate; bitrate = Math.max(0,
Math.trunc(raw))). Keep the existing null check for estimatedSendRate, then use
this sanitized bitrate in Probe(bitrate).encode to avoid silent BigInt
truncation downstream.
---
Nitpick comments:
In `@js/lite/src/lite/publisher.ts`:
- Around line 282-286: The timeout created for the Promise.race uses setTimeout
but never clears the timer when stream.reader.closed wins; update the code in
the loop (where timeout, result and stream.reader.closed are used) to capture
the timer id (e.g., let timer = setTimeout(...)), pass that into the timeout
Promise, and call clearTimeout(timer) right after awaiting Promise.race when
result !== "timeout" (before breaking) so the pending timer is cancelled and no
dangling callback fires later.
| const stats = await quic.getStats(); | ||
| const bitrate = stats.estimatedSendRate; | ||
| if (bitrate == null) continue; | ||
|
|
||
| let shouldSend: boolean; | ||
| if (lastSentBitrate === undefined || lastSentTime === undefined) { | ||
| shouldSend = true; | ||
| } else if (lastSentBitrate === 0) { | ||
| shouldSend = bitrate > 0; | ||
| } else { | ||
| const elapsed = performance.now() - lastSentTime; | ||
| const t = Math.max(PROBE_INTERVAL, Math.min(PROBE_MAX_AGE, elapsed)); | ||
| const range = PROBE_MAX_AGE - PROBE_INTERVAL; | ||
| const threshold = (PROBE_MAX_DELTA * (PROBE_MAX_AGE - t)) / range; | ||
| const change = Math.abs(bitrate - lastSentBitrate) / lastSentBitrate; | ||
| shouldSend = change >= threshold; | ||
| } | ||
|
|
||
| if (shouldSend) { | ||
| await new Probe(bitrate).encode(stream.writer, this.version); |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
# First, find and examine the Probe class
find . -name "probe.ts" -type fRepository: moq-dev/moq
Length of output: 82
🏁 Script executed:
# Also find Writer implementation to understand u53()
find . -name "*.ts" -type f | xargs grep -l "u53" | head -10Repository: moq-dev/moq
Length of output: 378
🏁 Script executed:
# Check the current implementation of estimatedSendRate handling in publisher.ts
cat -n js/lite/src/lite/publisher.ts | sed -n '285,310p'Repository: moq-dev/moq
Length of output: 1216
🏁 Script executed:
# Examine Probe class implementation
cat -n ./js/lite/src/lite/probe.tsRepository: moq-dev/moq
Length of output: 1512
🏁 Script executed:
# Search for Writer class definition and u53 implementation
rg "class Writer|\.u53\s*\(" --context 5 js/lite/src/lite/ | head -60Repository: moq-dev/moq
Length of output: 3157
🏁 Script executed:
# Check integers.ts which likely has numeric encoding logic
cat -n js/lite/src/catalog/integers.ts | head -100Repository: moq-dev/moq
Length of output: 118
🏁 Script executed:
# Find Writer implementation - it's imported from stream.ts
cat -n js/lite/src/stream.ts | head -150Repository: moq-dev/moq
Length of output: 5135
🏁 Script executed:
# Also search for u53 implementation directly
rg "u53\s*\(" js/lite/src/stream.ts -A 5Repository: moq-dev/moq
Length of output: 692
🏁 Script executed:
# Find Varint implementation
cat -n js/lite/src/varint.ts | head -80Repository: moq-dev/moq
Length of output: 2824
🏁 Script executed:
# Also check what happens with fractional input to encodeTo
rg "encodeTo" js/lite/src/varint.ts -A 10Repository: moq-dev/moq
Length of output: 675
🌐 Web query:
WebTransport API getStats estimatedSendRate type specification
💡 Result:
In the WebTransport specification, estimatedSendRate is a member of the WebTransportConnectionStats dictionary returned by WebTransport.getStats().
- IDL type:
unsigned long long?(nullable), defaulting tonull - Meaning/unit: an estimate of how fast queued data will be sent, in bits per second, computed by the user agent’s congestion control algorithm; it excludes framing overhead and represents application payload rate; it may be
nullif no estimate is available (and can becomenulleven if previously non-null). [1]
Spec location: WebTransport (W3C TR), §6.13 WebTransportConnectionStats. [1]
🌐 Web query:
WebTransport implementation estimatedSendRate fractional decimal bits per second
💡 Result:
estimatedSendRate in the WebTransport stats API is not fractional.
- In the WebTransport spec it is defined as
unsigned long long? estimatedSendRateand described as an estimate “in bits per second”. As an integer IDL type, it represents a whole-number count of bps (ornullif unavailable). [1] - In JavaScript, an IDL
unsigned long longis exposed as a Number (IEEE-754 double). That means it will still be conceptually an integer bps value, but values above 2^53 − 1 may lose integer precision due to Number limitations. [2]
If you’re seeing decimals, they’re coming from your own unit conversion/display (e.g., dividing by 1e6 to show Mbps), not from estimatedSendRate itself.
Sources: [1] WebTransport spec (WebTransportConnectionStats / estimatedSendRate) [2] Web IDL spec (unsigned long long ↔ ECMAScript Number precision)
Consider explicit normalization if fractional bitrates are possible, but per WebTransport spec this is unlikely.
The estimatedSendRate is specified in the WebTransport API as unsigned long long?, defined in bits per second—an integer type. If fractional values are somehow received and passed to Writer.u53(), the Varint.encodeTo() function will silently truncate via BigInt(v) rather than fail. However, since the API spec guarantees an integer type, normalization is defensive rather than critical. If you want to be explicit about truncation, the proposed fix using Math.max(0, Math.trunc(rawBitrate)) is reasonable, though the codebase may already assume compliance with the API contract.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@js/lite/src/lite/publisher.ts` around lines 288 - 307, Normalize and clamp
the bitrate before using it: when reading stats.estimatedSendRate from
quic.getStats() (used to construct new Probe(bitrate) and later encoded),
defensively convert to a non-negative integer via truncation (e.g. raw =
stats.estimatedSendRate; bitrate = Math.max(0, Math.trunc(raw))). Keep the
existing null check for estimatedSendRate, then use this sanitized bitrate in
Probe(bitrate).encode to avoid silent BigInt truncation downstream.
1 participant
Currently only for publishing, not actually used.