fix(web-transport-moq): send raw QUIC stream codes as is

web-transport-moq/src/error.rs

@@ -69,6 +69,9 @@ impl From<noq::ConnectionError> for SessionError {
 }
 
 /// Why a session closed, shared by the session and its streams. The first close wins.
+///
+/// It also knows whether the session is raw QUIC, which decides how its close and stream
+/// codes are encoded.
 #[derive(Debug)]
 pub(crate) struct CloseReason {
     // Raw QUIC carries the application code as is; HTTP/3 maps it into its own code space.
@@ -84,6 +87,26 @@ impl CloseReason {
         }
     }
 
+    /// The QUIC code to reset or stop a stream with.
+    pub(crate) fn encode_stream_code(&self, code: u32) -> noq::VarInt {
+        if self.raw {
+            return code.into();
+        }
+        noq::VarInt::try_from(web_transport_proto::error_to_http3(code)).unwrap()
+    }
+
+    /// Decode the peer's QUIC code for a stream reset or stop, or None if it is not one.
+    pub(crate) fn decode_stream_code(&self, code: noq::VarInt) -> Option<u32> {
+        let code = code.into_inner();
+        // An older raw peer maps its codes into the HTTP/3 range too. No u32 reaches that
+        // range, so the mapped form cannot be mistaken for a raw code.
+        match web_transport_proto::error_from_http3(code) {
+            Some(code) => Some(code),
+            None if self.raw => u32::try_from(code).ok(),
+            None => None,
+        }
+    }
+
     /// Record the close reason, returning false if one was already recorded.
     pub(crate) fn set(&self, err: SessionError) -> bool {
         self.reason.set(err).is_ok()
@@ -348,3 +371,35 @@ impl web_transport_trait::Error for ReadError {
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    /// WebTransport keeps the HTTP/3 mapping in both directions, and a code outside its
+    /// range is not a WebTransport code.
+    #[test]
+    fn http3_stream_codes_stay_mapped() {
+        let close = CloseReason::new(false);
+        let mapped = close.encode_stream_code(5);
+        assert_eq!(mapped.into_inner(), web_transport_proto::error_to_http3(5));
+        assert_eq!(close.decode_stream_code(mapped), Some(5));
+        assert_eq!(close.decode_stream_code(5u32.into()), None);
+    }
+
+    /// Raw QUIC sends the code as is and reads both the raw and the legacy mapped form.
+    #[test]
+    fn raw_stream_codes_pass_through() {
+        let close = CloseReason::new(true);
+        assert_eq!(close.encode_stream_code(5).into_inner(), 5);
+        assert_eq!(close.decode_stream_code(5u32.into()), Some(5));
+        assert_eq!(close.decode_stream_code(u32::MAX.into()), Some(u32::MAX));
+
+        let legacy = noq::VarInt::try_from(web_transport_proto::error_to_http3(5)).unwrap();
+        assert_eq!(close.decode_stream_code(legacy), Some(5));
+
+        // Past u32 and outside the HTTP/3 range: neither form.
+        let invalid = noq::VarInt::from_u64(1 << 40).unwrap();
+        assert_eq!(close.decode_stream_code(invalid), None);
+    }
+}

web-transport-moq/src/recv.rs

@@ -24,20 +24,25 @@ impl RecvStream {
         }
     }
 
