From 69cb853caa5ea397a0685cc410ca312ff7701d3c Mon Sep 17 00:00:00 2001 From: Brent G Date: Thu, 15 Oct 2026 14:44:02 +0000 Subject: [PATCH] fix(pair): say why Astation's connection closed during verification MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The WebSocket reader now records the close frame (code and reason) or the transport error, and the types of Astation messages it couldn't parse (dropped before, silently). 'atem pair' adds both to the error when the connection closes while it waits for Astation's reply. 🤖 Built with SMT --- src/pair_verify.rs | 43 +++++++++++++++++++++++- src/websocket_client.rs | 73 +++++++++++++++++++++++++++++++++++------ 2 files changed, 105 insertions(+), 11 deletions(-) diff --git a/src/pair_verify.rs b/src/pair_verify.rs index 58759c9..27dceeb 100644 --- a/src/pair_verify.rs +++ b/src/pair_verify.rs @@ -219,7 +219,10 @@ pub(crate) async fn next_verify_message(client: &mut AstationClient) -> Result return Ok(message), // Applied after the verification, against the pins it sets. Some(message) => client.hold_encryption(message), - None => anyhow::bail!("Astation connection closed during device verification"), + None => anyhow::bail!( + "Astation connection closed during device verification{}", + client.close_detail() + ), } } } @@ -588,6 +591,44 @@ mod tests { ); } + #[tokio::test] + async fn a_closed_connection_says_why_and_what_could_not_be_read() { + use tokio_tungstenite::tungstenite::protocol::{CloseFrame, frame::coding::CloseCode}; + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let url = format!("ws://{}", listener.local_addr().unwrap()); + let server = tokio::spawn(async move { + let (stream, _) = listener.accept().await.unwrap(); + let mut socket = tokio_tungstenite::accept_async(stream).await.unwrap(); + // A deviceVerified this atem can't parse is dropped, not fatal. + let unreadable = r#"{"type":"deviceVerified","data":{"device_verified":42}}"#; + socket.send(Message::Text(unreadable.into())).await.unwrap(); + socket + .close(Some(CloseFrame { + code: CloseCode::Policy, + reason: "signing failed".into(), + })) + .await + .unwrap(); + }); + let mut client = connected(&url).await; + let Err(failure) = await_device_verified(&mut client, Duration::from_secs(5)).await else { + panic!("expected a closed connection"); + }; + let dir = tempfile::tempdir().unwrap(); + let error = + abandon_verification(&client, &dir.path().join("cred_state.json"), "a", failure).await; + let text = format!("{error:#}"); + assert!( + text.contains("closed by Astation: 1008 \"signing failed\""), + "{text}" + ); + assert!( + text.contains("unreadable messages from Astation: deviceVerified"), + "{text}" + ); + server.await.unwrap(); + } + #[tokio::test] async fn astations_own_abort_is_not_echoed_back() { let dir = tempfile::tempdir().unwrap(); diff --git a/src/websocket_client.rs b/src/websocket_client.rs index d68160e..5a203a8 100644 --- a/src/websocket_client.rs +++ b/src/websocket_client.rs @@ -468,6 +468,27 @@ pub struct AstationClient { /// before Astation said who it is (a pairing-code relay connection never /// authenticates), or during device verification. held_encryption: Vec, + /// How the current socket ended and what it couldn't read, for errors. + transport_notes: std::sync::Arc>, +} + +/// What the reader saw that the caller can't: why the socket closed, and the +/// types of messages that didn't parse (dropped). Reset at each connect. +#[derive(Debug, Default)] +struct TransportNotes { + closed: Option, + unreadable: Vec, +} + +/// At most this many unreadable message types are remembered per socket. +const MAX_UNREADABLE_NOTES: usize = 8; + +/// The `type` of a message that didn't parse, never its content. +fn unreadable_type(text: &str) -> String { + serde_json::from_str::(text) + .ok() + .and_then(|v| v.get("type").and_then(|t| t.as_str()).map(|t| t.chars().take(64).collect())) + .unwrap_or_else(|| "not JSON".to_string()) } /// Status of an encryption message held for later; internal, not shown. @@ -493,9 +514,25 @@ impl AstationClient { connected_astation_id: None, answered_challenge: None, held_encryption: Vec::new(), + transport_notes: Default::default(), } } + /// Why the last socket closed and which Astation messages couldn't be + /// read on it, e.g. ` (closed by Astation: 1000 "bye")`; empty when + /// nothing is known. + pub fn close_detail(&self) -> String { + let notes = self.transport_notes.lock().unwrap(); + let mut parts = Vec::new(); + if let Some(closed) = ¬es.closed { + parts.push(closed.clone()); + } + if !notes.unreadable.is_empty() { + parts.push(format!("unreadable messages from Astation: {}", notes.unreadable.join(", "))); + } + if parts.is_empty() { String::new() } else { format!(" ({})", parts.join("; ")) } + } + /// The Astation this client authenticated with, once it said who it is. pub fn connected_astation_id(&self) -> Option<&str> { self.connected_astation_id.as_deref() @@ -639,26 +676,42 @@ impl AstationClient { }); // Spawn task to handle incoming messages + let notes = std::sync::Arc::new(std::sync::Mutex::new(TransportNotes::default())); + self.transport_notes = notes.clone(); let reader_task = tokio::spawn(async move { - while let Some(message) = read.next().await { + let closed = loop { + let Some(message) = read.next().await else { break "connection ended without a close frame".to_string() }; match message { Ok(Message::Text(text)) => { - if let Ok(astation_msg) = serde_json::from_str::(&text) { - if matches!(&astation_msg, AstationMessage::Terminal(_)) && text.len() > 32 * 1024 { continue; } - if tx.send(astation_msg).await.is_err() { - break; // Receiver has been dropped + match serde_json::from_str::(&text) { + Ok(astation_msg) => { + if matches!(&astation_msg, AstationMessage::Terminal(_)) && text.len() > 32 * 1024 { continue; } + if tx.send(astation_msg).await.is_err() { + break "closed by atem".to_string(); // Receiver has been dropped + } + } + Err(_) => { + let mut notes = notes.lock().unwrap(); + if notes.unreadable.len() < MAX_UNREADABLE_NOTES { notes.unreadable.push(unreadable_type(&text)); } } } } - Ok(Message::Close(_)) => { - break; + Ok(Message::Close(frame)) => { + break match frame { + Some(frame) => { + let reason: String = frame.reason.chars().take(120).collect(); + format!("closed by Astation: {} {reason:?}", u16::from(frame.code)) + } + None => "closed by Astation without a reason".to_string(), + }; } - Err(_) => { - break; + Err(error) => { + break format!("connection error: {error}"); } _ => {} } - } + }; + notes.lock().unwrap().closed = Some(closed); }); self.sender = Some(msg_tx);