diff --git a/transport/src/connection.rs b/transport/src/connection.rs index 9c59039..653a822 100644 --- a/transport/src/connection.rs +++ b/transport/src/connection.rs @@ -129,19 +129,30 @@ impl Sender { return Err(CommunicationError::MessageTooLarge); } - use tokio::io::AsyncWriteExt; + let len_bytes = (bytes.len() as u32).to_be_bytes(); + let write_result = async { + stream.write_all(&len_bytes).await?; + stream.write_all(&bytes).await?; + Ok::<(), wtransport::error::StreamWriteError>(()) + }; - timeout(policy.write_timeout, stream.write_u32(bytes.len() as u32)) - .await - .map_err(|_| CommunicationError::StreamError)? - .map_err(|_| CommunicationError::StreamError)?; - - timeout(policy.write_timeout, stream.write_all(&bytes)) - .await - .map_err(|_| CommunicationError::StreamError)? - .map_err(CommunicationError::from)?; - - Ok(()) + match timeout(policy.write_timeout, write_result).await { + Ok(Ok(())) => Ok(()), + Ok(Err(wtransport::error::StreamWriteError::Stopped(code))) => { + log::warn!( + "[Sender] write failed: peer sent STOP_SENDING (error code {code})" + ); + Err(CommunicationError::StreamClosed) + } + Ok(Err(other)) => { + log::warn!("[Sender] write failed: {other}"); + Err(CommunicationError::StreamError) + } + Err(_) => { + log::warn!("[Sender] write timed out (len={})", bytes.len()); + Err(CommunicationError::StreamError) + } + } } fn normalize_send_error(error: CommunicationError) -> CommunicationError { @@ -227,12 +238,23 @@ impl Sender { let mut stream = Self::open_uni_stream(conn, policy).await?; Self::write_frame(&mut stream, data, policy).await?; - timeout(policy.write_timeout, stream.finish()) - .await - .map_err(|_| CommunicationError::StreamError)? - .map_err(|_| CommunicationError::StreamError)?; - - Ok(()) + match timeout(policy.write_timeout, stream.finish()).await { + Ok(Ok(())) => Ok(()), + Ok(Err(wtransport::error::StreamWriteError::Stopped(code))) => { + log::warn!( + "[Sender] finish failed: peer sent STOP_SENDING (error code {code})" + ); + Err(CommunicationError::StreamClosed) + } + Ok(Err(other)) => { + log::warn!("[Sender] finish failed: {other}"); + Err(CommunicationError::StreamError) + } + Err(_) => { + log::warn!("[Sender] finish timed out"); + Err(CommunicationError::StreamError) + } + } } async fn send_close_frame( conn: &Connection, @@ -240,20 +262,35 @@ impl Sender { ) -> Result<(), CommunicationError> { let mut stream = Self::open_uni_stream(conn, policy).await?; - use tokio::io::AsyncWriteExt; - timeout( - policy.write_timeout, - stream.write_u32(policy.close_frame_len), - ) - .await - .map_err(|_| CommunicationError::StreamError)? - .map_err(|_| CommunicationError::StreamError)?; + let len_bytes = policy.close_frame_len.to_be_bytes(); + match timeout(policy.write_timeout, stream.write_all(&len_bytes)).await { + Ok(Ok(())) => {} + Ok(Err(wtransport::error::StreamWriteError::Stopped(code))) => { + log::warn!( + "[Sender] close frame write failed: peer sent STOP_SENDING (error code {code})" + ); + } + Ok(Err(other)) => { + log::warn!("[Sender] close frame write failed: {other}"); + } + Err(_) => { + log::warn!("[Sender] close frame write timed out"); + } + } - if let Err(e) = timeout(policy.write_timeout, stream.finish()) - .await - .map_err(|_| CommunicationError::StreamError)? - { - log::warn!("[Sender] close frame finish failed: {e}"); + match timeout(policy.write_timeout, stream.finish()).await { + Ok(Ok(())) => {} + Ok(Err(wtransport::error::StreamWriteError::Stopped(code))) => { + log::warn!( + "[Sender] close frame finish failed: peer sent STOP_SENDING (error code {code})" + ); + } + Ok(Err(other)) => { + log::warn!("[Sender] close frame finish failed: {other}"); + } + Err(_) => { + log::warn!("[Sender] close frame finish timed out"); + } } Ok(()) @@ -318,10 +355,23 @@ impl Sender { let _send_lock = self.send_guard.lock().await; let mut stream_opt = self.stream_guard.lock().await; if let Some(mut stream) = stream_opt.take() { - timeout(self.policy.write_timeout, stream.finish()) - .await - .map_err(|_| CommunicationError::StreamError)? - .map_err(|_| CommunicationError::StreamError)?; + match timeout(self.policy.write_timeout, stream.finish()).await { + Ok(Ok(())) => {} + Ok(Err(wtransport::error::StreamWriteError::Stopped(code))) => { + log::warn!( + "[Sender] finish_stream: peer sent STOP_SENDING (error code {code})" + ); + return Err(CommunicationError::StreamClosed); + } + Ok(Err(other)) => { + log::warn!("[Sender] finish_stream failed: {other}"); + return Err(CommunicationError::StreamError); + } + Err(_) => { + log::warn!("[Sender] finish_stream timed out"); + return Err(CommunicationError::StreamError); + } + } } Ok(()) } @@ -345,7 +395,12 @@ impl Sender { if let Some(mut stream) = stream_guard.lock().await.take() { match timeout(policy.write_timeout, stream.finish()).await { Ok(Ok(())) => {} - Ok(Err(e)) => log::warn!("[Sender] persistent stream finish failed: {e}"), + Ok(Err(wtransport::error::StreamWriteError::Stopped(code))) => log::warn!( + "[Sender] persistent stream finish failed: peer sent STOP_SENDING (error code {code})" + ), + Ok(Err(e)) => { + log::warn!("[Sender] persistent stream finish failed: {e}") + } Err(_) => log::warn!("[Sender] persistent stream finish timed out"), } } @@ -500,46 +555,67 @@ impl Receiver { stream: &mut wtransport::RecvStream, policy: &Policy, ) -> Result { - use std::io::ErrorKind; - use tokio::io::AsyncReadExt; + use wtransport::error::{StreamReadError, StreamReadExactError}; - let mut attempts = 0; - let len = loop { - match stream.read_u32().await { - Ok(len) => break len, - Err(e) => { - if e.kind() == ErrorKind::Interrupted && attempts < 3 { - attempts += 1; - tokio::time::sleep(std::time::Duration::from_millis(10)).await; - continue; - } - if e.kind() == ErrorKind::UnexpectedEof { - return Ok(ReceivedFrame::Idle); - } - log::warn!("[Receiver] read_u32 failed: {e}"); - return Err(CommunicationError::StreamError); - } + let mut len_buf = [0u8; 4]; + match timeout(policy.read_timeout, stream.read_exact(&mut len_buf)).await { + Ok(Ok(())) => {} + Ok(Err(StreamReadExactError::FinishedEarly(0))) => { + return Ok(ReceivedFrame::Idle); } - }; + Ok(Err(StreamReadExactError::FinishedEarly(n))) => { + log::warn!( + "[Receiver] length-prefix read ended early ({n}/4 bytes): stream closed by peer" + ); + return Err(CommunicationError::StreamError); + } + Ok(Err(StreamReadExactError::Read(StreamReadError::Reset(code)))) => { + log::warn!( + "[Receiver] length-prefix read failed: peer sent RESET_STREAM (error code {code})" + ); + return Err(CommunicationError::StreamError); + } + Ok(Err(other)) => { + log::warn!("[Receiver] length-prefix read failed: {other}"); + return Err(CommunicationError::StreamError); + } + Err(_) => { + log::warn!("[Receiver] length-prefix read timed out"); + return Err(CommunicationError::StreamError); + } + } + let len = u32::from_be_bytes(len_buf); if len == policy.close_frame_len { return Ok(ReceivedFrame::ClosedByPeer); } - let len = len as usize; + let len_usize = len as usize; if len as u64 > policy.max_message_size { return Err(CommunicationError::MessageTooLarge); } - let mut buf = vec![0u8; len]; + let mut buf = vec![0u8; len_usize]; match timeout(policy.read_timeout, stream.read_exact(&mut buf)).await { Ok(Ok(())) => {} - Ok(Err(e)) => match timeout(policy.read_timeout, stream.read_exact(&mut buf)).await { - Ok(Ok(())) => {} - _ => return Err(e.into()), - }, + Ok(Err(StreamReadExactError::FinishedEarly(n))) => { + log::warn!( + "[Receiver] body read ended early ({n}/{len_usize} bytes): stream closed by peer" + ); + return Err(CommunicationError::StreamError); + } + Ok(Err(StreamReadExactError::Read(StreamReadError::Reset(code)))) => { + log::warn!( + "[Receiver] body read failed: peer sent RESET_STREAM (error code {code})" + ); + return Err(CommunicationError::StreamError); + } + Ok(Err(other)) => { + log::warn!("[Receiver] body read failed: {other}"); + return Err(CommunicationError::StreamError); + } Err(_) => { - log::warn!("[Receiver] read_exact timed out (len={})", len); + log::warn!("[Receiver] body read timed out (len={len_usize})"); return Err(CommunicationError::StreamError); } } diff --git a/wasm/src/transport.rs b/wasm/src/transport.rs index 1bdf39c..4708b47 100644 --- a/wasm/src/transport.rs +++ b/wasm/src/transport.rs @@ -10,6 +10,52 @@ use crate::frame::parse_frame_value; const CLOSE_FRAME_LEN: u32 = u32::MAX; +/// Inspect a JS error value for a WebTransport stream error and log the +/// `streamErrorCode` carried by STOP_SENDING / RESET_STREAM. +/// +/// The browser's WebTransport API rejects write/close/read promises with a +/// `WebTransportError` whose `source` is `"stream"` and whose +/// `streamErrorCode` is the application error code from the peer's +/// STOP_SENDING (for send streams) or RESET_STREAM (for receive streams). +/// Per draft-ietf-webtrans-http3-15 ยง4.4, a WebTransport application MUST +/// provide an error code for those operations, so it is always present on +/// stream-level errors. +fn log_webtransport_error(error: &JsValue, context: &str) { + let source = js_sys::Reflect::get(error, &JsValue::from_str("source")) + .ok() + .and_then(|v| v.as_string()); + let stream_error_code = js_sys::Reflect::get(error, &JsValue::from_str("streamErrorCode")) + .ok() + .and_then(|v| v.as_f64()); + let message = error + .as_string() + .or_else(|| { + js_sys::Reflect::get(error, &JsValue::from_str("message")) + .ok() + .and_then(|v| v.as_string()) + }) + .unwrap_or_else(|| format!("{:?}", error)); + + let formatted = match (&source, stream_error_code) { + (Some(src), Some(code)) => format!( + "[WasmTransport] {context}: WebTransportError source={src} \ + streamErrorCode={code} ({message})" + ), + (Some(src), None) => { + format!("[WasmTransport] {context}: WebTransportError source={src} ({message})") + } + (None, _) => format!("[WasmTransport] {context}: {message}"), + }; + + if let Ok(console) = js_sys::Reflect::get(&js_sys::global(), &JsValue::from_str("console")) { + if let Ok(warn) = js_sys::Reflect::get(&console, &JsValue::from_str("warn")) + .and_then(|f| f.dyn_into::()) + { + let _ = warn.call1(&console, &JsValue::from_str(&formatted)); + } + } +} + /// Given a `SendStream` (old API with `.writable` or new API where stream IS a WritableStream), /// return the object to call `.getWriter()` on. fn resolve_stream_writable(send_stream: &JsValue) -> Result { @@ -169,7 +215,10 @@ impl WasmTransport { let write_promise = write_fn .call1(&writer_val, &chunk) .map_err(|e| js_error(&format!("write failed: {:?}", e)))?; - JsFuture::from(write_promise.unchecked_into::()).await?; + if let Err(e) = JsFuture::from(write_promise.unchecked_into::()).await { + log_webtransport_error(&e, "send_frame write"); + return Err(e); + } let close_fn = js_sys::Reflect::get(&writer_val, &JsValue::from_str("close")) .map_err(|_| js_error("missing close"))? @@ -178,7 +227,10 @@ impl WasmTransport { let close_promise = close_fn .call0(&writer_val) .map_err(|e| js_error(&format!("close failed: {:?}", e)))?; - JsFuture::from(close_promise.unchecked_into::()).await?; + if let Err(e) = JsFuture::from(close_promise.unchecked_into::()).await { + log_webtransport_error(&e, "send_frame close"); + return Err(e); + } Ok(()) } @@ -218,7 +270,10 @@ impl WasmTransport { .unchecked_into::(), ) .await - .map_err(|e| js_error(&format!("accept stream failed: {:?}", e)))?; + .map_err(|e| { + log_webtransport_error(&e, "open_next_stream accept"); + js_error(&format!("accept stream failed: {:?}", e)) + })?; let done = js_sys::Reflect::get(&result, &JsValue::from_str("done")) .ok() @@ -260,7 +315,10 @@ impl WasmTransport { .unchecked_into::(), ) .await - .map_err(|e| js_error(&format!("read failed: {:?}", e)))?; + .map_err(|e| { + log_webtransport_error(&e, "read_chunk"); + js_error(&format!("read failed: {:?}", e)) + })?; let done = js_sys::Reflect::get(&result, &JsValue::from_str("done")) .ok()