Merge pull request #143 from englishm-cloudflare/me/web-transport-upgrade

Author: englishm

Cargo.lock

@@ -13,7 +13,7 @@ members = [
 resolver = "2"
 
 [workspace.dependencies]
-web-transport = "0.3"
+web-transport = "0.10"
 tracing = "0.1"
 tracing-subscriber = { version = "0.3", features = ["env-filter"] }
 

Cargo.toml

@@ -14,7 +14,7 @@ categories = ["multimedia", "network-programming", "web-programming"]
 [dependencies]
 moq-transport = { path = "../moq-transport", version = "0.12" }
 web-transport = { workspace = true }
-web-transport-quinn = "0.3"
+web-transport-quinn = { version = "0.11", default-features = false, features = ["ring"] }
 
 rustls = { version = "0.23", features = ["ring"] }
 rustls-pemfile = "2"

moq-native-ietf/Cargo.toml

@@ -160,7 +160,7 @@ impl Endpoint {
 
         if let Some(mut config) = config.tls.server {
             config.alpn_protocols = vec![
-                web_transport_quinn::ALPN.to_vec(),
+                web_transport_quinn::ALPN.as_bytes().to_vec(),
                 moq_transport::setup::ALPN.to_vec(),
             ];
             config.key_log = Arc::new(rustls::KeyLogFile::new());
@@ -305,22 +305,28 @@ impl Server {
             server_name,
         );
 
-        let session = match alpn.as_bytes() {
-            web_transport_quinn::ALPN => {
-                // Wait for the CONNECT request.
-                let request = web_transport_quinn::accept(conn)
-                    .await
-                    .context("failed to receive WebTransport request")?;
-
-                // Accept the CONNECT request.
-                request
-                    .ok()
-                    .await
-                    .context("failed to respond to WebTransport request")?
-            }
-            // A bit of a hack to pretend like we're a WebTransport session
-            moq_transport::setup::ALPN => conn.into(),
-            _ => anyhow::bail!("unsupported ALPN: {}", alpn),
+        let alpn_bytes = alpn.as_bytes();
+        let session = if alpn_bytes == web_transport_quinn::ALPN.as_bytes() {
+            // Wait for the WebTransport CONNECT request (includes H3 SETTINGS exchange).
+            let request = web_transport_quinn::Request::accept(conn)
+                .await
+                .context("failed to receive WebTransport request")?;
+
+            // Accept the CONNECT request.
+            request
+                .ok()
+                .await
+                .context("failed to respond to WebTransport request")?
+        } else if alpn_bytes == moq_transport::setup::ALPN {
+            // Raw QUIC mode — create a "fake" WebTransport session with no H3 framing.
+            let request = url::Url::parse("moqt://localhost").unwrap();
+            web_transport_quinn::Session::raw(
+                conn,
+                request,
+                web_transport_quinn::proto::ConnectResponse::default(),
+            )
+        } else {
+            anyhow::bail!("unsupported ALPN: {}", alpn)
         };
 
         Ok((session.into(), connection_id_hex))
@@ -373,7 +379,7 @@ impl Client {
 
         // TODO support connecting to both ALPNs at the same time
         config.alpn_protocols = vec![match url.scheme() {
-            "https" => web_transport_quinn::ALPN.to_vec(),
+            "https" => web_transport_quinn::ALPN.as_bytes().to_vec(),
             "moqt" => moq_transport::setup::ALPN.to_vec(),
             _ => anyhow::bail!("url scheme must be 'https' or 'moqt'"),
         }];
@@ -426,8 +432,12 @@ impl Client {
             .to_string();
 
         let session = match url.scheme() {
-            "https" => web_transport_quinn::connect_with(connection, url).await?,
-            "moqt" => connection.into(),
+            "https" => web_transport_quinn::Session::connect(connection, url.clone()).await?,
+            "moqt" => web_transport_quinn::Session::raw(
+                connection,
+                url.clone(),
+                web_transport_quinn::proto::ConnectResponse::default(),
+            ),
             _ => unreachable!(),
         };
 

moq-native-ietf/src/quic.rs

@@ -18,7 +18,7 @@ path = "src/main.rs"
 [dependencies]
 moq-transport = { path = "../moq-transport", version = "0.12" }
 moq-native-ietf = { path = "../moq-native-ietf", version = "0.7" }
-web-transport = "0.3"
+web-transport = { workspace = true }
 
 url = "2"
 

moq-test-client/Cargo.toml

@@ -23,15 +23,6 @@ uuid = { version = "1", features = ["v4"] }
 
 web-transport = { workspace = true }
 
-# Used for is_graceful_close() to structurally match on connection errors
-# and decode WebTransport error codes. This couples moq-transport to the
-# quinn-based WebTransport implementation. When transitioning to a different
-# backend (e.g., tokio-quiche), these dependencies should be replaced with
-# equivalent types from the new implementation.
-web-transport-quinn = "0.3"
-web-transport-proto = "0.2"
-quinn = "0.11"
-
 paste = "1"
 futures = "0.3"
 serde = { version = "1", features = ["derive"] }

moq-transport/Cargo.toml

@@ -2,14 +2,8 @@ use crate::{coding, serve, setup};
 
 #[derive(thiserror::Error, Debug, Clone)]
 pub enum SessionError {
-    #[error("webtransport session: {0}")]
-    Session(#[from] web_transport::SessionError),
-
-    #[error("webtransport write: {0}")]
-    Write(#[from] web_transport::WriteError),
-
-    #[error("webtransport read: {0}")]
-    Read(#[from] web_transport::ReadError),
+    #[error("webtransport error: {0}")]
+    WebTransport(#[from] web_transport::Error),
 
     #[error("encode error: {0}")]
     Encode(#[from] coding::EncodeError),
@@ -53,9 +47,7 @@ impl SessionError {
             // PROTOCOL_VIOLATION (0x3) - The role negotiated in the handshake was violated
             Self::RoleViolation => 0x3,
             // INTERNAL_ERROR (0x1) - Generic internal errors
-            Self::Session(_) => 0x1,
-            Self::Read(_) => 0x1,
-            Self::Write(_) => 0x1,
+            Self::WebTransport(_) => 0x1,
             Self::Encode(_) => 0x1,
             Self::BoundsExceeded(_) => 0x1,
             Self::Internal => 0x1,
@@ -79,42 +71,48 @@ impl SessionError {
 
     /// Returns true if this error represents a graceful connection close.
     ///
-    /// A graceful close occurs when the peer sends APPLICATION_CLOSE with error code 0
-    /// (NO_ERROR). This is normal session termination, not an error condition.
+    /// For WebTransport, a graceful close is a `CLOSE_WEBTRANSPORT_SESSION` capsule
+    /// with code 0. For raw QUIC, it's `APPLICATION_CLOSE` with code 0 (NO_ERROR).
+    /// Both are normal session termination, not error conditions.
     ///
     /// This method checks for:
-    /// - WebTransport close with code 0 (HTTP/3 encoded as 0x52e4a40fa8db)
+    /// - WebTransport `Closed(0, _)` — web-transport-quinn v0.11+ typically converts
+    ///   HTTP/3-encoded `ApplicationClosed` codes into `WebTransportError::Closed(code, reason)`
+    ///   during `SessionError` conversion when decoding via `error_from_http3` succeeds
     /// - Raw QUIC `ApplicationClosed` with code 0
     /// - The local side closing the connection (`LocallyClosed`)
     ///
     /// ## Implementation Notes
     ///
-    /// We pattern match on `web_transport_quinn::SessionError` variants to access the
-    /// underlying `quinn::ConnectionError`. For WebTransport connections, the close code
-    /// is encoded using HTTP/3 error code space, which we decode using
-    /// `web_transport_proto::error_from_http3()`.
+    /// We pattern match on `web_transport_quinn::SessionError` variants. In v0.11+,
+    /// WebTransport graceful closes arrive as `WebTransportError::Closed(0, _)` because
+    /// the crate decodes HTTP/3 error codes at the `SessionError` level. For raw QUIC
+    /// connections, the close code is checked directly on `ConnectionError::ApplicationClosed`.
     ///
     /// **Coupling note**: This implementation is coupled to `web-transport-quinn` and
     /// `quinn`. When transitioning to a different WebTransport backend (e.g., tokio-quiche),
     /// ensure the replacement provides equivalent error introspection, or update this
     /// method to handle the new error types.
     pub fn is_graceful_close(&self) -> bool {
         match self {
-            Self::Session(session_err) => is_session_error_graceful(session_err),
-            Self::Read(read_err) => {
-                // ReadError::SessionError wraps SessionError
-                if let web_transport::ReadError::SessionError(session_err) = read_err {
-                    return is_session_error_graceful(session_err);
+            Self::WebTransport(wt_err) => match wt_err {
+                web_transport::Error::Session(session_err) => {
+                    is_session_error_graceful(session_err)
                 }
-                false
-            }
-            Self::Write(write_err) => {
-                // WriteError::SessionError wraps SessionError
-                if let web_transport::WriteError::SessionError(session_err) = write_err {
-                    return is_session_error_graceful(session_err);
+                web_transport::Error::Read(read_err) => {
+                    if let web_transport::quinn::ReadError::SessionError(session_err) = read_err {
+                        return is_session_error_graceful(session_err);
+                    }
+                    false
                 }
-                false
-            }
+                web_transport::Error::Write(write_err) => {
+                    if let web_transport::quinn::WriteError::SessionError(session_err) = write_err {
+                        return is_session_error_graceful(session_err);
+                    }
+                    false
+                }
+                _ => false,
+            },
             _ => false,
         }
     }
@@ -129,26 +127,37 @@ impl From<SessionError> for serve::ServeError {
     }
 }
 
-/// Helper to check if a `web_transport::SessionError` represents a graceful close.
+/// Helper to check if a `web_transport_quinn::SessionError` represents a graceful close.
 ///
-/// This handles both:
-/// - Raw QUIC connections: `ApplicationClosed` with code 0
-/// - WebTransport connections: `ApplicationClosed` with HTTP/3 encoded code that decodes to 0
-fn is_session_error_graceful(err: &web_transport::SessionError) -> bool {
-    use web_transport_quinn::SessionError;
+/// This handles:
+/// - WebTransport connections: `WebTransportError::Closed(0, _)` — web-transport-quinn v0.11+
+///   typically decodes HTTP/3-encoded close codes at this layer (when `SessionError` conversion
+///   applies), so graceful closes usually arrive here rather than as a raw
+///   `ConnectionError::ApplicationClosed`.
+/// - Raw QUIC connections: `ConnectionError::ApplicationClosed` with code 0
+/// - Local close: `ConnectionError::LocallyClosed`
+fn is_session_error_graceful(err: &web_transport::quinn::SessionError) -> bool {
+    use web_transport::quinn::{SessionError, WebTransportError};
 
     match err {
         SessionError::ConnectionError(conn_err) => is_connection_error_graceful(conn_err),
-        // WebTransportError doesn't represent connection close in 0.3.x
+        // WebTransport graceful close: peer sent close with code 0
+        SessionError::WebTransportError(WebTransportError::Closed(0, _)) => true,
+        // Other WebTransport errors (UnknownSession, read/write errors, non-zero close codes)
         SessionError::WebTransportError(_) => false,
         // SendDatagramError doesn't represent connection close
         SessionError::SendDatagramError(_) => false,
     }
 }
 
 /// Helper to check if a `quinn::ConnectionError` represents a graceful close.
-fn is_connection_error_graceful(err: &quinn::ConnectionError) -> bool {
-    use quinn::ConnectionError;
+///
+/// Note: In web-transport-quinn v0.11+, WebTransport `ApplicationClosed` with an HTTP/3-encoded
+/// close code is usually converted to `WebTransportError::Closed` during `SessionError` conversion
+/// when decoding succeeds. This function primarily handles raw QUIC (moqt:// ALPN) connections
+/// or non-decodable cases where the close code is not HTTP/3 encoded.
+fn is_connection_error_graceful(err: &web_transport::quinn::quinn::ConnectionError) -> bool {
+    use web_transport::quinn::quinn::ConnectionError;
 
     match err {
         ConnectionError::ApplicationClosed(close) => {
@@ -160,8 +169,9 @@ fn is_connection_error_graceful(err: &quinn::ConnectionError) -> bool {
             }
 
             // Check for WebTransport code 0 (HTTP/3 encoded)
-            // WebTransport code 0 maps to HTTP/3 code 0x52e4a40fa8db
-            if let Some(wt_code) = web_transport_proto::error_from_http3(code) {
+            // This is a fallback — in v0.11+, WebTransport closes are typically caught
+            // by is_session_error_graceful's WebTransportError::Closed branch.
+            if let Some(wt_code) = web_transport::quinn::proto::error_from_http3(code) {
                 return wt_code == 0;
             }
 

moq-transport/src/session/error.rs

@@ -394,7 +394,7 @@ impl Session {
     /// Create an outbound/client QUIC connection, by opening a bi-directional QUIC stream for
     /// MOQT control messaging.  Performs SETUP messaging and version negotiation.
     pub async fn connect(
-        mut session: web_transport::Session,
+        session: web_transport::Session,
         mlog_path: Option<PathBuf>,
     ) -> Result<(Session, Publisher, Subscriber), SessionError> {
         let mlog = mlog_path.and_then(|path| {
@@ -447,7 +447,7 @@ impl Session {
     /// Accepts an inbound/server QUIC connection, by accepting a bi-directional QUIC stream for
     /// MOQT control messaging.  Performs SETUP messaging and version negotiation.
     pub async fn accept(
-        mut session: web_transport::Session,
+        session: web_transport::Session,
         mlog_path: Option<PathBuf>,
     ) -> Result<(Session, Option<Publisher>, Option<Subscriber>), SessionError> {
         let mut mlog = mlog_path.and_then(|path| {
@@ -673,7 +673,7 @@ impl Session {
     /// Will read stream header to know what type of stream it is and create
     /// the appropriate stream handlers.
     async fn run_streams(
-        mut webtransport: web_transport::Session,
+        webtransport: web_transport::Session,
         subscriber: Option<Subscriber>,
     ) -> Result<(), SessionError> {
         let mut tasks = FuturesUnordered::new();
@@ -697,7 +697,7 @@ impl Session {
 
     /// Receives QUIC datagrams and processes them using the Subscriber logic
     async fn run_datagrams(
-        mut webtransport: web_transport::Session,
+        webtransport: web_transport::Session,
         mut subscriber: Option<Subscriber>,
     ) -> Result<(), SessionError> {
         loop {

moq-transport/src/session/mod.rs

@@ -68,7 +68,7 @@ impl Reader {
             // We always read at least once to avoid an infinite loop if some dingus puts remain=0
             loop {
                 let before_read = self.buffer.len();
-                if !self.stream.read_buf(&mut self.buffer).await? {
+                if self.stream.read_buf(&mut self.buffer).await?.is_none() {
                     tracing::warn!(
                         "[READER] decode: stream ended while waiting for data (have={} bytes, need={})",
                         self.buffer.len(),
@@ -113,7 +113,7 @@ impl Reader {
             return Ok(Some(data));
         }
 
-        let chunk = self.stream.read_chunk(max).await?;
+        let chunk = self.stream.read(max).await?;
         if let Some(ref data) = chunk {
             tracing::trace!("[READER] read_chunk: read {} bytes from stream", data.len());
         } else {
@@ -127,6 +127,6 @@ impl Reader {
             return Ok(false);
         }
 
-        Ok(!self.stream.read_buf(&mut self.buffer).await?)
+        Ok(self.stream.read_buf(&mut self.buffer).await?.is_none())
     }
 }