-    /// Report a session error as the session's close reason.
-    fn map_error(&self, e: impl Into<ReadError>) -> ReadError {
-        match e.into() {
-            ReadError::SessionError(e) => ReadError::SessionError(self.close.map(e)),
-            e => e,
+    /// Decode the peer's reset code, and report a session error as the session's close reason.
+    fn map_error(&self, e: noq::ReadError) -> ReadError {
+        match e {
+            noq::ReadError::Reset(code) => match self.close.decode_stream_code(code) {
+                Some(code) => ReadError::Reset(code),
+                None => ReadError::InvalidReset(code),
+            },
+            e => match ReadError::from(e) {
+                ReadError::SessionError(e) => ReadError::SessionError(self.close.map(e)),
+                e => e,
+            },
         }
     }
 
     /// Tell the other end to stop sending data with the given error code. See
     /// [`noq::RecvStream::stop`]. This is a u32 with WebTransport since it shares the error
-    /// space with HTTP/3.
+    /// space with HTTP/3. A raw QUIC session sends the code as is.
     pub fn stop(&mut self, code: u32) -> Result<(), noq::ClosedStream> {
-        let code = web_transport_proto::error_to_http3(code);
-        let code = noq::VarInt::try_from(code).unwrap();
+        let code = self.close.encode_stream_code(code);
         self.inner.stop(code)
     }
 
@@ -93,7 +98,7 @@ impl RecvStream {
     pub async fn received_reset(&mut self) -> Result<Option<u32>, SessionError> {
         match self.inner.received_reset().await {
             Ok(None) => Ok(None),
-            Ok(Some(code)) => Ok(web_transport_proto::error_from_http3(code.into_inner())),
+            Ok(Some(code)) => Ok(self.close.decode_stream_code(code)),
             Err(noq::ResetError::ConnectionLost(conn_err)) => Err(self.close.map(conn_err.into())),
             Err(noq::ResetError::ZeroRttRejected) => unreachable!("0-RTT not supported"),
         }

web-transport-moq/src/send.rs

@@ -24,19 +24,25 @@ impl SendStream {
         Self { stream, close }
     }
 
-    /// Report a session error as the session's close reason.
-    fn map_error(&self, e: impl Into<WriteError>) -> WriteError {
-        match e.into() {
-            WriteError::SessionError(e) => WriteError::SessionError(self.close.map(e)),
-            e => e,
+    /// Decode the peer's stop code, and report a session error as the session's close reason.
+    fn map_error(&self, e: noq::WriteError) -> WriteError {
+        match e {
+            noq::WriteError::Stopped(code) => match self.close.decode_stream_code(code) {
+                Some(code) => WriteError::Stopped(code),
+                None => WriteError::InvalidStopped(code),
+            },
+            e => match WriteError::from(e) {
+                WriteError::SessionError(e) => WriteError::SessionError(self.close.map(e)),
+                e => e,
+            },
         }
     }
 
     /// Abruptly reset the stream with the provided error code. See [`noq::SendStream::reset`].
     /// This is a u32 with WebTransport because we share the error space with HTTP/3.
+    /// A raw QUIC session sends the code as is.
     pub fn reset(&mut self, code: u32) -> Result<(), ClosedStream> {
-        let code = web_transport_proto::error_to_http3(code);
-        let code = noq::VarInt::try_from(code).unwrap();
+        let code = self.close.encode_stream_code(code);
         self.stream.reset(code).map_err(Into::into)
     }
 
@@ -48,7 +54,7 @@ impl SendStream {
     /// supported.
     pub async fn stopped(&self) -> Result<Option<u32>, SessionError> {
         match self.stream.stopped().await {
-            Ok(Some(code)) => Ok(web_transport_proto::error_from_http3(code.into_inner())),
+            Ok(Some(code)) => Ok(self.close.decode_stream_code(code)),
             Ok(None) => Ok(None),
             Err(noq::StoppedError::ConnectionLost(conn_err)) => {
                 Err(self.close.map(conn_err.into()))

web-transport-moq/tests/raw_stream_codes.rs

@@ -0,0 +1,167 @@
+//! A raw QUIC session sends RESET_STREAM and STOP_SENDING codes as is, so a plain QUIC
+//! peer agrees on them. It still reads the HTTP/3-mapped codes an older raw peer sends.
+
+use std::{net::Ipv4Addr, sync::Arc, time::Duration};
+
+use anyhow::{Context as _, Result};
+use rcgen::{CertifiedKey, KeyPair};
+use rustls::pki_types::{PrivateKeyDer, PrivatePkcs8KeyDer};
+use tokio::time::timeout;
+use web_transport_moq::{ReadError, Session, WriteError, noq, proto};
+use web_transport_trait::Error as _;
+
+const CODE: u32 = 5;
+
+/// With both crypto features enabled, as `--all-features` does, the crate cannot pick a
+/// provider for us, so the test installs one.
+fn install_crypto_provider() {
+    #[cfg(all(feature = "aws-lc-rs", feature = "ring"))]
+    {
+        let _ = rustls::crypto::aws_lc_rs::default_provider().install_default();
+    }
+}
+
+/// A raw session on the client and the plain QUIC connection it talks to, plus the
+/// endpoints that drive them.
+async fn raw_and_plain() -> Result<(Session, noq::Connection, [noq::Endpoint; 2])> {
+    install_crypto_provider();
+
+    let CertifiedKey { cert, signing_key } =
+        rcgen::generate_simple_self_signed(vec!["localhost".into()])?;
+    let key = PrivateKeyDer::Pkcs8(PrivatePkcs8KeyDer::from(KeyPair::serialize_der(
+        &signing_key,
+    )));
+
+    let server_config = noq::ServerConfig::with_single_cert(vec![cert.der().clone()], key)?;
+    let server = noq::Endpoint::server(server_config, (Ipv4Addr::LOCALHOST, 0).into())?;
+
+    let mut roots = rustls::RootCertStore::empty();
+    roots.add(cert.der().clone())?;
+    let client_config = noq::ClientConfig::with_root_certificates(Arc::new(roots))?;
+    let client = noq::Endpoint::client((Ipv4Addr::LOCALHOST, 0).into())?;
+
+    let connecting = client.connect_with(client_config, server.local_addr()?, "localhost")?;
+    let accept = async { anyhow::Ok(server.accept().await.context("no connection")?.await?) };
+    let (client_conn, server_conn) =
+        tokio::try_join!(async { anyhow::Ok(connecting.await?) }, accept)?;
+
+    Ok((Session::raw(client_conn), server_conn, [client, server]))
+}
+
+/// Open a uni stream from the plain peer and accept it on the session.
+async fn plain_to_raw(
+    session: &Session,
+    plain: &noq::Connection,
+) -> Result<(noq::SendStream, web_transport_moq::RecvStream)> {
+    let mut send = plain.open_uni().await?;
+    send.write_all(b"x").await?;
+    let mut recv = timeout(Duration::from_secs(5), session.accept_uni()).await??;
+    let mut buf = [0u8; 1];
+    recv.read_exact(&mut buf).await?;
+    Ok((send, recv))
+}
+
+/// Open a uni stream from the session and accept it on the plain peer.
+async fn raw_to_plain(
+    session: &Session,
+    plain: &noq::Connection,
+) -> Result<(web_transport_moq::SendStream, noq::RecvStream)> {
+    let mut send = session.open_uni().await?;
+    send.write_all(b"x").await?;
+    let mut recv = timeout(Duration::from_secs(5), plain.accept_uni()).await??;
+    let mut buf = [0u8; 1];
+    recv.read_exact(&mut buf).await?;
+    Ok((send, recv))
+}
+
+async fn read_err(recv: &mut web_transport_moq::RecvStream) -> ReadError {
+    let mut buf = [0u8; 1];
+    timeout(Duration::from_secs(5), recv.read(&mut buf))
+        .await
+        .expect("no reset")
+        .expect_err("read after a reset")
+}
+
+async fn write_err(send: &mut web_transport_moq::SendStream) -> WriteError {
+    timeout(Duration::from_secs(5), async {
+        loop {
+            if let Err(err) = send.write(&[0u8; 1024]).await {
+                return err;
+            }
+        }
+    })
+    .await
+    .expect("no stop")
+}
+
+#[tokio::test]
+async fn reset_is_sent_as_is() -> Result<()> {
+    let (session, plain, _endpoints) = raw_and_plain().await?;
+    let (mut send, mut recv) = raw_to_plain(&session, &plain).await?;
+
+    send.reset(CODE)?;
+
+    let mut buf = [0u8; 1];
+    let err = timeout(Duration::from_secs(5), recv.read(&mut buf)).await?;
+    assert!(
+        matches!(err, Err(noq::ReadError::Reset(code)) if code == CODE.into()),
+        "{err:?}"
+    );
+    Ok(())
+}
+
+#[tokio::test]
+async fn stop_is_sent_as_is() -> Result<()> {
+    let (session, plain, _endpoints) = raw_and_plain().await?;
+    let (send, mut recv) = plain_to_raw(&session, &plain).await?;
+
+    recv.stop(CODE)?;
+
+    let stopped = timeout(Duration::from_secs(5), send.stopped()).await??;
+    assert_eq!(stopped, Some(CODE.into()));
+    Ok(())
+}
+
+#[tokio::test]
+async fn peer_reset_is_read_as_is() -> Result<()> {
+    let (session, plain, _endpoints) = raw_and_plain().await?;
+    let (mut send, mut recv) = plain_to_raw(&session, &plain).await?;
+
+    send.reset(CODE.into())?;
+
+    let err = read_err(&mut recv).await;
+    assert_eq!(err.stream_error(), Some(CODE), "{err}");
+    Ok(())
+}
+
+#[tokio::test]
+async fn peer_stop_is_read_as_is() -> Result<()> {
+    let (session, plain, _endpoints) = raw_and_plain().await?;
+    let (mut send, mut recv) = raw_to_plain(&session, &plain).await?;
+
+    recv.stop(CODE.into())?;
+
+    let err = write_err(&mut send).await;
+    assert_eq!(err.stream_error(), Some(CODE), "{err}");
+    assert_eq!(send.stopped().await?, Some(CODE));
+    Ok(())
+}
+
+/// An older raw peer maps its codes into the HTTP/3 range, which still decodes.
+#[tokio::test]
+async fn legacy_peer_codes_are_unmapped() -> Result<()> {
+    let (session, plain, _endpoints) = raw_and_plain().await?;
+    let legacy = noq::VarInt::try_from(proto::error_to_http3(CODE))?;
+
+    let (mut send, mut recv) = plain_to_raw(&session, &plain).await?;
+    send.reset(legacy)?;
+    let err = read_err(&mut recv).await;
+    assert_eq!(err.stream_error(), Some(CODE), "{err}");
+
+    let (mut send, mut recv) = raw_to_plain(&session, &plain).await?;
+    recv.stop(legacy)?;
+    let err = write_err(&mut send).await;
+    assert_eq!(err.stream_error(), Some(CODE), "{err}");
+    assert_eq!(send.stopped().await?, Some(CODE));
+    Ok(())
+}