From fcec43d44b63e4f3d3390bb86efa5bcbcb644467 Mon Sep 17 00:00:00 2001 From: Steve Biedermann Date: Tue, 3 Nov 2026 17:05:29 +0100 Subject: [PATCH] fix: complete the CLI WebSocket close handshake Close the client explicitly on REPL exit and wait for the peer with a bounded deadline. Cover quit, exit, EOF, pending notifications, and incomplete handshakes without suppressing genuine protocol failures. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- CHANGELOG.md | 5 ++ crates/runa-cli/src/main.rs | 16 +++++- crates/runa-client/src/lib.rs | 98 +++++++++++++++++++++++++++++++++ docs/protocol.md | 4 ++ vscode/integration/repl.test.ts | 32 +++++++---- vscode/integration/server.ts | 12 +++- 6 files changed, 151 insertions(+), 16 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index ed54baf..539f02e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,11 @@ All notable changes to this project will be documented in this file. ## [Unreleased] +### Fixed + +- Complete the WebSocket close handshake when the CLI exits, preventing normal + disconnects from producing a connection-reset protocol error on the server. + ### Added - Mutable image-owned source files organized beneath `source/`, with independent diff --git a/crates/runa-cli/src/main.rs b/crates/runa-cli/src/main.rs index 91a667f..9d4ebc6 100644 --- a/crates/runa-cli/src/main.rs +++ b/crates/runa-cli/src/main.rs @@ -194,7 +194,10 @@ fn main() -> Result<(), Box> { println!("Runa — Rune Actors"); println!("Connected to {address}. The image remains running when you quit."); println!("Type `help` for commands. Use Up/Down for history and Tab for completion."); - run_interactive_repl(&mut server)?; + let result = run_interactive_repl(&mut server); + let close_result = server.close(); + result?; + close_result?; Ok(()) } @@ -928,6 +931,13 @@ mod tests { } let mut socket = tungstenite::accept_hdr(stream, DebugProtocol).unwrap(); while let Ok(message) = socket.read() { + if matches!(message, tungstenite::Message::Close(_)) { + assert!(matches!( + socket.read(), + Err(tungstenite::Error::ConnectionClosed) + )); + break; + } let Some(text) = message.to_text().ok() else { break; }; @@ -957,7 +967,7 @@ mod tests { .find_map(|line| line.strip_prefix("runa> actor = @")) .ok_or("spawn did not return an actor reference")?; assert_eq!(actor_ref.len(), 21); - drop(client); + client.close()?; let mut client = WebSocketClient::connect(address)?; let mut output = Vec::new(); @@ -980,7 +990,7 @@ mod tests { }; assert_eq!(snapshot.state, Value::Integer(12)); assert_eq!(snapshot.status, runa_protocol::ActorStatusDto::Idle); - drop(client); + client.close()?; worker.join().expect("WebSocket image worker completed"); Ok(()) } diff --git a/crates/runa-client/src/lib.rs b/crates/runa-client/src/lib.rs index 587007e..341da5f 100644 --- a/crates/runa-client/src/lib.rs +++ b/crates/runa-client/src/lib.rs @@ -90,6 +90,41 @@ impl WebSocketClient { result } + /// Complete the WebSocket close handshake without retrying failed connections. + pub fn close(&mut self) -> io::Result<()> { + if self.disconnected { + return Ok(()); + } + self.disconnected = true; + let result = self.close_inner(); + if result.is_err() { + let _ = self.socket.get_ref().shutdown(Shutdown::Both); + } + result + } + + fn close_inner(&mut self) -> io::Result<()> { + let timeout = Duration::from_secs(5); + let deadline = Instant::now() + timeout; + self.socket.get_ref().set_write_timeout(Some(timeout))?; + self.socket.close(None).map_err(io::Error::other)?; + loop { + let remaining = deadline + .checked_duration_since(Instant::now()) + .filter(|duration| !duration.is_zero()) + .ok_or_else(|| { + io::Error::new(io::ErrorKind::TimedOut, "WebSocket close timed out") + })?; + self.socket.get_ref().set_read_timeout(Some(remaining))?; + self.socket.get_ref().set_write_timeout(Some(remaining))?; + match self.socket.read() { + Ok(_) => {} + Err(tungstenite::Error::ConnectionClosed) => return Ok(()), + Err(error) => return Err(io::Error::other(error)), + } + } + } + fn request_inner(&mut self, command: ProtocolCommand) -> Result { self.next_request_id = self .next_request_id @@ -255,6 +290,69 @@ mod tests { (address, peer) } + #[test] + fn websocket_client_completes_close_handshake_with_pending_notifications() { + let (address, peer) = websocket_peer(|mut socket, request| { + socket + .send(Message::Text( + serde_json::to_string(&Response { + version: PROTOCOL_VERSION, + request_id: request.request_id, + result: Ok(Reply::Actors(Vec::new())), + }) + .unwrap() + .into(), + )) + .unwrap(); + socket + .send(Message::Text( + serde_json::to_string(&Notification { + version: PROTOCOL_VERSION, + event: ImageEvent::ImageChanged { revision: 1 }, + }) + .unwrap() + .into(), + )) + .unwrap(); + assert!(matches!(socket.read().unwrap(), Message::Close(_))); + // Reading drives the queued close reply and closes the server transport. + assert!(matches!( + socket.read(), + Err(tungstenite::Error::ConnectionClosed) + )); + }); + let mut client = WebSocketClient::connect(address).unwrap(); + client.request(ProtocolCommand::ListActors).unwrap(); + client.close().unwrap(); + client.close().unwrap(); + assert!(matches!(client.request(ProtocolCommand::ListActors), + Err(ClientError::Io(error)) if error.kind() == io::ErrorKind::NotConnected)); + peer.join().unwrap(); + } + + #[test] + fn websocket_client_reports_an_incomplete_close_handshake() { + let (address, peer) = websocket_peer(|mut socket, request| { + socket + .send(Message::Text( + serde_json::to_string(&Response { + version: PROTOCOL_VERSION, + request_id: request.request_id, + result: Ok(Reply::Actors(Vec::new())), + }) + .unwrap() + .into(), + )) + .unwrap(); + assert!(matches!(socket.read().unwrap(), Message::Close(_))); + // Deliberately drop the transport without flushing the close reply. + }); + let mut client = WebSocketClient::connect(address).unwrap(); + client.request(ProtocolCommand::ListActors).unwrap(); + assert!(client.close().is_err()); + peer.join().unwrap(); + } + #[test] fn websocket_client_validates_notifications_without_losing_response_correlation() { let (address, peer) = websocket_peer(|mut socket, request| { diff --git a/docs/protocol.md b/docs/protocol.md index 3039f41..5f35b33 100644 --- a/docs/protocol.md +++ b/docs/protocol.md @@ -154,6 +154,10 @@ the actor tree and open snapshot documents. Manual refresh remains available. ## WebSocket transport and notifications +On `quit`, `exit`, or end-of-input, the CLI completes the WebSocket close +handshake with a 5-second deadline before releasing the connection. A failed +handshake is reported rather than treating an abrupt reset as a normal close. + Connect to `ws://127.0.0.1:3737/` (or a configured numeric loopback endpoint) and offer the `runa-debug` WebSocket subprotocol. The server requires and echoes that subprotocol. Plain TCP JSON Lines is no longer accepted by the network diff --git a/vscode/integration/repl.test.ts b/vscode/integration/repl.test.ts index 8701c5b..6caeed1 100644 --- a/vscode/integration/repl.test.ts +++ b/vscode/integration/repl.test.ts @@ -61,6 +61,8 @@ test( ); const [code] = await exited; assert.equal(code, 0, stderr); + assert.equal(stderr, ""); + assert.doesNotMatch(image.diagnostics(), /protocol error|reset without closing handshake/i); assert.doesNotMatch(stdout, /error:/i); await changed; assert.ok(revisions.length > 0); @@ -70,16 +72,24 @@ test( assert.deepEqual(actor.state, { type: "integer", data: 12n }); assert.deepEqual(actor.queued, []); assert.equal(actor.reserved, null); - const overridden = spawn(executable, ["--connect", image.address], { - env: { ...process.env, RUNA_CONNECT: "invalid" }, - stdio: ["pipe", "pipe", "pipe"], - }); - t.after(() => { - if (overridden.exitCode === null && overridden.signalCode === null) overridden.kill(); - }); - const overriddenExit = once(overridden, "exit"); - overridden.stdin.end("actors\nquit\n"); - const [overrideCode] = await overriddenExit; - assert.equal(overrideCode, 0); + for (const input of ["actors\nquit\n", "exit\n", ""]) { + const overridden = spawn(executable, ["--connect", image.address], { + env: { ...process.env, RUNA_CONNECT: "invalid" }, + stdio: ["pipe", "pipe", "pipe"], + }); + t.after(() => { + if (overridden.exitCode === null && overridden.signalCode === null) overridden.kill(); + }); + let diagnostics = ""; + overridden.stderr.on("data", (chunk: Buffer) => { + diagnostics += chunk.toString(); + }); + const overriddenExit = once(overridden, "exit"); + overridden.stdin.end(input); + const [overrideCode] = await overriddenExit; + assert.equal(overrideCode, 0, diagnostics); + assert.equal(diagnostics, ""); + assert.doesNotMatch(image.diagnostics(), /protocol error|reset without closing handshake/i); + } }, ); diff --git a/vscode/integration/server.ts b/vscode/integration/server.ts index cf3dc1b..5ddcace 100644 --- a/vscode/integration/server.ts +++ b/vscode/integration/server.ts @@ -6,7 +6,11 @@ import { json } from "../test/fixtures"; import WebSocket from "ws"; import { decodeNotification, WEBSOCKET_SUBPROTOCOL } from "../src/protocol"; -export async function startImage(): Promise<{ address: string; stop: () => Promise }> { +export async function startImage(): Promise<{ + address: string; + stop: () => Promise; + diagnostics: () => string; +}> { const executable = resolve( __dirname, "../../../target/debug", @@ -15,6 +19,10 @@ export async function startImage(): Promise<{ address: string; stop: () => Promi const child = spawn(executable, ["--listen", "127.0.0.1:0"], { stdio: ["ignore", "ignore", "pipe"], }); + let diagnostics = ""; + child.stderr.on("data", (chunk: Buffer) => { + diagnostics = (diagnostics + chunk.toString()).slice(-8192); + }); const stop = () => stopImage(child); let address: string; try { @@ -52,7 +60,7 @@ export async function startImage(): Promise<{ address: string; stop: () => Promi await stop(); throw error; } - return { address, stop }; + return { address, stop, diagnostics: () => diagnostics }; } async function stopImage(child: ChildProcess): Promise {