alkdev/alkhttp
6
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(websocket): carry per-cause close reasons on CloseWith (WS-15)

Browse Source 
- WriteMsg::CloseWith now carries (code, reason); the write pump sent
  the hardcoded "text messages not supported" string for the idle
  (1001) and inbound-frame (1011) closes as well as the text (1002)
  close
- close_reason module: canonical per-cause reason strings, shared by
  the close frames and the stream-error diagnostics (they already
  matched for idle/oversize; now single-sourced)
- reason asserts added to the idle-stall and forever-dribble 1001
  tests (reason names the no-chunk-progress cause) and a new text
  frame test asserts 1002 + the distinct text reason

Verification: scripts/verify.sh OK (342 passed), clippy -D warnings
clean, fmt clean
This commit is contained in:
glm-5.3-flash
2026-08-30 21:03:29 +00:00
parent 9ef2352568
commit d6d5509c6d
1 changed files with 112 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/websocket/byte_adapter.rs
+112 -27
View File
@@ -223,8 +223,9 @@ pub use alkcall::channels::wire::MAX_CHUNK_LEN;
pub(crate) enum WriteMsg {
Bytes(Vec<u8>),
/// 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<M, S, F>(
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<M, S, F>(
} 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<M, S, F>(
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<M, S>(
/// 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<M, S>(
mut ws_sink: futures::stream::SplitSink<S, <M as WsFraming>::Msg>,
mut write_rx: futures_mpsc::Receiver<WriteMsg>,
@@ -420,12 +442,9 @@ async fn run_write_pump<M, S>(
let mut pending: Vec<u8> = Vec::new();
while let Some(msg) = write_rx.next().await {
match msg {
WriteMsg::CloseWith(code) => {
WriteMsg::CloseWith(code, reason) => {
let _ = ws_sink
.send(<M as WsFraming>::close_message(
code,
"text messages not supported",
))
.send(<M as WsFraming>::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<u16> = 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<u16> = 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