Fix close capsules#179
Conversation
…compatibility Write a CloseWebTransportSession capsule inside an HTTP/3 DATA frame on the CONNECT stream (RFC 9297), enabling browsers to receive close codes and reasons via the W3C WebTransport API's `WebTransport.closed` promise. Adds Http3CapsuleReader to handle capsules split across multiple DATA frames, and reworks Session::close() to deliver the capsule before closing the QUIC connection. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
When a peer sends a CloseWebTransportSession capsule, the background task stores the (code, reason) and calls conn.close(), causing all active Quinn operations to fail with LocallyClosed. Previously only Session::closed() checked the capsule—now every error path maps LocallyClosed to WebTransportError::Closed(code, reason). - Add CloseCapsule type alias and map_conn_error/map_session_error/ map_send_datagram_error helpers on Session and SessionAccept - Thread close_capsule into RecvStream and SendStream; add map_read_error/map_write_error helpers - Store local close info in close() so locally-initiated closes also surface WebTransportError::Closed consistently Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
Use OnceLock for set-once, first-writer-wins semantics between close() and the background capsule reader task. This eliminates the Mutex<Option<(u32, String)>> in favor of lock-free reads on every operation's error path. - close() uses error.set() as the sole guard — if it wins, it owns the capsule write and connection teardown; if the background task already set a remote close error, close() returns early - Background task symmetrically bails out if close() won the race - Consolidated three near-identical error mappers (map_conn_error, map_session_error, map_send_datagram_error) into a single map_error(impl Into<SessionError>) - closed() always awaits conn.closed() to ensure the capsule is delivered before returning, unlike quinn's early-return pattern - Removed connect_send reference from background task since the OnceLock guard makes it unnecessary Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
SessionAccept used a single unfold stream for accepting connections, which only stores one waker. When multiple tasks call accept_bi() or accept_uni() concurrently, only the last caller's waker is retained and earlier callers hang forever. Add per-method waker vecs (bi_wakers, uni_wakers) to SessionAccept. On Pending, push the caller's waker; on Ready, drain and wake all stored wakers so other callers retry. Replace the ready!() macro with explicit match to ensure the waker push isn't bypassed by its hidden early return on Poll::Pending. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
|
I was hesitant about error propagation myself, I think it's fine to not do it in the lib. My main reason for that was python bindings - because of how exceptions are usually handled in python generally and in similar libs (like websockets), it could be a bit more ergonomic to have a single exception with close code. But it's pretty straightforward to implement this propagation in python bindings just for python. |
I think it would be good in a follow-up PR. I've always meant to do it, because it's technically more correct, but it's surprisingly annoying. |
WalkthroughThis change reworks capsule and session handling across crates. It adds 🚥 Pre-merge checks | ✅ 2 | ❌ 1❌ Failed checks (1 inconclusive)
✅ 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: 3
🧹 Nitpick comments (1)
web-transport-proto/src/capsule.rs (1)
289-293: Avoid full-buffer allocation when skipping non-DATA frames.Current skip path allocates
lenbytes just to discard them. Streaming tosink()avoids extra allocations and is safer under large frame sizes.♻️ Suggested refactor
if frame_type != Frame::DATA { if len > 0 { - let mut skip = vec![0u8; len]; - self.stream.read_exact(&mut skip).await?; + let mut sink = tokio::io::sink(); + let mut take = (&mut self.stream).take(len as u64); + let copied = tokio::io::copy(&mut take, &mut sink).await?; + if copied != len as u64 { + return Err(CapsuleError::UnexpectedEnd); + } } continue; }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@web-transport-proto/src/capsule.rs` around lines 289 - 293, The code currently allocates a buffer `vec![0u8; len]` and calls `self.stream.read_exact(&mut skip).await?` when `frame_type != Frame::DATA`, which should be avoided for large skips; replace that allocation with streaming the bytes into a sink by using `self.stream.take(len as u64)` and `tokio::io::copy` (e.g. create a `let mut reader = self.stream.take(len as u64); tokio::io::copy(&mut reader, &mut tokio::io::sink()).await?`) so you discard the bytes without allocating, and ensure `tokio::io::copy` and `tokio::io::sink` are imported/available; keep the conditional on `Frame::DATA` unchanged.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@web-transport-proto/src/capsule.rs`:
- Around line 276-279: Replace the current blind handling of VarInt::read in
capsule.rs with a helper read_frame_type_or_eof that uses self.stream to attempt
reading the varint and returns Ok(None) only when EOF occurs before any byte is
read, Err(CapsuleError::UnexpectedEnd) when EOF happens mid-varint, and
Ok(Some(Frame(...))) on success; update the call site that currently does
VarInt::read(&mut self.stream).await -> Frame(...) / Err(_) => Ok(false) to call
read_frame_type_or_eof, convert Ok(Some(frame)) to the existing flow, Ok(None)
to Ok(false) (clean EOF), and propagate Err(CapsuleError::UnexpectedEnd) instead
of treating it as clean EOF.
In `@web-transport-quinn/src/session.rs`:
- Line 353: The code currently casts capsule_bytes.len() to u32 and calls
VarInt::from_u32(...).encode(&mut frame), which can silently truncate large
lengths; replace that with VarInt::try_from(capsule_bytes.len()) and handle the
Result by failing closed (returning an error) when the conversion fails, then
call .encode(&mut frame) on the successful VarInt; update the surrounding
function (the call site that builds the capsule frame) to propagate or map the
conversion error instead of allowing a silent truncation.
- Around line 311-325: The tokio::spawn call in close() can panic outside a
Tokio runtime; use tokio::runtime::Handle::try_current() to detect a runtime and
only spawn when available, otherwise perform a synchronous fallback: if
Handle::try_current() is Ok(handle) call handle.spawn(async move {
Self::close_with_capsule(conn, send, capsule, code, timeout).await }); else
perform an immediate close on the cloned conn (e.g., call conn.close(...) with
the code/reason and drop the send so the capsule won't be sent) so the code
paths avoid panicking; update the close() branch that creates capsule and uses
tokio::spawn to use this conditional logic referencing close(),
Self::close_with_capsule, tokio::spawn, and
tokio::runtime::Handle::try_current().
---
Nitpick comments:
In `@web-transport-proto/src/capsule.rs`:
- Around line 289-293: The code currently allocates a buffer `vec![0u8; len]`
and calls `self.stream.read_exact(&mut skip).await?` when `frame_type !=
Frame::DATA`, which should be avoided for large skips; replace that allocation
with streaming the bytes into a sink by using `self.stream.take(len as u64)` and
`tokio::io::copy` (e.g. create a `let mut reader = self.stream.take(len as u64);
tokio::io::copy(&mut reader, &mut tokio::io::sink()).await?`) so you discard the
bytes without allocating, and ensure `tokio::io::copy` and `tokio::io::sink` are
imported/available; keep the conditional on `Frame::DATA` unchanged.
ℹ️ Review info
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (5)
web-transport-proto/src/capsule.rsweb-transport-quinn/Cargo.tomlweb-transport-quinn/src/recv.rsweb-transport-quinn/src/send.rsweb-transport-quinn/src/session.rs
| if self.session_id.is_some() { | ||
| // Take the send stream for the capsule write. | ||
| let send = self.connect_send.lock().unwrap().take(); | ||
|
|
||
| if let Some(send) = send { | ||
| let reason = String::from_utf8_lossy(reason).into_owned(); | ||
| let conn = self.conn.clone(); | ||
| let capsule = | ||
| web_transport_proto::Capsule::CloseWebTransportSession { code, reason }; | ||
| let timeout = (self.rtt() * 3).max(Duration::from_millis(100)); | ||
|
|
||
| tokio::spawn(async move { | ||
| Self::close_with_capsule(conn, send, capsule, code, timeout).await; | ||
| }); | ||
| } |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
# First, let's look at the session.rs file structure and find the close method
git ls-files | grep -i sessionRepository: moq-dev/web-transport
Length of output: 197
🏁 Script executed:
# Read the session.rs file to understand the close method and its context
cat -n web-transport-quinn/src/session.rs | sed -n '300,330p'Repository: moq-dev/web-transport
Length of output: 1674
🏁 Script executed:
# Get a broader context around the close method
cat -n web-transport-quinn/src/session.rs | sed -n '280,350p'Repository: moq-dev/web-transport
Length of output: 3548
🏁 Script executed:
# Let's check how close() is called in the codebase
rg 'close\(' web-transport-quinn/src/ -B2 -A2 --type rsRepository: moq-dev/web-transport
Length of output: 91
🏁 Script executed:
# Let's check if there are any tests or examples showing how the Session is used
fd -e rs -path "*/tests/*" -o -path "*/examples/*" web-transport-quinn/Repository: moq-dev/web-transport
Length of output: 233
🏁 Script executed:
# Check the Cargo.toml for Tokio version and features
cat web-transport-quinn/Cargo.tomlRepository: moq-dev/web-transport
Length of output: 1356
🌐 Web query:
tokio::spawn panic no runtime context
💡 Result:
That panic means you called tokio::spawn(...) (or something that internally needs Tokio’s “reactor”, like tokio::time, networking, etc.) when no Tokio runtime is set as the current thread’s context. In that situation Tokio can’t find an executor/reactor and panics with messages like “there is no reactor running, must be called from the context of a Tokio 1.x runtime”. [2]
Fix options
1) Run the code inside a Tokio runtime (most common)
Use #[tokio::main] (or #[tokio::test]) and call tokio::spawn from there. [5]
#[tokio::main]
async fn main() {
tokio::spawn(async {
// ...
}).await.unwrap();
}2) If you manually build a Runtime, use rt.spawn or rt.enter()
Tokio’s docs explicitly note that Runtime::enter() “enters the runtime context” and “will also allow you to call methods such as tokio::spawn”. [4]
use tokio::runtime::Runtime;
fn main() {
let rt = Runtime::new().unwrap();
// Option A: spawn via the runtime handle
rt.spawn(async { /* ... */ });
// Option B: set thread-local runtime context so `tokio::spawn` works
let _guard = rt.enter();
tokio::spawn(async { /* ... */ });
rt.block_on(async { /* keep runtime alive / await joinhandles */ });
}3) If you’re in a framework, ensure it’s using Tokio (feature flags / runtime mismatch)
Some frameworks run futures on a non-Tokio executor by default; you must enable their Tokio integration (example: iced needs features = ["tokio"]). [6]
Common related pitfall
If you see spawn_local called from outside of a task::LocalSet, that’s different: spawn_local requires a LocalSet (still within a Tokio runtime). [3]
🏁 Script executed:
# Check where Session::close is actually called
rg 'close\(' web-transport-quinn/src/ -B2 -A2 | head -100Repository: moq-dev/web-transport
Length of output: 4851
🏁 Script executed:
# Look for public API surface and usages
rg 'Session' web-transport-quinn/src/lib.rs -B2 -A2Repository: moq-dev/web-transport
Length of output: 47
🏁 Script executed:
# Check examples or bin for usage patterns
find web-transport-quinn -name "*.rs" -type f | xargs grep -l "close()" | head -5Repository: moq-dev/web-transport
Length of output: 99
🏁 Script executed:
# Check lib.rs to understand the public API
cat web-transport-quinn/src/lib.rsRepository: moq-dev/web-transport
Length of output: 2456
🏁 Script executed:
# Look for examples
find web-transport-quinn -name "*.rs" -type f | head -20Repository: moq-dev/web-transport
Length of output: 594
🏁 Script executed:
# Check if there are any examples that show usage
ls -la web-transport-quinn/examples/ 2>/dev/null || echo "No examples directory"Repository: moq-dev/web-transport
Length of output: 509
🏁 Script executed:
# Check parent examples or tests
ls -la examples/ 2>/dev/null || echo "No examples at root"Repository: moq-dev/web-transport
Length of output: 84
🏁 Script executed:
# Look for test files that call close()
find . -name "*.rs" -type f -exec grep -l "\.close(" {} \; 2>/dev/null | head -10Repository: moq-dev/web-transport
Length of output: 449
🏁 Script executed:
# Let's check the echo client example to see how close() is called
cat web-transport-quinn/examples/echo-client.rsRepository: moq-dev/web-transport
Length of output: 3320
🏁 Script executed:
# Check the echo server example too
cat web-transport-quinn/examples/echo-server.rsRepository: moq-dev/web-transport
Length of output: 4348
🏁 Script executed:
# Let's look at the implementation again to see if there's any error handling or guards
cat -n web-transport-quinn/src/session.rs | sed -n '302,330p'Repository: moq-dev/web-transport
Length of output: 1492
🏁 Script executed:
# Check if Session implements any trait that documents runtime requirements
rg 'impl.*Session|pub trait' web-transport-quinn/src/session.rs -A5 | head -30Repository: moq-dev/web-transport
Length of output: 936
🏁 Script executed:
# Look at web_transport_trait to understand the trait definition
cat web-transport-trait/src/lib.rs 2>/dev/null || find . -path "*/web-transport-trait/*" -name "*.rs" | head -3Repository: moq-dev/web-transport
Length of output: 11229
🏁 Script executed:
# Check if Session is used in any Drop implementations or sync contexts
rg 'impl.*Drop.*for|drop\(' web-transport-quinn/src/ -B3 -A5Repository: moq-dev/web-transport
Length of output: 47
🏁 Script executed:
# Check if there are any tests that might call close from sync context
find . -name "*.rs" -type f -exec grep -l "\.close(" {} \; | xargs grep -B10 "\.close(" | grep -E "^--$|#\[test\]|fn test_" | head -20Repository: moq-dev/web-transport
Length of output: 118
🏁 Script executed:
# Look for any documentation or comments about runtime requirements
rg 'tokio|runtime|async' web-transport-quinn/src/session.rs | head -20Repository: moq-dev/web-transport
Length of output: 1316
🏁 Script executed:
# Let's check the Session Drop implementation to see if close() could be called on drop
rg 'impl.*Drop|fn drop\(' web-transport-quinn/src/session.rs -A10Repository: moq-dev/web-transport
Length of output: 47
🏁 Script executed:
# Let's check how Session is created - is it always in async context?
rg 'pub.*fn.*new|pub.*fn.*connect|pub.*fn.*accept' web-transport-quinn/src/ | grep -v "accept_" | head -15Repository: moq-dev/web-transport
Length of output: 1456
🏁 Script executed:
# Check if Session has a Drop implementation or if it's derived
rg 'impl Drop|#\[derive' web-transport-quinn/src/session.rs | head -10Repository: moq-dev/web-transport
Length of output: 81
🏁 Script executed:
# Let's also verify the trait signature for close
rg 'fn close' web-transport-trait/src/lib.rs -B2 -A2Repository: moq-dev/web-transport
Length of output: 832
🏁 Script executed:
# Let's check if there's any actual error handling around tokio::spawn currently
cat -n web-transport-quinn/src/session.rs | sed -n '315,330p'Repository: moq-dev/web-transport
Length of output: 836
Guard tokio::spawn in close() against missing runtime.
The synchronous close() method unconditionally spawns an async task at line 322. This will panic if called outside a Tokio runtime context. Since the method is public and the trait signature requires it to be synchronous, callers may invoke it from sync code without an active runtime.
Use tokio::runtime::Handle::try_current() to detect runtime availability and fall back to direct connection close:
Suggested fix
let timeout = (self.rtt() * 3).max(Duration::from_millis(100));
- tokio::spawn(async move {
- Self::close_with_capsule(conn, send, capsule, code, timeout).await;
- });
+ if let Ok(handle) = tokio::runtime::Handle::try_current() {
+ handle.spawn(async move {
+ Self::close_with_capsule(conn, send, capsule, code, timeout).await;
+ });
+ } else {
+ tracing::warn!("no tokio runtime available; closing QUIC connection directly");
+ let http3_code: quinn::VarInt = web_transport_proto::error_to_http3(code)
+ .try_into()
+ .unwrap();
+ conn.close(http3_code, reason.as_bytes());
+ }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@web-transport-quinn/src/session.rs` around lines 311 - 325, The tokio::spawn
call in close() can panic outside a Tokio runtime; use
tokio::runtime::Handle::try_current() to detect a runtime and only spawn when
available, otherwise perform a synchronous fallback: if Handle::try_current() is
Ok(handle) call handle.spawn(async move { Self::close_with_capsule(conn, send,
capsule, code, timeout).await }); else perform an immediate close on the cloned
conn (e.g., call conn.close(...) with the code/reason and drop the send so the
capsule won't be sent) so the code paths avoid panicking; update the close()
branch that creates capsule and uses tokio::spawn to use this conditional logic
referencing close(), Self::close_with_capsule, tokio::spawn, and
tokio::runtime::Handle::try_current().
There was a problem hiding this comment.
♻️ Duplicate comments (2)
web-transport-quinn/src/session.rs (1)
322-324:⚠️ Potential issue | 🟠 MajorGuard
tokio::spawnin syncclose()to avoid runtime-context panics.
Session::closeis synchronous, but Line 322 unconditionally callstokio::spawn. Ifclose()is invoked outside a Tokio runtime, this can panic and skip clean teardown.🔧 Proposed fix
- tokio::spawn(async move { - Self::close_with_capsule(conn, send, capsule, code, timeout).await; - }); + if let Ok(handle) = tokio::runtime::Handle::try_current() { + handle.spawn(async move { + Self::close_with_capsule(conn, send, capsule, code, timeout).await; + }); + } else { + tracing::warn!("no tokio runtime available; closing connection directly"); + let http3_code: quinn::VarInt = web_transport_proto::error_to_http3(code) + .try_into() + .unwrap(); + conn.close(http3_code, reason.as_bytes()); + }#!/bin/bash # Verify sync close() contains unguarded tokio::spawn and lacks runtime guard. rg -n "pub fn close\\(|tokio::spawn\\(|Handle::try_current" web-transport-quinn/src/session.rs -C3🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@web-transport-quinn/src/session.rs` around lines 322 - 324, Session::close currently calls tokio::spawn unconditionally which will panic when called outside a Tokio runtime; guard that call by checking for a current runtime with tokio::runtime::Handle::try_current() and branch: if a handle exists use handle.spawn (or tokio::spawn) to run Self::close_with_capsule(conn, send, capsule, code, timeout).await, otherwise run the async close on a newly created runtime or a std::thread (e.g. create a Runtime and block_on Self::close_with_capsule or spawn a thread that builds a Runtime and runs block_on) so the synchronous close() never relies on an existing Tokio context. Ensure you update the branch where tokio::spawn is invoked (the call to Self::close_with_capsule) to use this guarded approach.web-transport-proto/src/capsule.rs (1)
300-304:⚠️ Potential issue | 🟠 MajorNormalize truncated DATA-frame payload reads to
UnexpectedEnd.Line 303 uses
read_exact(...).await?, which maps short DATA payloads toCapsuleError::Io(UnexpectedEof)instead ofCapsuleError::UnexpectedEnd. This is inconsistent with the truncation handling at lines 280, 284, and 295, which all explicitly map toUnexpectedEnd.🔧 Proposed fix
if len > 0 { let start = self.buf.len(); self.buf.resize(start + len, 0); - self.stream.read_exact(&mut self.buf[start..]).await?; + self.stream + .read_exact(&mut self.buf[start..]) + .await + .map_err(|e| { + if e.kind() == std::io::ErrorKind::UnexpectedEof { + CapsuleError::UnexpectedEnd + } else { + CapsuleError::from(e) + } + })?; } return Ok(true);🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@web-transport-proto/src/capsule.rs` around lines 300 - 304, The read path currently uses self.stream.read_exact(&mut self.buf[start..]).await? which will convert a short read into CapsuleError::Io(UnexpectedEof), making it inconsistent with the other truncated DATA handling; change this read to explicitly map short/UnexpectedEof results to CapsuleError::UnexpectedEnd (same variant used at lines 280/284/295) — e.g. await the read and map Err(UnexpectedEof) (or inspect the io::Error kind) to return CapsuleError::UnexpectedEnd while preserving other IO errors as CapsuleError::Io, keeping the use of self.buf and the existing start/len logic.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Duplicate comments:
In `@web-transport-proto/src/capsule.rs`:
- Around line 300-304: The read path currently uses self.stream.read_exact(&mut
self.buf[start..]).await? which will convert a short read into
CapsuleError::Io(UnexpectedEof), making it inconsistent with the other truncated
DATA handling; change this read to explicitly map short/UnexpectedEof results to
CapsuleError::UnexpectedEnd (same variant used at lines 280/284/295) — e.g.
await the read and map Err(UnexpectedEof) (or inspect the io::Error kind) to
return CapsuleError::UnexpectedEnd while preserving other IO errors as
CapsuleError::Io, keeping the use of self.buf and the existing start/len logic.
In `@web-transport-quinn/src/session.rs`:
- Around line 322-324: Session::close currently calls tokio::spawn
unconditionally which will panic when called outside a Tokio runtime; guard that
call by checking for a current runtime with
tokio::runtime::Handle::try_current() and branch: if a handle exists use
handle.spawn (or tokio::spawn) to run Self::close_with_capsule(conn, send,
capsule, code, timeout).await, otherwise run the async close on a newly created
runtime or a std::thread (e.g. create a Runtime and block_on
Self::close_with_capsule or spawn a thread that builds a Runtime and runs
block_on) so the synchronous close() never relies on an existing Tokio context.
Ensure you update the branch where tokio::spawn is invoked (the call to
Self::close_with_capsule) to use this guarded approach.
ℹ️ Review info
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (3)
web-transport-proto/src/capsule.rsweb-transport-proto/src/varint.rsweb-transport-quinn/src/session.rs
|
Thanks @ai-and-i |
2 participants
Extracted from #176
The main thing I removed was propagating the close error to streams. It's fine to return locally closed, instead of the specific remote connection error. If it's important we can make a separate PR for it.