0.3.5-alpha.4

crates/common/game-realtime-webtransport-lib/src/webtransport.rs

@@ -1,11 +1,10 @@
 // file: crates/common/game-realtime-webtransport-lib/src/webtransport.rs
-// version: 2
+// version: 3
 
 const CERTIFICATE_HASH_SIZE: usize = 32;
 const LOCAL_CERTIFICATE_CLOCK_SKEW: std::time::Duration = std::time::Duration::from_secs(60);
 const LOCAL_CERTIFICATE_VALIDITY: std::time::Duration = std::time::Duration::from_secs(7 * 24 * 60 * 60);
 const PRIMARY_FRAME_HEADER_SIZE: usize = 4;
-const PRIMARY_FRAME_MAX_PAYLOAD_SIZE: usize = 1024 * 1024;
 const TRACING_TARGET: &str = "games::realtime::webtransport";
 
 /// SHA-256 fingerprint of one certificate accepted by the native WebTransport client.
@@ -91,11 +90,12 @@ impl WebTransportServerIdentity {
     }
 }
 
-/// Native WebTransport client endpoint and pinned server-certificate fingerprint.
+/// Native WebTransport client endpoint, certificate pin and reliable-path configuration.
 #[derive(Clone, Debug, Eq, PartialEq)]
 pub struct WebTransportClientConfig {
     endpoint: url::Url,
     certificate_hash: WebTransportCertificateHash,
+    transport: crate::WebTransportConfig,
 }
 
 impl WebTransportClientConfig {
@@ -111,7 +111,14 @@ impl WebTransportClientConfig {
         if parsed.host().is_none() {
             return Err(invalid_configuration("WebTransport endpoint must contain a host"));
         }
-        return Ok(Self { endpoint: parsed, certificate_hash });
+        return Ok(Self { endpoint: parsed, certificate_hash, transport: crate::WebTransportConfig::default() });
+    }
+
+    /// Returns a copy with explicit reliable-path limits and deadlines.
+    #[must_use]
+    pub fn with_transport_config(mut self, transport: crate::WebTransportConfig) -> Self {
+        self.transport = transport;
+        return self;
     }
 
     /// Returns the validated WebTransport endpoint URL.
@@ -125,19 +132,33 @@ impl WebTransportClientConfig {
     pub fn certificate_hash(&self) -> &WebTransportCertificateHash {
         return &self.certificate_hash;
     }
+
+    /// Returns the reliable-path limits and deadlines.
+    #[must_use]
+    pub fn transport_config(&self) -> crate::WebTransportConfig {
+        return self.transport;
+    }
 }
 
-/// Native WebTransport server bind address and TLS identity.
+/// Native WebTransport server bind address, TLS identity and reliable-path configuration.
 pub struct WebTransportServerConfig {
     bind_address: std::net::SocketAddr,
     identity: WebTransportServerIdentity,
+    transport: crate::WebTransportConfig,
 }
 
 impl WebTransportServerConfig {
     /// Creates native server configuration for the requested bind address and TLS identity.
     #[must_use]
     pub fn new(bind_address: std::net::SocketAddr, identity: WebTransportServerIdentity) -> Self {
-        return Self { bind_address, identity };
+        return Self { bind_address, identity, transport: crate::WebTransportConfig::default() };
+    }
+
+    /// Returns a copy with explicit reliable-path limits and deadlines.
+    #[must_use]
+    pub fn with_transport_config(mut self, transport: crate::WebTransportConfig) -> Self {
+        self.transport = transport;
+        return self;
     }
 
     /// Returns the requested UDP bind address.
@@ -151,16 +172,23 @@ impl WebTransportServerConfig {
     pub fn certificate_hash(&self) -> &WebTransportCertificateHash {
         return self.identity.certificate_hash();
     }
+
+    /// Returns the reliable-path limits and deadlines.
+    #[must_use]
+    pub fn transport_config(&self) -> crate::WebTransportConfig {
+        return self.transport;
+    }
 }
 
 /// Established native WebTransport session before or while the single primary application stream is selected.
 pub struct WebTransportSession {
     inner: web_transport_quinn::Session,
+    transport: crate::WebTransportConfig,
 }
 
 impl WebTransportSession {
-    fn new(inner: web_transport_quinn::Session) -> Self {
-        return Self { inner };
+    fn new(inner: web_transport_quinn::Session, transport: crate::WebTransportConfig) -> Self {
+        return Self { inner, transport };
     }
 
     /// Accepts the peer-created primary bidirectional stream and adapts it to the transport-neutral realtime contract.
@@ -168,30 +196,44 @@ impl WebTransportSession {
     /// The native WebTransport wrapper writes the required stream/session header while opening the stream, so the peer can
     /// accept it before the first application frame is sent.
     pub async fn accept_primary_connection(self) -> Result<WebTransportConnection, game_realtime_transport_lib::TransportError> {
-        let (sender, receiver) = match self.inner.accept_bi().await {
-            Ok(value) => value,
-            Err(error) => {
+        let timeout = self.transport.primary_stream_timeout();
+        let accepted = tokio::time::timeout(timeout, self.inner.accept_bi()).await;
+        let (sender, receiver) = match accepted {
+            Ok(Ok(value)) => value,
+            Ok(Err(error)) => {
                 let mapped = transport_error(game_realtime_transport_lib::TransportErrorKind::Protocol, error.to_string());
                 tracing::warn!(target: TRACING_TARGET, detail = mapped.detail(), "WebTransport primary bidirectional stream accept failed");
                 return Err(mapped);
             },
+            Err(_) => {
+                let mapped = timeout_error("WebTransport primary bidirectional stream accept", timeout);
+                tracing::warn!(target: TRACING_TARGET, timeout_ms = duration_millis(timeout), "WebTransport primary bidirectional stream accept timed out");
+                return Err(mapped);
+            },
         };
         tracing::debug!(target: TRACING_TARGET, peer = %self.inner.remote_address(), "WebTransport primary bidirectional stream accepted");
-        return Ok(WebTransportConnection::new(self.inner, sender, receiver));
+        return Ok(WebTransportConnection::new(self.inner, sender, receiver, self.transport));
     }
 
     /// Opens the single primary bidirectional stream and adapts it to the transport-neutral realtime contract.
     pub async fn open_primary_connection(self) -> Result<WebTransportConnection, game_realtime_transport_lib::TransportError> {
-        let (sender, receiver) = match self.inner.open_bi().await {
-            Ok(value) => value,
-            Err(error) => {
+        let timeout = self.transport.primary_stream_timeout();
+        let opened = tokio::time::timeout(timeout, self.inner.open_bi()).await;
+        let (sender, receiver) = match opened {
+            Ok(Ok(value)) => value,
+            Ok(Err(error)) => {
                 let mapped = transport_error(game_realtime_transport_lib::TransportErrorKind::Protocol, error.to_string());
                 tracing::warn!(target: TRACING_TARGET, detail = mapped.detail(), "WebTransport primary bidirectional stream open failed");
                 return Err(mapped);
             },
+            Err(_) => {
+                let mapped = timeout_error("WebTransport primary bidirectional stream open", timeout);
+                tracing::warn!(target: TRACING_TARGET, timeout_ms = duration_millis(timeout), "WebTransport primary bidirectional stream open timed out");
+                return Err(mapped);
+            },
         };
         tracing::debug!(target: TRACING_TARGET, peer = %self.inner.remote_address(), "WebTransport primary bidirectional stream opened");
-        return Ok(WebTransportConnection::new(self.inner, sender, receiver));
+        return Ok(WebTransportConnection::new(self.inner, sender, receiver, self.transport));
     }
 
     /// Returns the remote UDP socket address backing the established QUIC connection.
@@ -215,11 +257,17 @@ pub struct WebTransportConnection {
     receiver: web_transport_quinn::RecvStream,
     sender: web_transport_quinn::SendStream,
     session: web_transport_quinn::Session,
+    transport: crate::WebTransportConfig,
 }
 
 impl WebTransportConnection {
-    fn new(session: web_transport_quinn::Session, sender: web_transport_quinn::SendStream, receiver: web_transport_quinn::RecvStream) -> Self {
-        return Self { receiver, sender, session };
+    fn new(
+        session: web_transport_quinn::Session,
+        sender: web_transport_quinn::SendStream,
+        receiver: web_transport_quinn::RecvStream,
+        transport: crate::WebTransportConfig,
+    ) -> Self {
+        return Self { receiver, sender, session, transport };
     }
 }
 
@@ -230,8 +278,24 @@ impl game_realtime_transport_lib::RealtimeConnection for WebTransportConnection
     fn split(self) -> (Self::Sender, Self::Receiver) {
         let receiver_session = self.session.clone();
         return (
-            crate::WebTransportSender { inner: self.sender, _session: self.session },
-            crate::WebTransportReceiver { inner: self.receiver, _session: receiver_session },
+            crate::WebTransportSender {
+                inner: self.sender,
+                _session: self.session,
+                max_message_size: self.transport.max_message_size(),
+                send_timeout: self.transport.send_timeout(),
+                terminal: false,
+            },
+            crate::WebTransportReceiver {
+                inner: self.receiver,
+                _session: receiver_session,
+                max_message_size: self.transport.max_message_size(),
+                header: [0_u8; PRIMARY_FRAME_HEADER_SIZE],
+                header_read: 0,
+                payload: Vec::new(),
+                payload_read: 0,
+                clean_closed: false,
+                terminal: false,
+            },
         );
     }
 }
@@ -240,6 +304,137 @@ impl game_realtime_transport_lib::RealtimeConnection for WebTransportConnection
 pub struct WebTransportReceiver {
     inner: web_transport_quinn::RecvStream,
     _session: web_transport_quinn::Session,
+    max_message_size: usize,
+    header: [u8; PRIMARY_FRAME_HEADER_SIZE],
+    header_read: usize,
+    payload: Vec<u8>,
+    payload_read: usize,
+    clean_closed: bool,
+    terminal: bool,
+}
+
+impl WebTransportReceiver {
+    /// Abruptly stops the reliable receive direction with one WebTransport application error code.
+    pub fn abort(&mut self, code: u32) -> Result<(), game_realtime_transport_lib::TransportError> {
+        if self.clean_closed || self.terminal {
+            return Err(transport_error(game_realtime_transport_lib::TransportErrorKind::Closed, "WebTransport receiver is already terminal"));
+        }
+        return match self.inner.stop(code) {
+            Ok(()) => {
+                self.terminal = true;
+                tracing::debug!(target: TRACING_TARGET, code = code, "WebTransport primary receive stream aborted");
+                Ok(())
+            },
+            Err(error) => {
+                self.terminal = true;
+                let mapped = transport_error(game_realtime_transport_lib::TransportErrorKind::Closed, error.to_string());
+                tracing::warn!(target: TRACING_TARGET, code = code, detail = mapped.detail(), "WebTransport primary receive stream abort failed");
+                Err(mapped)
+            },
+        };
+    }
+
+    async fn receive_frame(&mut self) -> Result<game_realtime_transport_lib::TransportReceive, game_realtime_transport_lib::TransportError> {
+        if self.clean_closed {
+            return Ok(game_realtime_transport_lib::TransportReceive::Closed);
+        }
+        if self.terminal {
+            return Err(transport_error(
+                game_realtime_transport_lib::TransportErrorKind::Aborted,
+                "WebTransport receiver is unavailable after a terminal stream failure or abort",
+            ));
+        }
+        loop {
+            if self.header_read < PRIMARY_FRAME_HEADER_SIZE {
+                let read = self.inner.read(&mut self.header[self.header_read..]).await;
+                match read {
+                    Ok(Some(0)) => return self.fail_protocol("WebTransport primary stream returned an empty read in the middle of a frame header"),
+                    Ok(Some(value)) => {
+                        self.header_read += value;
+                        continue;
+                    },
+                    Ok(None) => {
+                        if self.header_read == 0 {
+                            self.clean_closed = true;
+                            tracing::debug!(target: TRACING_TARGET, "remote WebTransport primary stream closed cleanly");
+                            return Ok(game_realtime_transport_lib::TransportReceive::Closed);
+                        }
+                        return self.fail_protocol("WebTransport primary stream closed in the middle of a frame header");
+                    },
+                    Err(error) => return self.fail_read(error),
+                }
+            }
+            if self.payload.is_empty() && self.payload_read == 0 {
+                let payload_len = u32::from_be_bytes(self.header) as usize;
+                if payload_len > self.max_message_size {
+                    let error = message_too_large(payload_len, self.max_message_size);
+                    self.stop_after_failure(FRAME_TOO_LARGE_ERROR_CODE);
+                    return Err(error);
+                }
+                if payload_len == 0 {
+                    self.reset_frame_state();
+                    tracing::trace!(target: TRACING_TARGET, payload_len = 0, "framed WebTransport payload received");
+                    return Ok(game_realtime_transport_lib::TransportReceive::Message(game_realtime_transport_lib::TransportMessage::new(Vec::new())));
+                }
+                self.payload = vec![0_u8; payload_len];
+            }
+            if self.payload_read < self.payload.len() {
+                let read = self.inner.read(&mut self.payload[self.payload_read..]).await;
+                match read {
+                    Ok(Some(0)) => return self.fail_protocol("WebTransport primary stream returned an empty read in the middle of a frame payload"),
+                    Ok(Some(value)) => {
+                        self.payload_read += value;
+                        if self.payload_read < self.payload.len() {
+                            continue;
+                        }
+                    },
+                    Ok(None) => return self.fail_protocol("WebTransport primary stream closed in the middle of a frame payload"),
+                    Err(error) => return self.fail_read(error),
+                }
+            }
+            let payload = core::mem::take(&mut self.payload);
+            self.reset_frame_state();
+            tracing::trace!(target: TRACING_TARGET, payload_len = payload.len(), "framed WebTransport payload received");
+            return Ok(game_realtime_transport_lib::TransportReceive::Message(game_realtime_transport_lib::TransportMessage::new(payload)));
+        }
+    }
+
+    fn fail_protocol(&mut self, detail: &str) -> Result<game_realtime_transport_lib::TransportReceive, game_realtime_transport_lib::TransportError> {
+        let error = protocol_error(detail);
+        self.stop_after_failure(FRAME_PROTOCOL_ERROR_CODE);
+        return Err(error);
+    }
+
+    fn fail_read(
+        &mut self,
+        error: web_transport_quinn::ReadError,
+    ) -> Result<game_realtime_transport_lib::TransportReceive, game_realtime_transport_lib::TransportError> {
+        self.terminal = true;
+        let mapped = map_read_error(error);
+        tracing::warn!(target: TRACING_TARGET, kind = %mapped.kind(), detail = mapped.detail(), "WebTransport primary stream receive failed");
+        return Err(mapped);
+    }
+
+    fn reset_frame_state(&mut self) {
+        self.header = [0_u8; PRIMARY_FRAME_HEADER_SIZE];
+        self.header_read = 0;
+        self.payload.clear();
+        self.payload_read = 0;
+    }
+
+    fn stop_after_failure(&mut self, code: u32) {
+        let _ = self.inner.stop(code);
+        self.terminal = true;
+    }
+}
+
+impl Drop for WebTransportReceiver {
+    fn drop(&mut self) {
+        if !self.clean_closed && !self.terminal {
+            let _ = self.inner.stop(STREAM_CANCELLED_ERROR_CODE);
+            self.terminal = true;
+        }
+    }
 }
 
 impl game_realtime_transport_lib::RealtimeReceiver for WebTransportReceiver {
@@ -251,22 +446,7 @@ impl game_realtime_transport_lib::RealtimeReceiver for WebTransportReceiver {
         Self: 'a;
 
     fn receive(&mut self) -> Self::ReceiveFuture<'_> {
-        return Box::pin(async move {
-            let payload_len = match read_frame_payload_len(&mut self.inner).await {
-                Ok(Some(value)) => value,
-                Ok(None) => {
-                    tracing::debug!(target: TRACING_TARGET, "remote WebTransport primary stream closed cleanly");
-                    return Ok(game_realtime_transport_lib::TransportReceive::Closed);
-                },
-                Err(error) => return Err(error),
-            };
-            let payload = match read_frame_payload(&mut self.inner, payload_len).await {
-                Ok(value) => value,
-                Err(error) => return Err(error),
-            };
-            tracing::trace!(target: TRACING_TARGET, payload_len = payload.len(), "framed WebTransport payload received");
-            return Ok(game_realtime_transport_lib::TransportReceive::Message(game_realtime_transport_lib::TransportMessage::new(payload)));
-        });
+        return Box::pin(async move { self.receive_frame().await });
     }
 }
 
@@ -274,6 +454,39 @@ impl game_realtime_transport_lib::RealtimeReceiver for WebTransportReceiver {
 pub struct WebTransportSender {
     inner: web_transport_quinn::SendStream,
     _session: web_transport_quinn::Session,
+    max_message_size: usize,
+    send_timeout: std::time::Duration,
+    terminal: bool,
+}
+
+impl WebTransportSender {
+    /// Abruptly resets the reliable send direction with one WebTransport application error code.
+    pub fn abort(&mut self, code: u32) -> Result<(), game_realtime_transport_lib::TransportError> {
+        if self.terminal {
+            return Err(transport_error(game_realtime_transport_lib::TransportErrorKind::Closed, "WebTransport sender is already terminal"));
+        }
+        self.terminal = true;
+        return match self.inner.reset(code) {
+            Ok(()) => {
+                tracing::debug!(target: TRACING_TARGET, code = code, "WebTransport primary send stream aborted");
+                Ok(())
+            },
+            Err(error) => {
+                let mapped = transport_error(game_realtime_transport_lib::TransportErrorKind::Closed, error.to_string());
+                tracing::warn!(target: TRACING_TARGET, code = code, detail = mapped.detail(), "WebTransport primary send stream abort failed");
+                Err(mapped)
+            },
+        };
+    }
+}
+
+impl Drop for WebTransportSender {
+    fn drop(&mut self) {
+        if !self.terminal {
+            let _ = self.inner.reset(STREAM_CANCELLED_ERROR_CODE);
+            self.terminal = true;
+        }
+    }
 }
 
 impl game_realtime_transport_lib::RealtimeSender for WebTransportSender {
@@ -288,6 +501,10 @@ impl game_realtime_transport_lib::RealtimeSender for WebTransportSender {
 
     fn close(&mut self) -> Self::CloseFuture<'_> {
         return Box::pin(async move {
+            if self.terminal {
+                return Err(transport_error(game_realtime_transport_lib::TransportErrorKind::Closed, "WebTransport sender is already terminal"));
+            }
+            self.terminal = true;
             return match self.inner.finish() {
                 Ok(()) => {
                     tracing::debug!(target: TRACING_TARGET, "local WebTransport primary stream close initiated");
@@ -304,36 +521,102 @@ impl game_realtime_transport_lib::RealtimeSender for WebTransportSender {
 
     fn send(&mut self, message: game_realtime_transport_lib::TransportMessage) -> Self::SendFuture<'_> {
         return Box::pin(async move {
+            if self.terminal {
+                return Err(transport_error(
+                    game_realtime_transport_lib::TransportErrorKind::Aborted,
+                    "WebTransport sender is unavailable after close, abort, cancellation or terminal send failure",
+                ));
+            }
             let payload_len = message.len();
-            let frame_header = match frame_header(payload_len) {
+            let frame_header = match frame_header(payload_len, self.max_message_size) {
                 Ok(value) => value,
                 Err(error) => return Err(error),
             };
-            if let Err(error) = self.inner.write_all(&frame_header).await {
-                let mapped = transport_error(game_realtime_transport_lib::TransportErrorKind::Protocol, error.to_string());
-                tracing::warn!(target: TRACING_TARGET, payload_len = payload_len, detail = mapped.detail(), "WebTransport frame header send failed");
-                return Err(mapped);
-            }
-            if let Err(error) = self.inner.write_all(message.as_bytes()).await {
-                let mapped = transport_error(game_realtime_transport_lib::TransportErrorKind::Protocol, error.to_string());
-                tracing::warn!(target: TRACING_TARGET, payload_len = payload_len, detail = mapped.detail(), "WebTransport frame payload send failed");
-                return Err(mapped);
-            }
-            tracing::trace!(target: TRACING_TARGET, payload_len = payload_len, "framed WebTransport payload sent");
-            return Ok(());
+            let send_timeout = self.send_timeout;
+            let mut guard = SendOperationGuard::new(&mut self.inner, &mut self.terminal);
+            let operation = guard.write_frame(&frame_header, message.as_bytes());
+            let result = tokio::time::timeout(send_timeout, operation).await;
+            return match result {
+                Ok(Ok(())) => {
+                    guard.complete();
+                    tracing::trace!(target: TRACING_TARGET, payload_len = payload_len, "framed WebTransport payload sent");
+                    Ok(())
+                },
+                Ok(Err(error)) => {
+                    guard.abort(SEND_FAILURE_ERROR_CODE);
+                    let mapped = map_write_error(error);
+                    tracing::warn!(target: TRACING_TARGET, payload_len = payload_len, kind = %mapped.kind(), detail = mapped.detail(), "WebTransport framed send failed");
+                    Err(mapped)
+                },
+                Err(_) => {
+                    guard.abort(SEND_TIMEOUT_ERROR_CODE);
+                    let mapped = timeout_error("WebTransport framed send", send_timeout);
+                    tracing::warn!(target: TRACING_TARGET, payload_len = payload_len, timeout_ms = duration_millis(send_timeout), "WebTransport framed send timed out under flow control/backpressure");
+                    Err(mapped)
+                },
+            };
         });
     }
 }
 
+struct SendOperationGuard<'a> {
+    stream: &'a mut web_transport_quinn::SendStream,
+    terminal: &'a mut bool,
+    armed: bool,
+}
+
+impl<'a> SendOperationGuard<'a> {
+    fn new(stream: &'a mut web_transport_quinn::SendStream, terminal: &'a mut bool) -> Self {
+        return Self { stream, terminal, armed: true };
+    }
+
+    async fn write_frame(&mut self, header: &[u8; PRIMARY_FRAME_HEADER_SIZE], payload: &[u8]) -> Result<(), web_transport_quinn::WriteError> {
+        if let Err(error) = self.stream.write_all(header).await {
+            return Err(error);
+        }
+        if let Err(error) = self.stream.write_all(payload).await {
+            return Err(error);
+        }
+        return Ok(());
+    }
+
+    fn abort(&mut self, code: u32) {
+        if self.armed {
+            let _ = self.stream.reset(code);
+            *self.terminal = true;
+            self.armed = false;
+        }
+    }
+
+    fn complete(&mut self) {
+        self.armed = false;
+    }
+}
+
+impl Drop for SendOperationGuard<'_> {
+    fn drop(&mut self) {
+        if self.armed {
+            let _ = self.stream.reset(STREAM_CANCELLED_ERROR_CODE);
+            *self.terminal = true;
+            self.armed = false;
+        }
+    }
+}
+
 /// Bound native WebTransport server endpoint that accepts HTTP/3 WebTransport sessions.
 pub struct WebTransportListener {
     server: web_transport_quinn::Server,
     local_addr: std::net::SocketAddr,
+    transport: crate::WebTransportConfig,
 }
 
 impl WebTransportListener {
     /// Binds a native WebTransport server using TLS 1.3 and the configured certificate identity.
     pub fn bind(config: WebTransportServerConfig) -> Result<Self, game_realtime_transport_lib::TransportError> {
+        if let Err(error) = config.transport.validate() {
+            return Err(error);
+        }
+        let transport = config.transport;
         let certificate = web_transport_quinn::quinn::rustls::pki_types::CertificateDer::from(config.identity.certificate_der);
         let private_key = web_transport_quinn::quinn::rustls::pki_types::PrivatePkcs8KeyDer::from(config.identity.private_key_pkcs8_der);
         let private_key = web_transport_quinn::quinn::rustls::pki_types::PrivateKeyDer::Pkcs8(private_key);
@@ -354,7 +637,7 @@ impl WebTransportListener {
             },
         };
         tracing::info!(target: TRACING_TARGET, address = %local_addr, "WebTransport listener bound");
-        return Ok(Self { server, local_addr });
+        return Ok(Self { server, local_addr, transport });
     }
 
     /// Returns the concrete UDP socket address, including an ephemeral port selected by the OS.
@@ -364,6 +647,9 @@ impl WebTransportListener {
     }
 
     /// Accepts one native WebTransport CONNECT request and returns the established session.
+    ///
+    /// Waiting for the next peer remains intentionally unbounded. Once a WebTransport request is surfaced, the final server
+    /// response is bounded by the configured connection deadline.
     pub async fn accept(&mut self) -> Result<WebTransportSession, game_realtime_transport_lib::TransportError> {
         let request = match self.server.accept().await {
             Some(value) => value,
@@ -374,37 +660,60 @@ impl WebTransportListener {
             },
         };
         let peer = request.conn().remote_address();
-        let session = match request.ok().await {
-            Ok(value) => value,
-            Err(error) => {
+        let timeout = self.transport.connect_timeout();
+        let accepted = tokio::time::timeout(timeout, request.ok()).await;
+        let session = match accepted {
+            Ok(Ok(value)) => value,
+            Ok(Err(error)) => {
                 let mapped = transport_error(game_realtime_transport_lib::TransportErrorKind::Accept, error.to_string());
                 tracing::warn!(target: TRACING_TARGET, peer = %peer, detail = mapped.detail(), "WebTransport server handshake failed");
                 return Err(mapped);
             },
+            Err(_) => {
+                let mapped = timeout_error("WebTransport server handshake response", timeout);
+                tracing::warn!(target: TRACING_TARGET, peer = %peer, timeout_ms = duration_millis(timeout), "WebTransport server handshake response timed out");
+                return Err(mapped);
+            },
         };
         tracing::info!(target: TRACING_TARGET, peer = %peer, "WebTransport peer accepted");
-        return Ok(WebTransportSession::new(session));
+        return Ok(WebTransportSession::new(session, self.transport));
     }
 }
 
 /// Establishes one native WebTransport session using an exact SHA-256 certificate pin.
 pub async fn connect(config: &WebTransportClientConfig) -> Result<WebTransportSession, game_realtime_transport_lib::TransportError> {
+    if let Err(error) = config.transport.validate() {
+        return Err(error);
+    }
     let client = match web_transport_quinn::ClientBuilder::new().with_server_certificate_hashes(vec![config.certificate_hash.as_bytes().to_vec()]) {
         Ok(value) => value,
         Err(error) => return Err(invalid_configuration(error.to_string())),
     };
-    let session = match client.connect(config.endpoint.clone()).await {
-        Ok(value) => value,
-        Err(error) => {
+    let timeout = config.transport.connect_timeout();
+    let connected = tokio::time::timeout(timeout, client.connect(config.endpoint.clone())).await;
+    let session = match connected {
+        Ok(Ok(value)) => value,
+        Ok(Err(error)) => {
             let mapped = transport_error(game_realtime_transport_lib::TransportErrorKind::Connect, error.to_string());
             tracing::warn!(target: TRACING_TARGET, endpoint = config.endpoint.as_str(), detail = mapped.detail(), "WebTransport client connection failed");
             return Err(mapped);
         },
+        Err(_) => {
+            let mapped = timeout_error("WebTransport client connect", timeout);
+            tracing::warn!(target: TRACING_TARGET, endpoint = config.endpoint.as_str(), timeout_ms = duration_millis(timeout), "WebTransport client connection timed out");
+            return Err(mapped);
+        },
     };
     tracing::info!(target: TRACING_TARGET, endpoint = config.endpoint.as_str(), peer = %session.remote_address(), "WebTransport client connected");
-    return Ok(WebTransportSession::new(session));
+    return Ok(WebTransportSession::new(session, config.transport));
 }
 
+const FRAME_PROTOCOL_ERROR_CODE: u32 = 0x10;
+const FRAME_TOO_LARGE_ERROR_CODE: u32 = 0x11;
+const SEND_FAILURE_ERROR_CODE: u32 = 0x12;
+const SEND_TIMEOUT_ERROR_CODE: u32 = 0x13;
+const STREAM_CANCELLED_ERROR_CODE: u32 = 0x14;
+
 fn certificate_hash(certificate_der: &[u8]) -> Result<WebTransportCertificateHash, game_realtime_transport_lib::TransportError> {
     let certificate = web_transport_quinn::quinn::rustls::pki_types::CertificateDer::from(certificate_der.to_vec());
     let provider = web_transport_quinn::crypto::default_provider();
@@ -418,13 +727,13 @@ fn certificate_hash(certificate_der: &[u8]) -> Result<WebTransportCertificateHas
     return Ok(WebTransportCertificateHash::from_sha256(bytes));
 }
 
-fn frame_header(payload_len: usize) -> Result<[u8; PRIMARY_FRAME_HEADER_SIZE], game_realtime_transport_lib::TransportError> {
-    if payload_len > PRIMARY_FRAME_MAX_PAYLOAD_SIZE {
-        return Err(message_too_large(payload_len));
+fn frame_header(payload_len: usize, max_message_size: usize) -> Result<[u8; PRIMARY_FRAME_HEADER_SIZE], game_realtime_transport_lib::TransportError> {
+    if payload_len > max_message_size {
+        return Err(message_too_large(payload_len, max_message_size));
     }
     let payload_len = match u32::try_from(payload_len) {
         Ok(value) => value,
-        Err(_) => return Err(message_too_large(payload_len)),
+        Err(_) => return Err(message_too_large(payload_len, max_message_size)),
     };
     return Ok(payload_len.to_be_bytes());
 }
@@ -433,60 +742,57 @@ fn invalid_configuration(detail: impl Into<String>) -> game_realtime_transport_l
     return transport_error(game_realtime_transport_lib::TransportErrorKind::InvalidConfiguration, detail);
 }
 
-fn message_too_large(payload_len: usize) -> game_realtime_transport_lib::TransportError {
+fn map_read_error(error: web_transport_quinn::ReadError) -> game_realtime_transport_lib::TransportError {
+    let kind = match &error {
+        web_transport_quinn::ReadError::Reset(_) => game_realtime_transport_lib::TransportErrorKind::Aborted,
+        web_transport_quinn::ReadError::ClosedStream => game_realtime_transport_lib::TransportErrorKind::Closed,
+        web_transport_quinn::ReadError::SessionError(session) => map_session_error_kind(session),
+        web_transport_quinn::ReadError::InvalidReset(_) | web_transport_quinn::ReadError::IllegalOrderedRead => {
+            game_realtime_transport_lib::TransportErrorKind::Protocol
+        },
+    };
+    return transport_error(kind, error.to_string());
+}
+
+fn map_session_error_kind(error: &web_transport_quinn::SessionError) -> game_realtime_transport_lib::TransportErrorKind {
+    if matches!(error, web_transport_quinn::SessionError::WebTransportError(web_transport_quinn::WebTransportError::Closed(_, _))) {
+        return game_realtime_transport_lib::TransportErrorKind::Closed;
+    }
+    return game_realtime_transport_lib::TransportErrorKind::Io;
+}
+
+fn map_write_error(error: web_transport_quinn::WriteError) -> game_realtime_transport_lib::TransportError {
+    let kind = match &error {
+        web_transport_quinn::WriteError::Stopped(_) => game_realtime_transport_lib::TransportErrorKind::Aborted,
+        web_transport_quinn::WriteError::ClosedStream => game_realtime_transport_lib::TransportErrorKind::Closed,
+        web_transport_quinn::WriteError::SessionError(session) => map_session_error_kind(session),
+        web_transport_quinn::WriteError::InvalidStopped(_) => game_realtime_transport_lib::TransportErrorKind::Protocol,
+    };
+    return transport_error(kind, error.to_string());
+}
+
+fn message_too_large(payload_len: usize, max_message_size: usize) -> game_realtime_transport_lib::TransportError {
     return transport_error(
         game_realtime_transport_lib::TransportErrorKind::MessageTooLarge,
-        format!("WebTransport primary frame payload size {payload_len} exceeds {PRIMARY_FRAME_MAX_PAYLOAD_SIZE} bytes"),
+        format!("WebTransport primary frame payload size {payload_len} exceeds configured maximum {max_message_size} bytes"),
     );
 }
 
-async fn read_frame_payload(stream: &mut web_transport_quinn::RecvStream, payload_len: usize) -> Result<Vec<u8>, game_realtime_transport_lib::TransportError> {
-    let mut payload = vec![0_u8; payload_len];
-    let mut offset = 0_usize;
-    while offset < payload_len {
-        let read = match stream.read(&mut payload[offset..]).await {
-            Ok(Some(value)) => value,
-            Ok(None) => return Err(protocol_error("WebTransport primary stream closed in the middle of a frame payload")),
-            Err(error) => return Err(protocol_error(error.to_string())),
-        };
-        if read == 0 {
-            return Err(protocol_error("WebTransport primary stream returned an empty read in the middle of a frame payload"));
-        }
-        offset += read;
-    }
-    return Ok(payload);
-}
-
-async fn read_frame_payload_len(stream: &mut web_transport_quinn::RecvStream) -> Result<Option<usize>, game_realtime_transport_lib::TransportError> {
-    let mut header = [0_u8; PRIMARY_FRAME_HEADER_SIZE];
-    let mut offset = 0_usize;
-    while offset < PRIMARY_FRAME_HEADER_SIZE {
-        let read = match stream.read(&mut header[offset..]).await {
-            Ok(Some(value)) => value,
-            Ok(None) => {
-                if offset == 0 {
-                    return Ok(None);
-                }
-                return Err(protocol_error("WebTransport primary stream closed in the middle of a frame header"));
-            },
-            Err(error) => return Err(protocol_error(error.to_string())),
-        };
-        if read == 0 {
-            return Err(protocol_error("WebTransport primary stream returned an empty read in the middle of a frame header"));
-        }
-        offset += read;
-    }
-    let payload_len = u32::from_be_bytes(header) as usize;
-    if payload_len > PRIMARY_FRAME_MAX_PAYLOAD_SIZE {
-        return Err(message_too_large(payload_len));
-    }
-    return Ok(Some(payload_len));
-}
-
 fn protocol_error(detail: impl Into<String>) -> game_realtime_transport_lib::TransportError {
     return transport_error(game_realtime_transport_lib::TransportErrorKind::Protocol, detail);
 }
 
+fn timeout_error(operation: &str, timeout: std::time::Duration) -> game_realtime_transport_lib::TransportError {
+    return transport_error(
+        game_realtime_transport_lib::TransportErrorKind::Timeout,
+        format!("{operation} exceeded configured deadline of {} ms", duration_millis(timeout)),
+    );
+}
+
+fn duration_millis(duration: std::time::Duration) -> u128 {
+    return duration.as_millis();
+}
+
 fn transport_error(kind: game_realtime_transport_lib::TransportErrorKind, detail: impl Into<String>) -> game_realtime_transport_lib::TransportError {
     return game_realtime_transport_lib::TransportError::new(kind, detail);
 }