[Fix] Connections

mtp-webserver/src/h3.rs

@@ -143,6 +143,11 @@ pub(crate) async fn run_driver(
                                     return;
                                 }
                             };
+                            tracing::debug!(
+                                remote = %remote_addr,
+                                session_id = ?session.session_id(),
+                                "accepted WebTransport MTP session"
+                            );
                             tokio::spawn(run_session_requests(
                                 session.clone(),
                                 router.clone(),

mtp-webserver/src/transport.rs

@@ -32,6 +32,8 @@ pub struct H3TransportSender {
 
 pub struct H3TransportReceiver {
     stream: H3RecvStream,
+    quinn: quinn::Connection,
+    read_exact_calls: u64,
 }
 
 impl H3TransportConnection {
@@ -76,18 +78,42 @@ impl TransportSendStream for H3TransportSender {
 #[async_trait::async_trait]
 impl TransportRecvStream for H3TransportReceiver {
     async fn read_exact(&mut self, buf: &mut [u8]) -> Result<(), CommunicationError> {
+        let first_read = self.read_exact_calls == 0;
+        self.read_exact_calls += 1;
         self.stream
             .read_exact(buf)
             .await
-            .map(|_| ())
+            .map(|_| {
+                if first_read {
+                    tracing::debug!(
+                        remote = %self.quinn.remote_address(),
+                        bytes = buf.len(),
+                        "received first bytes from WebTransport MTP stream"
+                    );
+                }
+            })
             .map_err(|error| {
-                if error.kind() == std::io::ErrorKind::UnexpectedEof {
+                if error.kind() == std::io::ErrorKind::UnexpectedEof
-                    // Browser control frames are sent on one-frame uni streams.
+                    || self.quinn.close_reason().is_some()
-                    // Reaching FIN while looking for another frame is normal.
+                {
+                    /*
+                     * Reaching FIN, or losing the enclosing QUIC connection,
+                     * is a normal stream-closure path. Do not turn it into a
+                     * frame-header failure and close the connection again.
+                     */
                     return CommunicationError::StreamClosed;
                 }
-                error!("[mtp-webserver] receive stream read_exact failed ({} bytes): {error}", buf.len());
+                error!(
-                tracing::warn!(len = buf.len(), %error, "WebTransport receive stream read_exact failed");
+                    "[mtp-webserver] receive stream read_exact failed ({} bytes): {error}",
+                    buf.len()
+                );
+                tracing::warn!(
+                    remote = %self.quinn.remote_address(),
+                    first_read,
+                    len = buf.len(),
+                    %error,
+                    "WebTransport receive stream read_exact failed"
+                );
                 CommunicationError::StreamError
             })
     }
@@ -101,6 +127,9 @@ impl TransportRecvStream for H3TransportReceiver {
                 Ok(Some(buf))
             }
             Err(error) => {
+                if self.quinn.close_reason().is_some() {
+                    return Err(CommunicationError::StreamClosed);
+                }
                 error!(
                     "[mtp-webserver] receive stream read failed (max {} bytes): {error}",
                     max
@@ -167,10 +196,27 @@ impl TransportConnection for H3TransportConnection {
         loop {
             match self.session.accept_uni().await {
                 Ok(Some((id, stream))) if id == self.session.session_id() => {
-                    return Ok(H3TransportReceiver { stream });
+                    let stream_id = h3::quic::RecvStream::recv_id(&stream);
+                    tracing::debug!(
+                        remote = %self.quinn.remote_address(),
+                        session_id = ?self.session.session_id(),
+                        stream_id = ?stream_id,
+                        "accepted WebTransport MTP receive stream"
+                    );
+                    return Ok(H3TransportReceiver {
+                        stream,
+                        quinn: self.quinn.clone(),
+                        read_exact_calls: 0,
+                    });
                 }
-                Ok(Some(_)) => {
+                Ok(Some((stream_session_id, _stream))) => {
                     consecutive_errors = 0;
+                    tracing::debug!(
+                        remote = %self.quinn.remote_address(),
+                        session_id = ?self.session.session_id(),
+                        stream_session_id = ?stream_session_id,
+                        "ignored WebTransport receive stream belonging to another session"
+                    );
                     continue;
                 }
                 Ok(None) => return Err(CommunicationError::StreamClosed),
@@ -306,7 +352,18 @@ async fn accept_web_connection_inner(
                 connection_id,
             },
         )
-        .await?;
+        .await;
+    #[cfg(feature = "crypto")]
+    if let Err(error) = &result {
+        tracing::warn!(
+            remote = %remote_addr,
+            connection_id,
+            %error,
+            "WebTransport MTP handshake failed"
+        );
+    }
+    #[cfg(feature = "crypto")]
+    let result = result?;
     #[cfg(not(feature = "crypto"))]
     let result = engine.accept(&sender, &receiver).await?;
 

transport/src/generic_connection.rs

@@ -333,6 +333,18 @@ impl<C: TransportConnection> GenericReceiver<C> {
                                 break 'stream;
                             }
                             Err(_) => {
+                                if frames == 0 {
+                                    tracing::warn!(
+                                        timeout = ?policy.read_timeout,
+                                        "MTP receive stream timed out before its first complete frame"
+                                    );
+                                } else {
+                                    tracing::debug!(
+                                        frames,
+                                        timeout = ?policy.read_timeout,
+                                        "MTP receive stream idle timeout"
+                                    );
+                                }
                                 break;
                             }
                         }
@@ -376,6 +388,9 @@ impl<C: TransportConnection> GenericReceiver<C> {
                             )
                             .await;
                             if !matches!(&body_read, Ok(Ok(()))) {
+                                if matches!(&body_read, Ok(Err(CommunicationError::StreamClosed))) {
+                                    break 'stream;
+                                }
                                 tracing::warn!(
                                     pipe_chunk_len = chunk_len,
                                     ?body_read,

wasm/src/client/authentication.rs

@@ -8,6 +8,14 @@ use crate::config::ConnectionConfig;
 use crate::error::js_error;
 use crate::transport::WasmTransport;
 
+fn server_rejection_message(outcome: &CommunicationValue) -> Option<&str> {
+    (outcome.get_data(DataType::Connected) == Some(&DataValue::BoolFalse)).then(|| {
+        outcome
+            .get_str(DataType::ErrorMessage)
+            .unwrap_or("host rejected the connection")
+    })
+}
+
 #[wasm_bindgen]
 #[allow(deprecated)]
 impl WasmClient {
@@ -79,6 +87,18 @@ impl WasmClient {
                         .unwrap_or("host does not support this protocol version"),
                 ));
             }
+
+            // Generic host rejections are IdentificationResponse frames with
+            // Connected=false. They intentionally do not carry a negotiated
+            // Version because negotiation never completed. Check this before
+            // reading Version, otherwise a useful server error such as an
+            // authentication timeout is reported as the misleading
+            // "host omitted a valid negotiated protocol version".
+            if let Some(message) = server_rejection_message(&outcome) {
+                self.set_state_if_current(generation, ConnectionState::Disconnected);
+                return Err(js_error(message));
+            }
+
             let negotiated_version = match outcome.get_data(DataType::Version) {
                 Some(DataValue::Str(version)) => mtp_codec::Version::parse(version)
                     .ok_or_else(|| js_error("host omitted a valid negotiated protocol version"))?,
@@ -628,3 +648,24 @@ impl WasmClient {
         Ok(server_challenge)
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::server_rejection_message;
+    use mtp_codec::{CommunicationType, CommunicationValue, DataType, DataValue};
+
+    #[test]
+    fn reports_rejection_reason_without_a_negotiated_version() {
+        let response = CommunicationValue::new(CommunicationType::IdentificationResponse)
+            .add_typed_default(DataType::Connected, DataValue::BoolFalse)
+            .add_typed_default(
+                DataType::ErrorMessage,
+                DataValue::Str("authentication handshake timed out".into()),
+            );
+
+        assert_eq!(
+            server_rejection_message(&response),
+            Some("authentication handshake timed out")
+        );
+    }
+}