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);