diff --git a/src/websocket/byte_adapter.rs b/src/websocket/byte_adapter.rs index 66d9f16..135238a 100644 --- a/src/websocket/byte_adapter.rs +++ b/src/websocket/byte_adapter.rs @@ -223,8 +223,9 @@ pub use alkcall::channels::wire::MAX_CHUNK_LEN; pub(crate) enum WriteMsg { Bytes(Vec), - /// Close with the given code (e.g. the 1002 text-rejection). - CloseWith(u16), + /// Close with the given code and reason (e.g. the 1002 + /// text-rejection). + CloseWith(u16, &'static str), } /// The write-pump failure channel (WS-04/HY-09, WS-05): a shared slot @@ -248,6 +249,20 @@ fn send_write_error(slot: &WriteErrorSlot, reason: &'static str) { } } +/// The close-frame reason strings, one distinct per cause (WS-15): the +/// numeric code alone collides across causes (the size-cap close and +/// the write-stall close share 1011), and an operator reading a wire +/// capture should be able to tell them apart without the stream error +/// channel. Sent in both the close frame the peer receives and the +/// stream-error diagnostic; `reason_for` maps a bare close code to its +/// canonical default for the fall-through arms. +pub(crate) mod close_reason { + pub(crate) const IDLE_READ_TIMEOUT: &str = + "connection made no inbound chunk progress past the read timeout"; + pub(crate) const TEXT_NOT_SUPPORTED: &str = "text messages not supported"; + pub(crate) const INBOUND_FRAME_REJECTED: &str = "inbound frame rejected (size cap)"; +} + /// WS-protocol adapter for one socket flavor: the only places the axum /// and tungstenite message types differ. The generic pump /// ([`run_read_pump`] / [`run_write_pump`]) is written against this @@ -335,13 +350,13 @@ async fn run_read_pump( Some(budget) => match tokio::time::timeout(budget, ws_stream.next()).await { Ok(msg) => msg, Err(_elapsed) => { - send_write_error( - &write_error_slot_for_read, - "connection made no inbound chunk progress past the read timeout", - ); + send_write_error(&write_error_slot_for_read, close_reason::IDLE_READ_TIMEOUT); let _ = write_tx_for_read .clone() - .send(WriteMsg::CloseWith(WS_GOING_AWAY)) + .send(WriteMsg::CloseWith( + WS_GOING_AWAY, + close_reason::IDLE_READ_TIMEOUT, + )) .await; break; } @@ -363,7 +378,10 @@ async fn run_read_pump( } else if M::is_text(&m) { let _ = write_tx_for_read .clone() - .send(WriteMsg::CloseWith(WS_PROTOCOL_ERROR)) + .send(WriteMsg::CloseWith( + WS_PROTOCOL_ERROR, + close_reason::TEXT_NOT_SUPPORTED, + )) .await; break; } else if M::is_close(&m) { @@ -373,11 +391,14 @@ async fn run_read_pump( Err(()) => { send_write_error( &write_error_slot_for_read, - "inbound frame rejected (size cap)", + close_reason::INBOUND_FRAME_REJECTED, ); let _ = write_tx_for_read .clone() - .send(WriteMsg::CloseWith(WS_INTERNAL_ERROR)) + .send(WriteMsg::CloseWith( + WS_INTERNAL_ERROR, + close_reason::INBOUND_FRAME_REJECTED, + )) .await; break; } @@ -406,9 +427,10 @@ async fn fail_write_pump( /// bytes for complete chunks (8-byte header, length validated against /// `MAX_CHUNK_LEN` — WS-04/HY-09), byte-caps the accumulator /// (`PENDING_BUFFER_CAP` — WS-05), and emits one WS binary message per -/// `WS_MESSAGE_CAP` piece. Ends with a WS Close after the queue /// drains; `CloseWith` requests (`WriteMsg::CloseWith`) map to a close -/// frame with that code. +/// frame with the requested code and its per-cause reason (WS-15) — +/// the code alone is ambiguous across causes (idle 1001 vs protocol +/// 1002 vs internal 1011 arms), so the pump never invents one. async fn run_write_pump( mut ws_sink: futures::stream::SplitSink::Msg>, mut write_rx: futures_mpsc::Receiver, @@ -420,12 +442,9 @@ async fn run_write_pump( let mut pending: Vec = Vec::new(); while let Some(msg) = write_rx.next().await { match msg { - WriteMsg::CloseWith(code) => { + WriteMsg::CloseWith(code, reason) => { let _ = ws_sink - .send(::close_message( - code, - "text messages not supported", - )) + .send(::close_message(code, reason)) .await; break; } @@ -1206,8 +1225,64 @@ mod tests { assert!(saw_close, "the shutdown path emitted a WS Close frame"); } + /// WS-15 acceptance (text arm): a text frame triggers the 1002 + /// protocol-error close whose reason names the text cause — + /// distinct from the idle (1001) and oversize (1011) reasons the + /// other tests assert. + #[tokio::test] + async fn tungstenite_text_close_carries_protocol_reason() { + let (client_io, server_io) = tokio::io::duplex(1 << 16); + let ws = tokio_tungstenite::WebSocketStream::from_raw_socket( + client_io, + tokio_tungstenite::tungstenite::protocol::Role::Client, + None, + ) + .await; + let (_stream, pumps) = split_tungstenite_to_bytes(ws); + + let peer = tokio_tungstenite::WebSocketStream::from_raw_socket( + server_io, + tokio_tungstenite::tungstenite::protocol::Role::Server, + None, + ) + .await; + let (mut peer_sink, mut peer_stream) = peer.split(); + use futures::{SinkExt, StreamExt}; + + let mut eof_rx = pumps.read_eof(); + peer_sink + .send(tokio_tungstenite::tungstenite::Message::Text( + "text frame".into(), + )) + .await + .expect("text write accepted"); + + let (close, eof_fired) = tokio::time::timeout(std::time::Duration::from_secs(5), async { + let close: Option<(u16, String)> = loop { + match peer_stream.next().await { + Some(Ok(tokio_tungstenite::tungstenite::Message::Close(cf))) => { + break cf.map(|f| (u16::from(f.code), f.reason.to_string())) + } + Some(Ok(_)) => continue, + Some(Err(_)) | None => break None, + } + }; + let _ = eof_rx.changed().await; + (close, *eof_rx.borrow()) + }) + .await + .expect("read pump ends after the text frame (close requested)"); + + let (code, reason) = close.expect("close frame with code + reason"); + assert_eq!(code, WS_PROTOCOL_ERROR, "text frame closed with 1002"); + assert_eq!( + reason, "text messages not supported", + "close reason names the text cause, got {reason:?}" + ); + assert!(eof_fired, "EOF signal fired for the from_wss machinery"); + } + /// WS-01 acceptance: a dribbling peer cannot park the read pump - /// past the knob. The idle-read timeout fires with no inbound /// message inside the window: the pump sends the 1001 GoingAway /// close to the peer, ends, and fires the EOF watch signal — the /// from_wss monitor/sweep input — so the demux/channel teardown @@ -1236,14 +1311,15 @@ mod tests { // The peer sends nothing (the stall). The read pump must end // within the knob (+ slack), not hang: observe both the EOF // signal the from_wss machinery consumes and the 1001 close - // frame the peer receives. + // frame the peer receives. WS-15: the close reason names the + // idle-read cause, not a leftover default. let mut eof_rx = pumps.read_eof(); let (close_seen, eof_fired) = tokio::time::timeout(std::time::Duration::from_secs(5), async { - let close: Option = loop { + let close: Option<(u16, String)> = loop { match peer_stream.next().await { Some(Ok(tokio_tungstenite::tungstenite::Message::Close(cf))) => { - break cf.map(|f| u16::from(f.code)) + break cf.map(|f| (u16::from(f.code), f.reason.to_string())) } Some(Ok(_)) => continue, Some(Err(_)) | None => break None, @@ -1255,9 +1331,14 @@ mod tests { .await .expect("stalled connection must be torn down within the deadline"); + let (code, reason) = close_seen.expect("close frame with code + reason"); + assert_eq!( + code, WS_GOING_AWAY, + "peer received the 1001 GoingAway close" + ); assert!( - close_seen == Some(WS_GOING_AWAY), - "peer received the 1001 GoingAway close, got {close_seen:?}" + reason.contains("no inbound chunk progress"), + "close reason names the idle-read cause, got {reason:?}" ); assert!(eof_fired, "EOF signal fired for the from_wss machinery"); } @@ -1327,10 +1408,10 @@ mod tests { }); let outcome = tokio::time::timeout(std::time::Duration::from_secs(10), async { - let close: Option = loop { + let close: Option<(u16, String)> = loop { match peer_stream.next().await { Some(Ok(tokio_tungstenite::tungstenite::Message::Close(cf))) => { - break cf.map(|f| u16::from(f.code)) + break cf.map(|f| (u16::from(f.code), f.reason.to_string())) } Some(Ok(_)) => continue, Some(Err(_)) | None => break None, @@ -1342,11 +1423,15 @@ mod tests { .await .expect("dribble must hit the progress deadline within 10 s"); dribble.abort(); + let (code, reason) = outcome.expect("close frame with code + reason"); assert_eq!( - outcome, - Some(WS_GOING_AWAY), + code, WS_GOING_AWAY, "the forever-dribble is evicted with 1001 despite arriving messages" ); + assert!( + reason.contains("no inbound chunk progress"), + "close reason names the idle-read cause, got {reason:?}" + ); } /// Progress semantics, survivor side (WS-13 accept 2): a session