moq-native: WebSocket fallback with explicit enum #812
Conversation
We'll get a cleaner diff when we add another connection method to the struct in a future commit.
| default_value = "200ms", | ||
| value_parser = humantime::parse_duration, | ||
| )] | ||
| #[serde(with = "humantime_serde")] |
There was a problem hiding this comment.
I'd initially avoided this crate because it's not clear that it's still maintained. It hasn't been updated in about 4 years so I wasn't sure if people would look askance at taking that dependency. However it does save a few lines of code here and in general I think humantime has nice ergonomics.
rs/moq-clock/src/main.rs
Outdated
| match session { | ||
| moq_native::WebTransportSessionAny::Quinn(session) => { | ||
| let session = moq_lite::Session::connect(session, publish, subscribe).await?; | ||
| session.closed().await | ||
| } | ||
| moq_native::WebTransportSessionAny::WebSocket(session) => { | ||
| let session = moq_lite::Session::connect(session, publish, subscribe).await?; | ||
| session.closed().await | ||
| } | ||
| } |
There was a problem hiding this comment.
Here's where we need to explicitly handle the possible transport implementations.
jhiatt-verkada
force-pushed
the
websocket-fallback-with-explicit-enum
branch
2 times, most recently
from
8cad64a to
ddf53c2
Compare
If desired, use Client::connect_with_fallback() to use a WebSocket-based transport when the QUIC connection fails. The logic mirrors the TypeScript implementation by affording the QUIC codepath a small (configurable) headstart.
jhiatt-verkada
force-pushed
the
websocket-fallback-with-explicit-enum
branch
from
ddf53c2 to
0f07786
Compare
And remove <S> from Session so it's muuuch easier to use.
|
Heh I probably could have make this into a separate PR. I haven't tested this outside of running @jhiatt-verkada By removing |
Might as well with such a breaking API.
WalkthroughThe PR overhauls connection/session flows and client/server APIs across crates. Clients gain new Pre-merge checks and finishing touches❌ Failed checks (1 warning)
✅ Passed checks (2 passed)
✨ Finishing touches
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 (2)
rs/moq-native/src/client.rs (2)
11-12: Consider bounded cache or TTL forWEBSOCKET_WON.The
WEBSOCKET_WONset grows unboundedly as the client connects to different servers. For long-running applications connecting to many unique servers, this could accumulate memory indefinitely. Consider adding a TTL or using an LRU cache with a bounded size.
304-310: Consider handling poisoned mutex gracefully.Using
.unwrap()onMutex::lock()will panic if the mutex is poisoned (e.g., a thread panicked while holding it). While rare, this could crash the application. Consider usinglock().unwrap_or_else(|e| e.into_inner())to recover from poisoned state, or propagate as an error.🔎 Proposed fix
- match self.websocket_delay { - Some(delay) if !WEBSOCKET_WON.lock().unwrap().contains(&key) => { + match self.websocket_delay { + Some(delay) if !WEBSOCKET_WON.lock().unwrap_or_else(|e| e.into_inner()).contains(&key) => {
📜 Review details
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
⛔ Files ignored due to path filters (1)
Cargo.lockis excluded by!**/*.lock
📒 Files selected for processing (12)
rs/hang-cli/src/client.rsrs/hang-cli/src/server.rsrs/hang/examples/video.rsrs/libmoq/src/session.rsrs/moq-clock/src/main.rsrs/moq-lite/src/session.rsrs/moq-native/Cargo.tomlrs/moq-native/examples/chat.rsrs/moq-native/src/client.rsrs/moq-native/src/server.rsrs/moq-relay/src/cluster.rsrs/moq-relay/src/connection.rs
🧰 Additional context used
📓 Path-based instructions (2)
**/*.rs
📄 CodeRabbit inference engine (CLAUDE.md)
In Rust crates, tests should be integrated within source files using inline test modules
Files:
rs/moq-relay/src/connection.rsrs/moq-relay/src/cluster.rsrs/moq-native/examples/chat.rsrs/hang-cli/src/server.rsrs/moq-clock/src/main.rsrs/moq-lite/src/session.rsrs/hang-cli/src/client.rsrs/hang/examples/video.rsrs/moq-native/src/client.rsrs/moq-native/src/server.rsrs/libmoq/src/session.rs
rs/**/Cargo.toml
📄 CodeRabbit inference engine (CLAUDE.md)
For Rust development, use the workspace configuration in
rs/Cargo.toml
Files:
rs/moq-native/Cargo.toml
🧠 Learnings (3)
📚 Learning: 2025-12-20T13:14:11.182Z
Learnt from: Frando
Repo: moq-dev/moq PR: 794
File: rs/moq-native/src/iroh/client.rs:40-42
Timestamp: 2025-12-20T13:14:11.182Z
Learning: In rs/moq-native/src/iroh/server.rs, the Server::close method uses `&mut self` instead of `&self` (unlike Client::close) because the Server struct contains `FuturesUnordered<BoxFuture<...>>` which is Send but not Sync, and using `&self` would require Self: Sync. This is documented in the code and is an acceptable trade-off.
Applied to files:
rs/moq-relay/src/connection.rsrs/moq-native/examples/chat.rsrs/hang-cli/src/server.rsrs/moq-lite/src/session.rsrs/moq-native/src/server.rs
📚 Learning: 2025-12-20T13:22:14.684Z
Learnt from: Frando
Repo: moq-dev/moq PR: 794
File: justfile:150-156
Timestamp: 2025-12-20T13:22:14.684Z
Learning: In moq-relay, WebTransport connections automatically prepend the URL path (e.g., `/anon`) as a prefix to all broadcasts. However, raw QUIC and iroh connections have no HTTP layer, so there's no way to send a path. To match the expected broadcast path for the web UI, the prefix must be manually added to the broadcast name (e.g., `anon/{{name}}`) when publishing via iroh or raw QUIC.
Applied to files:
rs/moq-native/examples/chat.rsrs/hang-cli/src/server.rs
📚 Learning: 2025-12-10T04:00:14.871Z
Learnt from: CR
Repo: moq-dev/moq PR: 0
File: CLAUDE.md:0-0
Timestamp: 2025-12-10T04:00:14.871Z
Learning: Applies to rs/**/Cargo.toml : For Rust development, use the workspace configuration in `rs/Cargo.toml`
Applied to files:
rs/moq-native/Cargo.toml
🧬 Code graph analysis (6)
rs/moq-relay/src/connection.rs (1)
rs/hang/src/model/broadcast.rs (1)
subscribe(82-84)
rs/moq-native/examples/chat.rs (1)
rs/hang-cli/src/client.rs (1)
client(6-31)
rs/moq-lite/src/session.rs (1)
rs/libmoq/src/session.rs (2)
connect(15-40)close(60-63)
rs/hang/examples/video.rs (1)
rs/hang-cli/src/client.rs (1)
client(6-31)
rs/moq-native/src/client.rs (1)
rs/moq-lite/src/session.rs (2)
new(25-29)connect(35-90)
rs/moq-native/src/server.rs (1)
rs/hang-cli/src/server.rs (1)
accept(43-68)
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (1)
- GitHub Check: Check
🔇 Additional comments (21)
rs/hang/examples/video.rs (1)
34-37: LGTM!The simplified single-step connection flow is clean and consistent with the broader API refactoring. The comment helpfully mentions
connect_with_fallbackas an alternative for WebSocket support.rs/moq-native/src/server.rs (2)
248-256: Good rename fromclosetoreject.The new name better conveys the semantic intent—rejecting a request with an HTTP status code rather than closing an established connection.
258-270: Clean consolidation of transport and MoQ handshakes.The
acceptmethod correctly handles both transport types:
- WebTransport requires async
.ok().await- QUIC is synchronous
.ok()Then delegates to
moq_lite::Session::acceptfor the MoQ-level handshake. This aligns well with the client-sideconnectconsolidation.rs/moq-lite/src/session.rs (3)
8-10: Good use of trait objects to simplify the public API.Erasing the generic via
Arc<dyn SessionInner>allowsSessionto be returned uniformly from different transport backends (QUIC, WebSocket) without exposing generics to callers. This enables theconnect_with_fallbackpattern where the transport type is determined at runtime.Also applies to: 24-29
167-170: Verify:closed()now always returnsErr.The method now unconditionally wraps the underlying error in
Error::Transportand returns it as an error. Previously, clean session closures might have been distinguishable from error cases. Confirm this is the intended behavior—the TODO comment on line 166 suggests this area is pending refinement.
173-186: SessionInner trait implementation is sound.The trait is correctly marked
Send + Syncfor use withArc<dyn SessionInner>. The blanket implementation forS: web_transport_trait::Sessionproperly delegates to the underlying session methods.One observation: the
closed()method boxes both the future and the error. This adds slight overhead but is necessary for type erasure. Given that session closure is not a hot path, this is an acceptable trade-off.rs/moq-native/examples/chat.rs (1)
33-36: LGTM!The simplified connection flow is consistent with the
video.rsexample and the broader API refactoring. The comments clearly explain the options available.rs/hang-cli/src/server.rs (1)
81-82: LGTM!The server-side flow now mirrors the client-side simplification. The
Request::acceptmethod handles both the transport-level handshake and MoQ session establishment in a single call.rs/moq-relay/src/cluster.rs (1)
270-274: LGTM!The cluster connection now uses the consolidated
client.connect()API. The error context appropriately reflects that this is a connection failure to a remote node.rs/hang-cli/src/client.rs (1)
13-16: LGTM!The transition to
connect_with_fallbackis clean and correct. The logging placement before the connection attempt is appropriate for debugging, and passingorigin.consumerfor publishing aligns with the updated API semantics.rs/moq-relay/src/connection.rs (2)
32-34: LGTM!The rename from
closetorejectis semantically clearer for the error case.
53-60: LGTM!The unified
accept()API simplifies the flow by encapsulating session creation. The comment explaining the reversed subscribe/publish semantics for relays is helpful.rs/libmoq/src/session.rs (1)
50-53: LGTM!The transition to the unified
client.connect(url, publish, consume)API simplifies the flow and maintains proper error handling withArc-wrapped errors.rs/moq-clock/src/main.rs (3)
56-57: LGTM!Creating the origin before the match is clean, as only one branch will execute and consume its parts.
66-71: LGTM!The
connect_with_fallbackusage for publishing is correct:Some(origin.consumer)for publishing tracks,Nonefor subscribing. The.context("session closed")provides clearer error messages than the previousmap_err.
76-76: LGTM!Correct API usage for subscribing:
Nonefor publish,origin.producerfor consume.rs/moq-native/src/client.rs (5)
38-53: LGTM!The
ClientWebSocketconfiguration struct is well-designed with sensible defaults (200ms delay) and proper CLI/serde integration usinghumantimefor duration parsing.
174-184: LGTM!The unified
connect()method cleanly combines QUIC connection and MoQ handshake into a single call.
217-223: Race logic handles failure cases correctly.The
tokio::select!with pattern matching works as intended: when a branch's future completes but its pattern doesn't match (returnsNone), that branch is disabled andselect!waits for the remaining branch. Theelseonly triggers when all branches complete without matching.
340-348: LGTM with note on state update.The WebSocket session creation and
WEBSOCKET_WONupdate logic is correct. The warning log level appropriately signals fallback usage for debugging.
327-338: No action needed. The third parameter toconnect_async_with_configis correctly identified asdisable_naglein tokio-tungstenite 0.24. The code comment is accurate.
| humantime = "2.3" | ||
| humantime-serde = "1.1" |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
# Check the workspace configuration
head -30 rs/Cargo.tomlRepository: moq-dev/moq
Length of output: 127
🏁 Script executed:
# Read the full rs/moq-native/Cargo.toml to verify the dependencies
cat -n rs/moq-native/Cargo.tomlRepository: moq-dev/moq
Length of output: 2209
🏁 Script executed:
# Check if web-transport-ws is defined in workspace dependencies
grep -A 20 "\[workspace.dependencies\]" rs/Cargo.tomlRepository: moq-dev/moq
Length of output: 101
🏁 Script executed:
# Find all Cargo.toml files in the repository
find . -name "Cargo.toml" -type fRepository: moq-dev/moq
Length of output: 296
🏁 Script executed:
# Check if there's a root Cargo.toml with workspace definition
git ls-files | grep -i cargoRepository: moq-dev/moq
Length of output: 306
🏁 Script executed:
# Check the root Cargo.toml for workspace dependencies
cat -n Cargo.tomlRepository: moq-dev/moq
Length of output: 1003
🌐 Web query:
tokio-tungstenite latest version crate
💡 Result:
The latest released version is tokio-tungstenite 0.28.0. [1][2]
Sources:
- docs.rs documentation for tokio-tungstenite (0.28.0). [1]
- tokio-tungstenite GitHub repository / project pages. [2]
🌐 Web query:
humantime crate version 2.3 latest
💡 Result:
Yes — the latest published humantime release is 2.3.0 (current as of Jan 3, 2026). [1][2]
Sources:
[1] humantime 2.3.0 documentation (docs.rs).
[2] Fedora package listing showing humantime 2.3.0.
🌐 Web query:
humantime-serde crate version 1.1 latest
💡 Result:
The latest published version of the humantime-serde crate is 1.1.1. [1][2]
Sources:
- docs.rs listing for humantime-serde (v1.1.1). [1]
- Fedora/crates packaging information showing humantime-serde 1.1.1. [2]
Update tokio-tungstenite to the latest version.
While humantime (2.3.0) and web-transport-ws (workspace reference) are correctly pinned, tokio-tungstenite = "0.24" is outdated. The latest version is 0.28.0. Update to tokio-tungstenite = "0.28" to align with current releases and security updates.
humantime-serde = "1.1" could also be more specific (latest is 1.1.1), though the semantic versioning allows the current specification.
🤖 Prompt for AI Agents
In rs/moq-native/Cargo.toml around lines 26 to 27, update the outdated
dependency version: change tokio-tungstenite from "0.24" to "0.28" to use the
latest release (0.28.0), and optionally pin humantime-serde to "1.1.1" for
specificity; adjust the version strings in the Cargo.toml dependency entries
accordingly and run cargo update to refresh the lockfile.
|
@jhiatt-verkada Let me know if this doesn't work for you, thanks for the PR! |
There was a problem hiding this comment.
Actionable comments posted: 3
🧹 Nitpick comments (2)
rs/moq-native/src/client.rs (2)
304-311: Address TODO: Use if-let chains for cleaner syntax.The TODO comment suggests using if-let chains (stabilized in Rust 1.64) to simplify this match expression.
🔎 Suggested refactor using if-let chains
- // TODO if let chain - match self.websocket_delay { - Some(delay) if !WEBSOCKET_WON.lock().unwrap().contains(&key) => { - tokio::time::sleep(delay).await; - tracing::debug!(%url, delay_ms = %delay.as_millis(), "QUIC not yet connected, attempting WebSocket fallback"); - } - _ => {} - } + if let Some(delay) = self.websocket_delay + && !WEBSOCKET_WON.lock().unwrap().contains(&key) + { + tokio::time::sleep(delay).await; + tracing::debug!(%url, delay_ms = %delay.as_millis(), "QUIC not yet connected, attempting WebSocket fallback"); + }
328-340: Consider making WebSocket config values configurable.The WebSocket message size (64 MB) and frame size (16 MB) limits are hardcoded. Consider making these configurable through
ClientWebSocketif different limits are needed in the future.🔎 Example: Add config fields
Add to
ClientWebSocket:#[serde(skip_serializing_if = "Option::is_none")] pub max_message_size: Option<usize>, #[serde(skip_serializing_if = "Option::is_none")] pub max_frame_size: Option<usize>,Then use
config.websocket.max_message_size.unwrap_or(64 << 20)in the connection code.
📜 Review details
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
⛔ Files ignored due to path filters (1)
Cargo.lockis excluded by!**/*.lock
📒 Files selected for processing (2)
rs/moq-native/Cargo.tomlrs/moq-native/src/client.rs
🚧 Files skipped from review as they are similar to previous changes (1)
- rs/moq-native/Cargo.toml
🧰 Additional context used
📓 Path-based instructions (1)
**/*.rs
📄 CodeRabbit inference engine (CLAUDE.md)
In Rust crates, tests should be integrated within source files using inline test modules
Files:
rs/moq-native/src/client.rs
🧠 Learnings (2)
📓 Common learnings
Learnt from: Frando
Repo: moq-dev/moq PR: 794
File: rs/moq-native/src/iroh/client.rs:40-42
Timestamp: 2025-12-20T13:14:11.182Z
Learning: In rs/moq-native/src/iroh/server.rs, the Server::close method uses `&mut self` instead of `&self` (unlike Client::close) because the Server struct contains `FuturesUnordered<BoxFuture<...>>` which is Send but not Sync, and using `&self` would require Self: Sync. This is documented in the code and is an acceptable trade-off.
Learnt from: CR
Repo: moq-dev/moq PR: 0
File: CLAUDE.md:0-0
Timestamp: 2025-12-10T04:00:14.871Z
Learning: Core protocol implementation in the `moq` layer must be generic and not contain media-specific logic; CDN/relay does not know anything about media
📚 Learning: 2025-12-20T13:14:11.182Z
Learnt from: Frando
Repo: moq-dev/moq PR: 794
File: rs/moq-native/src/iroh/client.rs:40-42
Timestamp: 2025-12-20T13:14:11.182Z
Learning: In rs/moq-native/src/iroh/server.rs, the Server::close method uses `&mut self` instead of `&self` (unlike Client::close) because the Server struct contains `FuturesUnordered<BoxFuture<...>>` which is Send but not Sync, and using `&self` would require Self: Sync. This is documented in the code and is an acceptable trade-off.
Applied to files:
rs/moq-native/src/client.rs
🧬 Code graph analysis (1)
rs/moq-native/src/client.rs (1)
rs/moq-lite/src/session.rs (2)
new(25-29)connect(35-90)
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (1)
- GitHub Check: Check
🔇 Additional comments (6)
rs/moq-native/src/client.rs (6)
39-54: LGTM: Clean configuration structure.The
ClientWebSocketconfig with optional delay is well-designed. The use ofhumantimefor duration parsing provides good ergonomics, and the optional nature allows flexibility in configuration.
72-82: LGTM: Configuration plumbing is correct.The WebSocket config is properly integrated into
ClientConfigand plumbed through to theClientstruct, following the same pattern as the existingtlsfield.Also applies to: 98-98, 167-173
175-185: LGTM: Clean two-phase connection.The
connectmethod properly separates transport establishment (connect_quic) from protocol handshake (moq_lite::Session::connect), providing a clean QUIC-only connection path.
187-225: LGTM: Well-implemented connection race.The fallback logic correctly races QUIC and WebSocket connections, handling per-transport failures gracefully while ensuring at least one succeeds. The use of
tokio::select!is appropriate here, and both session types correctly implement theweb_transport_trait::Sessiontrait required bymoq_lite::Session::connect.
227-291: LGTM: Return type now explicit.Making the return type of
connect_quicexplicit (web_transport_quinn::Session) improves code clarity without changing functionality.
313-325: LGTM: URL scheme conversion is correct.The scheme conversion logic properly handles all expected protocols (http, https, moql, moqt, ws, wss) and provides clear error messages for unsupported schemes.
| // Track servers (hostname:port) where WebSocket won the race, so we won't give QUIC a headstart next time | ||
| static WEBSOCKET_WON: LazyLock<Mutex<HashSet<(String, u16)>>> = LazyLock::new(|| Mutex::new(HashSet::new())); |
There was a problem hiding this comment.
Memory leak: WEBSOCKET_WON grows unbounded.
The global WEBSOCKET_WON set is never cleared and will accumulate entries for every unique server the client connects to over its lifetime. In long-running processes (such as relays or services), this constitutes a memory leak.
Consider adding a bounded cache with LRU eviction or a TTL mechanism to prevent unbounded growth.
Consider more granular tracking key.
The key (hostname, port) doesn't include the URL path. If the same server hosts multiple services on different paths with different transport characteristics, they'll all share the same WebSocket-won status. Consider whether path-level granularity is needed for your use case.
🤖 Prompt for AI Agents
In rs/moq-native/src/client.rs around lines 12-13, WEBSOCKET_WON is a global
unbounded HashSet that will grow indefinitely; replace it with a thread-safe
bounded cache (e.g., an LRU cache or a TTL-backed cache) so entries are evicted
when capacity or time limits are reached, and ensure the cache API is wrapped
behind the same Mutex or a concurrent map to preserve thread-safety; also change
the key from (String, u16) to include the request path (e.g., (String, u16,
String) or a normalized URL string) if you need per-path granularity, and add a
configurable capacity/TTL constant and a simple periodic or on-access eviction
strategy so long-running processes don’t leak memory.
|
|
||
| // Wrap WebSocket in WebTransport compatibility layer | ||
| // Similar to what the relay does: web_transport_ws::Session::new(socket, true) | ||
| let session = web_transport_ws::Session::new(ws_stream, false); |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Search for web_transport_ws::Session::new usage patterns in the codebase to verify parameter semantics
rg -nP -C3 'web_transport_ws::Session::new\(' --type rustRepository: moq-dev/moq
Length of output: 1108
Clarify the comment about is_server parameter difference between client and relay.
The false parameter is correct for the client (it should not be a server). However, the comment on line 342 is misleading—it references the relay's usage showing web_transport_ws::Session::new(socket, true) while the client uses false. This creates confusion about whether they're truly similar. Update the comment to clarify: the relay uses true because it accepts inbound connections (acts as server), while the client uses false because it initiates outbound connections (acts as client).
🤖 Prompt for AI Agents
In rs/moq-native/src/client.rs around line 343, the inline comment about the
is_server boolean is misleading because it references the relay using
web_transport_ws::Session::new(socket, true) without clarifying roles; update
the comment to explicitly state that the relay passes true because it accepts
inbound connections (acts as a server), whereas the client passes false here
because it initiates outbound connections (acts as a client), so keep the false
parameter and revise the comment text accordingly to avoid confusion.
| tracing::warn!(%url, "using WebSocket fallback"); | ||
| WEBSOCKET_WON.lock().unwrap().insert(key); |
There was a problem hiding this comment.
Use info! or debug! log level instead of warn!.
Line 345 logs "using WebSocket fallback" at the warn! level, but this is expected behavior when QUIC fails, not a warning condition. Consider using info! or debug! instead.
🔎 Suggested fix
- tracing::warn!(%url, "using WebSocket fallback");
+ tracing::info!(%url, "using WebSocket fallback");📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| tracing::warn!(%url, "using WebSocket fallback"); | |
| WEBSOCKET_WON.lock().unwrap().insert(key); | |
| tracing::info!(%url, "using WebSocket fallback"); | |
| WEBSOCKET_WON.lock().unwrap().insert(key); |
🤖 Prompt for AI Agents
In rs/moq-native/src/client.rs around lines 345 to 346, replace the
tracing::warn!(%url, "using WebSocket fallback"); call with a lower-severity log
(e.g., tracing::info!(%url, "using WebSocket fallback"); or
tracing::debug!(...)) so this expected fallback is not treated as a warning;
keep the subsequent WEBSOCKET_WON.lock().unwrap().insert(key); unchanged and
ensure the tracing macro you choose is in scope.
@kixelated I appreciate your work in getting this merged! I just did a build from |
2 participants
This is an alternate approach to #808 . In this version the return type of
connect_with_fallback()is an enum withQuinn+WebSocketvariants, and it's up to the caller to explicitly dispatch tomoq_liteover the different types.Advantages
Disadvantages
moq_native::Client::connect_with_fallback()is not a simple replacement formoq_native::Client::connect()as the resultingmoq_lite::Sessioninstances are distinct types