From c979d8c237753fae08086acd8c357e6ae385739a Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 30 Sep 2026 20:42:34 -0700 Subject: [PATCH 1/3] fix(quinn): preserve raw QUIC application close codes Co-Authored-By: GPT-6 --- rs/web-transport-quinn/src/error.rs | 90 ++++++++++++++++++++++++++- rs/web-transport-quinn/src/recv.rs | 14 +++-- rs/web-transport-quinn/src/send.rs | 18 +++--- rs/web-transport-quinn/src/session.rs | 42 ++++--------- rs/web-transport-quinn/tests/raw.rs | 44 +++++++++++++ 5 files changed, 165 insertions(+), 43 deletions(-) diff --git a/rs/web-transport-quinn/src/error.rs b/rs/web-transport-quinn/src/error.rs index 3ce3e80a..d968adb9 100644 --- a/rs/web-transport-quinn/src/error.rs +++ b/rs/web-transport-quinn/src/error.rs @@ -1,4 +1,4 @@ -use std::sync::Arc; +use std::sync::{Arc, OnceLock}; use thiserror::Error; @@ -67,6 +67,56 @@ impl From for SessionError { } } +/// The session's first close reason and its application-code space. +#[derive(Debug)] +pub(crate) struct CloseReason { + raw: bool, + reason: OnceLock, +} + +impl CloseReason { + pub(crate) fn new(raw: bool) -> Self { + Self { + raw, + reason: OnceLock::new(), + } + } + + pub(crate) fn get(&self) -> Option<&SessionError> { + self.reason.get() + } + + pub(crate) fn set(&self, err: SessionError) -> Result<(), SessionError> { + self.reason.set(err) + } + + pub(crate) fn map(&self, err: SessionError) -> SessionError { + let connection = match &err { + SessionError::ConnectionError(connection) + | SessionError::SendDatagramError(quinn::SendDatagramError::ConnectionLost( + connection, + )) => Some(connection), + SessionError::WebTransportError(WebTransportError::Closed(..)) => None, + _ => return err, + }; + if let Some(reason) = self.get() { + return reason.clone(); + } + if self.raw { + if let Some(quinn::ConnectionError::ApplicationClosed(close)) = connection { + if let Ok(code) = u32::try_from(close.error_code.into_inner()) { + return WebTransportError::Closed( + code, + String::from_utf8_lossy(&close.reason).into_owned(), + ) + .into(); + } + } + } + err + } +} + /// An error that can occur when reading/writing the WebTransport stream header. #[derive(Clone, Error, Debug)] pub enum WebTransportError { @@ -298,3 +348,41 @@ impl web_transport_trait::Error for ReadError { } } } + +#[cfg(test)] +mod tests { + use super::*; + use web_transport_trait::Error as _; + + fn peer_close(code: u64) -> SessionError { + quinn::ConnectionError::ApplicationClosed(quinn::ApplicationClose { + error_code: quinn::VarInt::from_u64(code).unwrap(), + reason: b"peer closed".as_slice().into(), + }) + .into() + } + + #[test] + fn raw_close_uses_the_application_code_space() { + let raw = CloseReason::new(true); + assert_eq!( + raw.map(peer_close(4075)).session_error(), + Some((4075, "peer closed".into())) + ); + assert_eq!( + raw.map(peer_close(u32::MAX as u64 + 1)).session_error(), + None + ); + } + + #[test] + fn http3_close_keeps_its_code_space() { + let h3 = CloseReason::new(false); + assert_eq!(h3.map(peer_close(4075)).session_error(), None); + assert_eq!( + h3.map(peer_close(web_transport_proto::error_to_http3(4075))) + .session_error(), + Some((4075, "peer closed".into())) + ); + } +} diff --git a/rs/web-transport-quinn/src/recv.rs b/rs/web-transport-quinn/src/recv.rs index d0b15bdd..a0dffccb 100644 --- a/rs/web-transport-quinn/src/recv.rs +++ b/rs/web-transport-quinn/src/recv.rs @@ -1,8 +1,9 @@ +use crate::error::CloseReason; use std::{ future::Future, io, pin::Pin, - sync::{Arc, OnceLock}, + sync::Arc, task::{ready, Context, Poll}, }; @@ -14,11 +15,11 @@ use crate::{ReadError, ReadExactError, ReadToEndError, SessionError}; #[derive(Debug)] pub struct RecvStream { inner: quinn::RecvStream, - error: Arc>, + error: Arc, } impl RecvStream { - pub(crate) fn new(stream: quinn::RecvStream, error: Arc>) -> Self { + pub(crate) fn new(stream: quinn::RecvStream, error: Arc) -> Self { Self { inner: stream, error, @@ -33,7 +34,10 @@ impl RecvStream { return ReadError::SessionError(err.clone()); } } - e + match e { + ReadError::SessionError(err) => ReadError::SessionError(self.error.map(err)), + err => err, + } } /// Tell the other end to stop sending data with the given error code. See [`quinn::RecvStream::stop`]. @@ -98,7 +102,7 @@ impl RecvStream { Ok(None) => Ok(None), Ok(Some(code)) => Ok(web_transport_proto::error_from_http3(code.into_inner())), Err(quinn::ResetError::ConnectionLost(conn_err)) => { - Err(self.error.get().cloned().unwrap_or_else(|| conn_err.into())) + Err(self.error.map(conn_err.into())) } Err(quinn::ResetError::ZeroRttRejected) => unreachable!("0-RTT not supported"), } diff --git a/rs/web-transport-quinn/src/send.rs b/rs/web-transport-quinn/src/send.rs index 0d02767a..3db8395e 100644 --- a/rs/web-transport-quinn/src/send.rs +++ b/rs/web-transport-quinn/src/send.rs @@ -1,7 +1,8 @@ +use crate::error::CloseReason; use std::{ io, pin::Pin, - sync::{Arc, OnceLock}, + sync::Arc, task::{ready, Context, Poll}, }; @@ -16,14 +17,14 @@ use crate::{ClosedStream, SessionError, WriteError}; #[derive(Debug)] pub struct SendStream { stream: quinn::SendStream, - error: Arc>, + error: Arc, // Retains the `stopped()` future across `poll_closed` calls. closed: crate::op::Op>, } impl SendStream { - pub(crate) fn new(stream: quinn::SendStream, error: Arc>) -> Self { + pub(crate) fn new(stream: quinn::SendStream, error: Arc) -> Self { Self { stream, error, @@ -42,7 +43,10 @@ impl SendStream { return WriteError::SessionError(err.clone()); } } - e + match e { + WriteError::SessionError(err) => WriteError::SessionError(self.error.map(err)), + err => err, + } } /// Abruptly reset the stream with the provided error code. See [`quinn::SendStream::reset`]. @@ -62,7 +66,7 @@ impl SendStream { Ok(Some(code)) => Ok(web_transport_proto::error_from_http3(code.into_inner())), Ok(None) => Ok(None), Err(quinn::StoppedError::ConnectionLost(conn_err)) => { - Err(self.error.get().cloned().unwrap_or_else(|| conn_err.into())) + Err(self.error.map(conn_err.into())) } Err(quinn::StoppedError::ZeroRttRejected) => unreachable!("0-RTT not supported"), } @@ -241,9 +245,7 @@ impl web_transport_trait::poll::SendStream for SendStream { }, Ok(None) => Ok(()), Err(quinn::StoppedError::ConnectionLost(conn_err)) => { - Err(WriteError::SessionError( - error.get().cloned().unwrap_or_else(|| conn_err.into()), - )) + Err(WriteError::SessionError(error.map(conn_err.into()))) } Err(quinn::StoppedError::ZeroRttRejected) => unreachable!("0-RTT not supported"), } diff --git a/rs/web-transport-quinn/src/session.rs b/rs/web-transport-quinn/src/session.rs index a9673e97..80a6a6a9 100644 --- a/rs/web-transport-quinn/src/session.rs +++ b/rs/web-transport-quinn/src/session.rs @@ -1,3 +1,4 @@ +use crate::error::CloseReason; use std::{ fmt, future::Future, @@ -82,7 +83,7 @@ pub struct Session { // Session error, set once by either local close() or the background task // when a remote CloseWebTransportSession capsule is received. // Uses OnceLock for set-once, first-writer-wins semantics with lock-free reads. - error: Arc>, + error: Arc, // The request sent by the client, or None for a raw QUIC session. request: Option, @@ -141,7 +142,7 @@ impl Session { let mut header_datagram = Vec::new(); session_id.encode(&mut header_datagram); - let error: Arc> = Arc::new(OnceLock::new()); + let error = Arc::new(CloseReason::new(false)); // Accept logic is stateful, so use an Arc to share it. let accept = SessionAccept::new(conn.clone(), session_id, error.clone()); @@ -179,11 +180,7 @@ impl Session { // Read capsules from the CONNECT recv stream until it's closed, // then record the close error and tear down the connection. - async fn run_recv( - conn: quinn::Connection, - recv: quinn::RecvStream, - error: Arc>, - ) { + async fn run_recv(conn: quinn::Connection, recv: quinn::RecvStream, error: Arc) { let close_info = Self::read_capsules(recv).await; let code = close_info.as_ref().map_or(0, |(c, _)| *c); @@ -508,27 +505,14 @@ impl Session { } // The owned-argument form, for futures that outlive the borrow that created them. - fn map_error_owned(error: &OnceLock, e: impl Into) -> SessionError { + fn map_error_owned(error: &CloseReason, e: impl Into) -> SessionError { map_error_with(error, e) } } /// Replace connection-level errors with the stored session error if available. -fn map_error_with(stored: &OnceLock, e: impl Into) -> SessionError { - { - let e = e.into(); - if let Some(err) = stored.get() { - if matches!( - &e, - SessionError::ConnectionError(_) - | SessionError::WebTransportError(WebTransportError::Closed(..)) - | SessionError::SendDatagramError(quinn::SendDatagramError::ConnectionLost(_)) - ) { - return err.clone(); - } - } - e - } +fn map_error_with(stored: &CloseReason, e: impl Into) -> SessionError { + stored.map(e.into()) } impl Session { @@ -568,7 +552,7 @@ impl Session { parked_accept_bi: Default::default(), settings: None, connect_send: Arc::new(Mutex::new(None)), - error: Arc::new(OnceLock::new()), + error: Arc::new(CloseReason::new(true)), request: None, response: None, alpn: Default::default(), @@ -714,7 +698,7 @@ pub struct SessionAccept { session_id: VarInt, // Shared session error for propagation to accepted streams. - error: Arc>, + error: Arc, // We also need to keep a reference to the qpack streams if the endpoint (incorrectly) creates them. // Again, this is just so they don't get closed until we drop the session. @@ -747,7 +731,7 @@ impl SessionAccept { pub(crate) fn new( conn: quinn::Connection, session_id: VarInt, - error: Arc>, + error: Arc, ) -> Self { // Create a stream that just outputs new streams, so it's easy to call from poll. let accept_uni = Box::pin(futures::stream::unfold(conn.clone(), |conn| async { @@ -1079,7 +1063,7 @@ impl Session { async fn open_uni_owned( conn: quinn::Connection, header: Bytes, - error: Arc>, + error: Arc, ) -> Result { let mut send = conn .open_uni() @@ -1100,7 +1084,7 @@ impl Session { async fn open_bi_owned( conn: quinn::Connection, header: Bytes, - error: Arc>, + error: Arc, ) -> Result<(SendStream, RecvStream), SessionError> { let (mut send, recv) = conn .open_bi() @@ -1162,7 +1146,7 @@ impl Session { async fn read_datagram_owned( conn: quinn::Connection, session_id: Option, - error: Arc>, + error: Arc, ) -> Result { let mut datagram = conn .read_datagram() diff --git a/rs/web-transport-quinn/tests/raw.rs b/rs/web-transport-quinn/tests/raw.rs index c49a8d67..a4b9b7e3 100644 --- a/rs/web-transport-quinn/tests/raw.rs +++ b/rs/web-transport-quinn/tests/raw.rs @@ -235,3 +235,47 @@ async fn raw_session_streams_omit_webtransport_header() -> Result<()> { Ok(()) } + +#[tokio::test] +async fn raw_peer_close_code_is_reported_by_session_and_streams() -> Result<()> { + use web_transport_trait::Error as _; + let (client, peer) = connect_raw().await?; + let session = Session::raw(client); + let mut send = session.open_uni().await?; + send.write_all(b"hello").await?; + let mut peer_recv = peer.accept_uni().await?; + let mut hello = [0; 5]; + peer_recv.read_exact(&mut hello).await?; + let mut peer_send = peer.open_uni().await?; + peer_send.write_all(b"hello").await?; + let mut recv = session.accept_uni().await?; + recv.read_exact(&mut hello).await?; + peer.close(4075u32.into(), b"application close"); + let closed = session.closed().await; + assert_eq!( + closed.session_error(), + Some((4075, "application close".into())) + ); + assert_eq!( + session.close_reason().unwrap().session_error(), + closed.session_error() + ); + assert_eq!( + send.write_all(b"after close") + .await + .unwrap_err() + .session_error(), + closed.session_error() + ); + assert_eq!( + recv.read(&mut hello).await.unwrap_err().session_error(), + closed.session_error() + ); + let stopped = futures::future::poll_fn(|cx| { + web_transport_trait::poll::SendStream::poll_closed(&mut send, cx) + }) + .await + .unwrap_err(); + assert_eq!(stopped.session_error(), closed.session_error()); + Ok(()) +} From 9870c44c521f758c54d005021b4661f7482243aa Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 1 Oct 2026 16:43:01 -0700 Subject: [PATCH 2/3] fix(quinn): latch raw peer closes before a local close Co-Authored-By: Claude Opus 5.5 --- rs/web-transport-quinn/src/error.rs | 21 ++++++++++++++++----- rs/web-transport-quinn/src/session.rs | 5 ++++- rs/web-transport-quinn/tests/raw.rs | 16 ++++++++++++++++ 3 files changed, 36 insertions(+), 6 deletions(-) diff --git a/rs/web-transport-quinn/src/error.rs b/rs/web-transport-quinn/src/error.rs index d968adb9..534047da 100644 --- a/rs/web-transport-quinn/src/error.rs +++ b/rs/web-transport-quinn/src/error.rs @@ -105,11 +105,17 @@ impl CloseReason { if self.raw { if let Some(quinn::ConnectionError::ApplicationClosed(close)) = connection { if let Ok(code) = u32::try_from(close.error_code.into_inner()) { - return WebTransportError::Closed( - code, - String::from_utf8_lossy(&close.reason).into_owned(), - ) - .into(); + // Raw sessions have no capsule task, so latch the peer close here. + return self + .reason + .get_or_init(|| { + WebTransportError::Closed( + code, + String::from_utf8_lossy(&close.reason).into_owned(), + ) + .into() + }) + .clone(); } } } @@ -369,6 +375,11 @@ mod tests { raw.map(peer_close(4075)).session_error(), Some((4075, "peer closed".into())) ); + assert!(raw + .set(quinn::ConnectionError::LocallyClosed.into()) + .is_err()); + + let raw = CloseReason::new(true); assert_eq!( raw.map(peer_close(u32::MAX as u64 + 1)).session_error(), None diff --git a/rs/web-transport-quinn/src/session.rs b/rs/web-transport-quinn/src/session.rs index 80a6a6a9..71f721b2 100644 --- a/rs/web-transport-quinn/src/session.rs +++ b/rs/web-transport-quinn/src/session.rs @@ -396,7 +396,10 @@ impl Session { pub fn close(&self, code: u32, reason: &[u8]) { // Record the local close error. First writer wins — if the background // task already set a remote close error, or close() was already called, - // this is a no-op. + // this is a no-op. A raw peer close that already arrived is latched first. + if let Some(err) = self.conn.close_reason() { + self.map_error(err); + } let err = SessionError::ConnectionError(quinn::ConnectionError::LocallyClosed); if self.error.set(err).is_err() { return; diff --git a/rs/web-transport-quinn/tests/raw.rs b/rs/web-transport-quinn/tests/raw.rs index a4b9b7e3..c267d9c2 100644 --- a/rs/web-transport-quinn/tests/raw.rs +++ b/rs/web-transport-quinn/tests/raw.rs @@ -279,3 +279,19 @@ async fn raw_peer_close_code_is_reported_by_session_and_streams() -> Result<()> assert_eq!(stopped.session_error(), closed.session_error()); Ok(()) } + +#[tokio::test] +async fn raw_peer_close_survives_local_close() -> Result<()> { + use web_transport_trait::Error as _; + let (client, peer) = connect_raw().await?; + let session = Session::raw(client.clone()); + peer.close(4075u32.into(), b"application close"); + // Wait on the quinn connection so the session never observes the close first. + client.closed().await; + session.close(1, b"cleanup"); + assert_eq!( + session.close_reason().unwrap().session_error(), + Some((4075, "application close".into())) + ); + Ok(()) +} From e2541d62efbe02cf75ecfcc7e64d878e289e1ab5 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Thu, 1 Oct 2026 16:56:33 -0700 Subject: [PATCH 3/3] fix(quinn): decode raw closes from the connection, not the HTTP/3 mapping Raw application codes in the WebTransport HTTP/3 range were reported as small WebTransport codes. The raw close handle now reads the connection's own close reason, so every code above u32::MAX is refused. Co-Authored-By: Claude Opus 5.5 --- rs/web-transport-quinn/src/error.rs | 86 ++++++++++++--------------- rs/web-transport-quinn/src/session.rs | 4 +- rs/web-transport-quinn/tests/raw.rs | 17 ++++++ 3 files changed, 58 insertions(+), 49 deletions(-) diff --git a/rs/web-transport-quinn/src/error.rs b/rs/web-transport-quinn/src/error.rs index 534047da..d2fa8281 100644 --- a/rs/web-transport-quinn/src/error.rs +++ b/rs/web-transport-quinn/src/error.rs @@ -70,14 +70,22 @@ impl From for SessionError { /// The session's first close reason and its application-code space. #[derive(Debug)] pub(crate) struct CloseReason { - raw: bool, + // The raw QUIC connection, whose application codes bypass the HTTP/3 mapping. + raw: Option, reason: OnceLock, } impl CloseReason { - pub(crate) fn new(raw: bool) -> Self { + pub(crate) fn http3() -> Self { Self { - raw, + raw: None, + reason: OnceLock::new(), + } + } + + pub(crate) fn raw(conn: quinn::Connection) -> Self { + Self { + raw: Some(conn), reason: OnceLock::new(), } } @@ -91,35 +99,37 @@ impl CloseReason { } pub(crate) fn map(&self, err: SessionError) -> SessionError { - let connection = match &err { - SessionError::ConnectionError(connection) - | SessionError::SendDatagramError(quinn::SendDatagramError::ConnectionLost( - connection, - )) => Some(connection), - SessionError::WebTransportError(WebTransportError::Closed(..)) => None, - _ => return err, - }; + if !matches!( + &err, + SessionError::ConnectionError(_) + | SessionError::WebTransportError(WebTransportError::Closed(..)) + | SessionError::SendDatagramError(quinn::SendDatagramError::ConnectionLost(_)) + ) { + return err; + } if let Some(reason) = self.get() { return reason.clone(); } - if self.raw { - if let Some(quinn::ConnectionError::ApplicationClosed(close)) = connection { - if let Ok(code) = u32::try_from(close.error_code.into_inner()) { - // Raw sessions have no capsule task, so latch the peer close here. - return self - .reason - .get_or_init(|| { - WebTransportError::Closed( - code, - String::from_utf8_lossy(&close.reason).into_owned(), - ) - .into() - }) - .clone(); + + // `err` was already decoded as HTTP/3, so a raw session decodes the connection's own reason. + // Raw sessions have no capsule task, so the peer close is latched here. + let Some(quinn::ConnectionError::ApplicationClosed(close)) = + self.raw.as_ref().and_then(|conn| conn.close_reason()) + else { + return err; + }; + self.reason + .get_or_init(|| match u32::try_from(close.error_code.into_inner()) { + Ok(code) => WebTransportError::Closed( + code, + String::from_utf8_lossy(&close.reason).into_owned(), + ) + .into(), + Err(_) => { + SessionError::ConnectionError(quinn::ConnectionError::ApplicationClosed(close)) } - } - } - err + }) + .clone() } } @@ -368,27 +378,9 @@ mod tests { .into() } - #[test] - fn raw_close_uses_the_application_code_space() { - let raw = CloseReason::new(true); - assert_eq!( - raw.map(peer_close(4075)).session_error(), - Some((4075, "peer closed".into())) - ); - assert!(raw - .set(quinn::ConnectionError::LocallyClosed.into()) - .is_err()); - - let raw = CloseReason::new(true); - assert_eq!( - raw.map(peer_close(u32::MAX as u64 + 1)).session_error(), - None - ); - } - #[test] fn http3_close_keeps_its_code_space() { - let h3 = CloseReason::new(false); + let h3 = CloseReason::http3(); assert_eq!(h3.map(peer_close(4075)).session_error(), None); assert_eq!( h3.map(peer_close(web_transport_proto::error_to_http3(4075))) diff --git a/rs/web-transport-quinn/src/session.rs b/rs/web-transport-quinn/src/session.rs index 71f721b2..07dea48d 100644 --- a/rs/web-transport-quinn/src/session.rs +++ b/rs/web-transport-quinn/src/session.rs @@ -142,7 +142,7 @@ impl Session { let mut header_datagram = Vec::new(); session_id.encode(&mut header_datagram); - let error = Arc::new(CloseReason::new(false)); + let error = Arc::new(CloseReason::http3()); // Accept logic is stateful, so use an Arc to share it. let accept = SessionAccept::new(conn.clone(), session_id, error.clone()); @@ -538,6 +538,7 @@ impl Session { /// (from `into_0rtt`, say) is fine, it just has no ALPN to report until it is done. pub fn raw(conn: quinn::Connection) -> Self { Self { + error: Arc::new(CloseReason::raw(conn.clone())), conn, session_id: None, header_uni: Default::default(), @@ -555,7 +556,6 @@ impl Session { parked_accept_bi: Default::default(), settings: None, connect_send: Arc::new(Mutex::new(None)), - error: Arc::new(CloseReason::new(true)), request: None, response: None, alpn: Default::default(), diff --git a/rs/web-transport-quinn/tests/raw.rs b/rs/web-transport-quinn/tests/raw.rs index c267d9c2..59b883a1 100644 --- a/rs/web-transport-quinn/tests/raw.rs +++ b/rs/web-transport-quinn/tests/raw.rs @@ -295,3 +295,20 @@ async fn raw_peer_close_survives_local_close() -> Result<()> { ); Ok(()) } + +#[tokio::test] +async fn raw_peer_close_above_u32_is_not_a_session_error() -> Result<()> { + use web_transport_trait::Error as _; + // Includes a code that would decode as WebTransport code 7 under the HTTP/3 mapping. + for code in [u32::MAX as u64 + 1, web_transport_proto::error_to_http3(7)] { + let (client, peer) = connect_raw().await?; + let session = Session::raw(client); + peer.close(quinn::VarInt::from_u64(code)?, b"too big"); + assert_eq!( + session.closed().await.session_error(), + None, + "code {code:#x}" + ); + } + Ok(()) +}