From 268902add302a601f54f4e716ac4b462dce5568c Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 9 Nov 2026 16:56:13 +0000 Subject: [PATCH 1/2] fix(serial): SerialClient greets every connection with hello (#443) A UART never closes, so a server learns that the client behind it changed only from a hello. SerialClient never sent one: a host process restarted while the board kept running reused subscription ids the old session still held, the board answered Denied, and mirrored records went silent until a board reset. aimdb-core: - EnvelopeCodec::hello (default None) names the request a client opens each connection with; AimxCodec returns AimX's hello. - ClientConfig::sends_hello sends that hello when the codec has one (else the old Ping/Pong). It skips frames of the previous session that arrive before the reply, resends an unanswered hello every HELLO_RESEND_MS, and stops the engine on a version_mismatch instead of redialing. aimdb-serial-connector: - SerialClient::new and over_port turn sends_hello on. - SerialPortDialer writes a frame delimiter on open, so a frame a previous process left half-written cannot swallow the hello. - tests/client_restart.rs: three host processes in turn on one live line keep mirroring (fails at the second without the fix), plus a lost hello with stray frames, and a refused version. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01WGkDJ6wpoRhHZ8DEGbXbf3 --- aimdb-core/CHANGELOG.md | 12 + aimdb-core/src/session/aimx/codec.rs | 10 + aimdb-core/src/session/client.rs | 133 ++++++++- aimdb-core/src/session/events.rs | 13 + aimdb-core/src/session/mod.rs | 14 +- aimdb-serial-connector/CHANGELOG.md | 11 + aimdb-serial-connector/src/connector.rs | 25 +- aimdb-serial-connector/src/lib.rs | 4 + .../tests/client_restart.rs | 266 ++++++++++++++++++ 9 files changed, 471 insertions(+), 17 deletions(-) create mode 100644 aimdb-serial-connector/tests/client_restart.rs diff --git a/aimdb-core/CHANGELOG.md b/aimdb-core/CHANGELOG.md index 635c3753..b7b58a03 100644 --- a/aimdb-core/CHANGELOG.md +++ b/aimdb-core/CHANGELOG.md @@ -45,6 +45,18 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ids as `OutboundRoutes`. Connectors build late-join snapshots from it instead of caching what they sent. Records without a current value (`SpmcRing`, or nothing written yet) render nothing. +- **`EnvelopeCodec::hello` (#443):** the request a client opens each + connection with under `ClientConfig::sends_hello`. Defaults to `None`, which + keeps the `Ping`/`Pong` handshake; `AimxCodec` returns AimX's `hello`. + +### Changed + +- **`ClientConfig::sends_hello` sends the codec's hello (#443)** when it has + one, instead of always a `Ping`, so it starts a session on servers that + serve one long connection. Frames of the previous session that arrive + before the reply are skipped, an unanswered hello is resent every + `HELLO_RESEND_MS` (1 s), and a server that refuses the protocol version + stops the client instead of leaving it redialing. ### Changed (wire) diff --git a/aimdb-core/src/session/aimx/codec.rs b/aimdb-core/src/session/aimx/codec.rs index 0a843273..5bcd3d47 100644 --- a/aimdb-core/src/session/aimx/codec.rs +++ b/aimdb-core/src/session/aimx/codec.rs @@ -392,6 +392,16 @@ impl EnvelopeCodec for AimxCodec { _ => Err(CodecError::Malformed), } } + + /// `hello` opens every AimX session, and a second one on a connection + /// tells the server the client changed (see `AimxSession::starts_session`). + fn hello(&self) -> Option<(&'static str, Payload)> { + let params = alloc::format!( + r#"{{"client":"aimdb-session-client","version":"{}"}}"#, + crate::remote::PROTOCOL_VERSION + ); + Some(("hello", Arc::from(params.as_bytes()))) + } } #[cfg(test)] diff --git a/aimdb-core/src/session/client.rs b/aimdb-core/src/session/client.rs index bdb872af..b38566ee 100644 --- a/aimdb-core/src/session/client.rs +++ b/aimdb-core/src/session/client.rs @@ -69,8 +69,17 @@ pub struct ClientConfig { /// [`RpcError::Internal`], a dropped `subscribe` ends its stream. Defaults to /// 256; clamped to `1..=MAX_COMMAND_QUEUE_CAP`. pub max_offline_queue: usize, - /// Send a Ping handshake on connect and await the Pong before serving caller - /// commands. A real protocol swaps Ping/Pong for its Hello. + /// Open every connection with a handshake and finish it before serving + /// caller commands. The handshake is the codec's + /// [`hello`](EnvelopeCodec::hello) request when it has one (AimX: `hello`), + /// else a `Ping`/`Pong` exchange. + /// + /// A `hello` is what tells a server on a connection that never closes (a + /// serial line) that a new client is behind it, so it ends the previous + /// client's session. Frames of that session that arrive before the reply + /// are discarded, and an unanswered `hello` is sent again every + /// [`HELLO_RESEND_MS`] for as long as the connection stays up. A server + /// that refuses the client's protocol version stops the engine. pub sends_hello: bool, } @@ -88,6 +97,14 @@ impl Default for ClientConfig { } } +/// How long (ms) the client waits for the reply to its `hello` before sending +/// it again (see [`ClientConfig::sends_hello`]). +pub const HELLO_RESEND_MS: u64 = 1000; + +/// Correlation id of the handshake `hello`. [`ClientHandle`] ids start at 1, +/// so no caller's call or subscription can share it. +const HELLO_ID: u64 = 0; + /// Exponential backoff (ms) for the `attempt`-th redial (1-based), capped at /// [`ClientConfig::max_reconnect_delay`]. fn backoff_delay(config: &ClientConfig, attempt: usize) -> u64 { @@ -361,6 +378,9 @@ enum Ended { Disconnected, /// Every [`ClientHandle`] was dropped — stop the engine. HandlesDropped, + /// The server refused the handshake for good (an incompatible protocol + /// version) — redialing cannot help, so stop the engine. + Refused, } /// On engine exit, close and drain the command channel so buffered/in-flight @@ -537,7 +557,7 @@ async fn client_loop( }; match drive_connection(conn, &codec, &cmd_rx, &prune_rx, &config, &ctx).await { - Ended::HandlesDropped => return, + Ended::HandlesDropped | Ended::Refused => return, Ended::Disconnected => match reconnect_after(&mut attempt, &config, &ctx).await { true => continue, false => return, @@ -589,6 +609,83 @@ fn sweep_abandoned(pending: &mut PendingCalls) { pending.retain(|_, reply| !reply.is_canceled()); } +/// Send the codec's `hello` and wait for its reply, resending it every +/// [`HELLO_RESEND_MS`] until one arrives (a frame lost on a lossy line, or a +/// peer that is not up yet). +/// +/// Every other frame is discarded: on a connection that outlived its previous +/// client, the server keeps streaming that client's events until it reads the +/// `hello`. A resent `hello` is harmless — the server handles them in order, +/// before anything this client sends next, and the extra reply that comes +/// back later matches no pending call, so the demux loop drops it. +async fn send_hello( + conn: &mut Box, + codec: &C, + out: &mut Vec, + method: &'static str, + params: Payload, + ctx: &crate::RuntimeContext, +) -> Result<(), Ended> +where + C: EnvelopeCodec + ?Sized, +{ + let mut resent = false; + loop { + out.clear(); + let hello = Inbound::Request { + id: HELLO_ID, + method: method.into(), + params: params.clone(), + }; + if codec.encode_inbound(hello, out).is_err() || conn.send(out).await.is_err() { + return Err(Ended::Disconnected); + } + let mut resend = ctx + .time() + .sleep(core::time::Duration::from_millis(HELLO_RESEND_MS)) + .fuse(); + loop { + let frame = { + let mut recv = conn.recv().fuse(); + select_biased! { + r = recv => Some(r), + _ = resend => None, + } + }; + let frame = match frame { + Some(Ok(Some(frame))) => frame, + Some(Ok(None) | Err(_)) => return Err(Ended::Disconnected), + // No reply in time: send it again. + None => break, + }; + match codec.decode_outbound(&frame) { + Ok(Outbound::Reply { + id: HELLO_ID, + result, + }) => { + return match result { + Err(RpcError::VersionMismatch) => { + report(SessionEvent::HelloRefused); + Err(Ended::Refused) + } + // Any other answer still came from a server that read + // the `hello`, which is all the handshake is for. + _ => Ok(()), + }; + } + // A frame of the session this `hello` replaces. + _ => continue, + } + } + if !resent { + report(SessionEvent::HelloUnanswered { + after_ms: HELLO_RESEND_MS, + }); + resent = true; + } + } +} + /// Drive one dialed [`Connection`]: optional handshake, then `biased` demux of /// server frames (resolve `Reply` by `id`, route `Event`/`Snapshot` to their /// subscription channels) interleaved with caller commands. Pending state is @@ -633,19 +730,27 @@ where .fuse() }); - // Handshake-as-caller: prove the link with Ping/Pong before serving commands. + // Handshake-as-caller, before serving commands: the codec's hello, which + // also starts a fresh session on the server, or else a Ping/Pong. if config.sends_hello { - out.clear(); - if codec.encode_inbound(Inbound::Ping, &mut out).is_err() || conn.send(&out).await.is_err() - { - return Ended::Disconnected; - } - match conn.recv().await { - Ok(Some(frame)) => match codec.decode_outbound(&frame) { - Ok(Outbound::Pong) => {} + if let Some((method, params)) = codec.hello() { + if let Err(ended) = send_hello(&mut conn, codec, &mut out, method, params, ctx).await { + return ended; + } + } else { + out.clear(); + if codec.encode_inbound(Inbound::Ping, &mut out).is_err() + || conn.send(&out).await.is_err() + { + return Ended::Disconnected; + } + match conn.recv().await { + Ok(Some(frame)) => match codec.decode_outbound(&frame) { + Ok(Outbound::Pong) => {} + _ => return Ended::Disconnected, + }, _ => return Ended::Disconnected, - }, - _ => return Ended::Disconnected, + } } } diff --git a/aimdb-core/src/session/events.rs b/aimdb-core/src/session/events.rs index e76b4e72..f5e02dfe 100644 --- a/aimdb-core/src/session/events.rs +++ b/aimdb-core/src/session/events.rs @@ -60,6 +60,10 @@ pub(crate) enum SessionEvent<'a> { topic: &'a str, skipped: u64, }, + HelloUnanswered { + after_ms: u64, + }, + HelloRefused, ReceiveFramingFailed, SendFrameRejected { bytes: usize, @@ -145,6 +149,15 @@ pub(crate) fn report(event: SessionEvent<'_>) { topic, skipped ), + SessionEvent::HelloUnanswered { after_ms } => log_warn!( + target: target::CLIENT, + "no reply to hello after {} ms; resending until the server answers", + after_ms + ), + SessionEvent::HelloRefused => log_error!( + target: target::CLIENT, + "server refused the hello: incompatible protocol version; stopping the client" + ), SessionEvent::ReceiveFramingFailed => { log_warn!( target: target::IO, diff --git a/aimdb-core/src/session/mod.rs b/aimdb-core/src/session/mod.rs index 81749c83..5b547139 100644 --- a/aimdb-core/src/session/mod.rs +++ b/aimdb-core/src/session/mod.rs @@ -42,7 +42,7 @@ mod topic_match; pub use topic_match::{is_wildcard, pattern_contains, topic_matches}; #[cfg(feature = "connector-session")] -pub use client::{pump_client, run_client, ClientConfig, ClientHandle}; +pub use client::{pump_client, run_client, ClientConfig, ClientHandle, HELLO_RESEND_MS}; #[cfg(feature = "connector-session")] pub use connector::{SessionClientConnector, SessionServerConnector}; #[cfg(feature = "connector-session")] @@ -559,6 +559,18 @@ pub trait EnvelopeCodec: Send + Sync { /// reply/event). The dual of [`encode`](EnvelopeCodec::encode); the result /// borrows the frame (`Outbound`'s `sub`/`topic` are `&str` slices into it). fn decode_outbound<'a>(&self, frame: &'a [u8]) -> Result, CodecError>; + + /// The request a client opens each connection with when + /// [`ClientConfig::sends_hello`](crate::session::ClientConfig::sends_hello) + /// is set: its method and params. It should be a method for which the + /// server's [`Session::starts_session`] holds, so a server serving one long + /// connection (a serial line) knows the client behind it changed. + /// + /// `None` (the default) falls back to a `Ping`/`Pong` handshake, which only + /// proves the link. + fn hello(&self) -> Option<(&'static str, Payload)> { + None + } } // =========================================================================== diff --git a/aimdb-serial-connector/CHANGELOG.md b/aimdb-serial-connector/CHANGELOG.md index c1f1a0ca..8d4c657a 100644 --- a/aimdb-serial-connector/CHANGELOG.md +++ b/aimdb-serial-connector/CHANGELOG.md @@ -7,6 +7,17 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Fixed + +- **A restarted `SerialClient` keeps mirroring (#443).** `SerialClient` sent + no `hello`, so a host process restarted while the board kept running stayed + in the old session: its subscriptions reused ids the old session still + held, the board refused them, and records mirrored from the board went + silent until a board reset. `SerialClient::new` and `over_port` now open + every connection with a `hello` and wait for the reply before subscribing. + `over_port` also writes a frame delimiter when it opens the port, so a + frame a previous process left half-written cannot swallow the `hello`. + ## [0.2.0] - 2026-11-03 ### Changed (breaking, wire) diff --git a/aimdb-serial-connector/src/connector.rs b/aimdb-serial-connector/src/connector.rs index 39190604..ce37bfa5 100644 --- a/aimdb-serial-connector/src/connector.rs +++ b/aimdb-serial-connector/src/connector.rs @@ -72,7 +72,7 @@ impl Dialer for SerialPortDialer { fn connect(&self) -> BoxFut<'_, TransportResult>> { Box::pin(async move { use tokio_serial::SerialPortBuilderExt; - let stream = tokio_serial::new(&self.path, self.baud) + let mut stream = tokio_serial::new(&self.path, self.baud) .open_native_async() .map_err(|_| TransportError::Io)?; // Discard bytes a previous session left in the OS input buffer; @@ -80,6 +80,14 @@ impl Dialer for SerialPortDialer { // and desyncs the stream until the next COBS sentinel. use tokio_serial::SerialPort; let _ = stream.clear(tokio_serial::ClearBuffer::Input); + // Close any frame a previous process left half-written on the + // line, so the board does not glue the `hello` onto it and drop + // both. A spare sentinel is an empty chunk the reader skips. + use tokio::io::AsyncWriteExt; + stream + .write_all(&[crate::framing::DELIM]) + .await + .map_err(|_| TransportError::Io)?; Ok( Box::new(framed(aimdb_tokio_adapter::net::TokioByteStream(stream))) as Box, @@ -94,6 +102,13 @@ pub struct SerialClient; impl SerialClient { /// Mirror records to and from the AimX peer on `stream`, served once. /// + /// The client opens the connection with a `hello` and waits for the reply + /// before it subscribes, so a server that kept running while its previous + /// client went away ends that client's session (see the crate docs, + /// "Sessions on a line that never closes"). A caller passing its own + /// config through `.with_config(...)` keeps that by leaving `sends_hello` + /// on. + /// /// Reconnect is **disabled** (unlike `ClientConfig::default`): the stream is /// moved in and cannot be re-acquired. `run_client` would stop anyway — it /// treats a dialer's `TransportError::Closed` as terminal — but saying so @@ -114,6 +129,7 @@ impl SerialClient { .scheme(DEFAULT_SCHEME) .with_config(ClientConfig { reconnect: false, + sends_hello: true, ..ClientConfig::default() }) } @@ -122,7 +138,8 @@ impl SerialClient { /// reconnecting after a drop. /// /// Keeps `ClientConfig`'s default `reconnect: true` — unlike - /// [`new`](Self::new), this dialer can genuinely reopen the device. + /// [`new`](Self::new), this dialer can genuinely reopen the device. Sends a + /// `hello` on every connection, as [`new`](Self::new) does. #[cfg(feature = "std")] pub fn over_port( path: impl Into, @@ -130,6 +147,10 @@ impl SerialClient { ) -> SessionClientConnector { SessionClientConnector::new(SerialPortDialer::new(path, baud), AimxCodec) .scheme(DEFAULT_SCHEME) + .with_config(ClientConfig { + sends_hello: true, + ..ClientConfig::default() + }) } } diff --git a/aimdb-serial-connector/src/lib.rs b/aimdb-serial-connector/src/lib.rs index 26e4050a..cfeb7142 100644 --- a/aimdb-serial-connector/src/lib.rs +++ b/aimdb-serial-connector/src/lib.rs @@ -31,6 +31,10 @@ //! and answers the new client from a fresh one. So every client that connects //! gets its events, however often clients come and go. //! +//! Every client sends one: `SerialClient` opens each connection with a +//! `hello` and waits for the reply before it subscribes, resending it if the +//! reply does not come, and `aimdb-client` sends one when it connects. +//! //! Until the next `hello`, a client that left keeps its subscriptions: they //! go on streaming into the UART and hold their buffer slots. Size the //! subscriber slots of a record with remote access for the subscribers a diff --git a/aimdb-serial-connector/tests/client_restart.rs b/aimdb-serial-connector/tests/client_restart.rs new file mode 100644 index 00000000..bb9035a5 --- /dev/null +++ b/aimdb-serial-connector/tests/client_restart.rs @@ -0,0 +1,266 @@ +//! A `SerialClient` that restarts while the board keeps running (#443). +//! +//! A UART never closes, so the server learns that the client behind it changed +//! only from a `hello`. `SerialClient` sends one on every connection: without +//! it, a restarted process numbers its subscriptions from 1 again, the old +//! session still holds those ids, and the server refuses them, so the +//! mirrored records go silent until the board is reset. + +#![cfg(feature = "std")] + +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +use aimdb_core::buffer::BufferCfg; +use aimdb_core::{AimDb, AimDbBuilder}; +use aimdb_serial_connector::framing::{encode_frame, FrameAccumulator}; +use aimdb_serial_connector::{SerialClient, SerialServer}; +use aimdb_tokio_adapter::net::TokioByteStream; +use aimdb_tokio_adapter::{TokioAdapter, TokioRecordRegistrarExt}; +use serde::{Deserialize, Serialize}; +use serde_json::{json, Value}; +use tokio::io::{AsyncReadExt, AsyncWriteExt, DuplexStream, ReadHalf, WriteHalf}; +use tokio::task::JoinHandle; + +#[derive(Debug, Clone, Serialize, Deserialize)] +struct Temp { + c: u64, +} + +/// Where bytes the board sends go: the current host process, or nowhere +/// while none has the port open. +type HostSink = Arc>>>; + +/// One host process: its database and the task carrying its bytes to the line. +struct Host { + db: Arc, + runner: JoinHandle<()>, + uplink: JoinHandle<()>, +} + +impl Host { + /// Kill the process: no goodbye reaches the board, as with a closed port. + async fn kill(self, sink: &HostSink) { + self.runner.abort(); + self.uplink.abort(); + *sink.lock().await = None; + } +} + +/// Start a host process that mirrors the board's `temp` over the line. +async fn start_host( + sink: &HostSink, + line_wr: &Arc>>, +) -> Host { + let (process_end, port_end) = tokio::io::duplex(64 * 1024); + let (mut port_rd, port_wr) = tokio::io::split(port_end); + *sink.lock().await = Some(port_wr); + let line_wr = line_wr.clone(); + let uplink = tokio::spawn(async move { + let mut buf = [0u8; 256]; + while let Ok(n) = port_rd.read(&mut buf).await { + if n == 0 || line_wr.lock().await.write_all(&buf[..n]).await.is_err() { + break; + } + } + }); + + let mut builder = AimDbBuilder::new() + .runtime(Arc::new(TokioAdapter)) + .with_connector(SerialClient::new(TokioByteStream(process_end))); + builder.configure::("temp", |reg| { + reg.buffer(BufferCfg::SingleLatest) + .with_remote_access() + .link_from("serial://temp") + .with_deserializer(|_ctx, d: &[u8]| { + serde_json::from_slice::(d).map_err(|e| e.to_string()) + }) + .finish(); + }); + let (db, runner) = builder.build().await.expect("build host db"); + Host { + db: Arc::new(db), + runner: tokio::spawn(runner.run()), + uplink, + } +} + +/// The latest `temp` the host has mirrored, if any. +fn mirrored(db: &AimDb) -> Option { + db.try_latest_as_json("temp") + .and_then(|s| s.value["c"].as_u64()) +} + +#[tokio::test] +async fn a_restarted_client_keeps_mirroring() { + // The line: the board holds one end for its whole life. + let (board_end, line_end) = tokio::io::duplex(64 * 1024); + + let mut builder = AimDbBuilder::new() + .runtime(Arc::new(TokioAdapter)) + .with_connector(SerialServer::new(TokioByteStream(board_end))); + builder.configure::("temp", |reg| { + reg.buffer(BufferCfg::SpmcRing { capacity: 16 }) + .with_remote_access(); + }); + let (board, runner) = builder.build().await.expect("build board db"); + let board = Arc::new(board); + tokio::spawn(runner.run()); + + // The record changes all the time, so events of the old session are + // still on the line while the next client says hello. + let produced = Arc::new(Mutex::new(0u64)); + { + let (board, produced) = (board.clone(), produced.clone()); + tokio::spawn(async move { + loop { + let c = { + let mut n = produced.lock().unwrap(); + *n += 1; + *n + }; + let _ = board.produce::("temp", Temp { c }); + tokio::time::sleep(Duration::from_millis(5)).await; + } + }); + } + + // The host side of the line, which outlives every host process on it. + let (mut line_rd, line_wr): (ReadHalf, _) = tokio::io::split(line_end); + let line_wr = Arc::new(tokio::sync::Mutex::new(line_wr)); + let sink: HostSink = Arc::new(tokio::sync::Mutex::new(None)); + { + let sink = sink.clone(); + tokio::spawn(async move { + let mut buf = [0u8; 1024]; + while let Ok(n) = line_rd.read(&mut buf).await { + if n == 0 { + break; + } + let mut sink = sink.lock().await; + if let Some(port) = sink.as_mut() { + if port.write_all(&buf[..n]).await.is_err() { + *sink = None; + } + } + } + }); + } + + for generation in 1..=3 { + let host = start_host(&sink, &line_wr).await; + let floor = *produced.lock().unwrap(); + let mut caught_up = false; + for _ in 0..150 { + tokio::time::sleep(Duration::from_millis(20)).await; + if mirrored(&host.db).is_some_and(|c| c > floor) { + caught_up = true; + break; + } + } + assert!( + caught_up, + "host process {generation}: the mirror never got past {floor} (latest: {:?})", + mirrored(&host.db) + ); + host.kill(&sink).await; + // Let the board stream into the void for a while, as it does while + // the host restarts. + tokio::time::sleep(Duration::from_millis(50)).await; + } +} + +/// A board that loses the first `hello`, has events of an earlier session in +/// flight, and answers the second `hello`: the client waits it out and only +/// then subscribes. A board refusing the version stops the client instead. +#[tokio::test] +async fn the_hello_is_resent_and_stray_frames_are_skipped() { + let (board_end, process_end) = tokio::io::duplex(64 * 1024); + let host = start_fake_host(process_end).await; + let (mut rd, mut wr) = tokio::io::split(board_end); + let mut acc = FrameAccumulator::new(); + + let first = next_frame(&mut rd, &mut acc).await; + assert_eq!(first["method"], "hello", "the client opens with a hello"); + assert_eq!(first["id"], 0); + // An earlier session's traffic, and no reply to the hello. + send(&mut wr, json!({"t":"pong"})).await; + send(&mut wr, json!({"t":"reply","id":7,"ok":null})).await; + + let started = tokio::time::Instant::now(); + let second = next_frame(&mut rd, &mut acc).await; + assert_eq!(second["method"], "hello", "the unanswered hello is resent"); + assert!(started.elapsed() >= Duration::from_millis(500)); + send(&mut wr, json!({"t":"reply","id":0,"ok":{}})).await; + + let sub = next_frame(&mut rd, &mut acc).await; + assert_eq!(sub["t"], "sub", "subscribes once greeted, got {sub}"); + assert_eq!(sub["topic"], "temp"); + host.abort(); +} + +#[tokio::test] +async fn a_refused_version_stops_the_client() { + let (board_end, process_end) = tokio::io::duplex(64 * 1024); + let host = start_fake_host(process_end).await; + let (mut rd, mut wr) = tokio::io::split(board_end); + let mut acc = FrameAccumulator::new(); + + let hello = next_frame(&mut rd, &mut acc).await; + assert_eq!(hello["method"], "hello"); + send( + &mut wr, + json!({"t":"reply","id":0,"err":"version_mismatch"}), + ) + .await; + + // Neither a subscribe nor another hello follows. + let mut buf = [0u8; 256]; + match tokio::time::timeout(Duration::from_millis(1500), rd.read(&mut buf)).await { + Err(_) | Ok(Ok(0)) => {} + Ok(r) => panic!( + "the client kept talking after a refusal: {r:?} {:?}", + &buf[..] + ), + } + host.abort(); +} + +/// A host process mirroring `temp` over `stream`, which a test drives by hand. +async fn start_fake_host(stream: DuplexStream) -> JoinHandle<()> { + let mut builder = AimDbBuilder::new() + .runtime(Arc::new(TokioAdapter)) + .with_connector(SerialClient::new(TokioByteStream(stream))); + builder.configure::("temp", |reg| { + reg.buffer(BufferCfg::SingleLatest) + .with_remote_access() + .link_from("serial://temp") + .with_deserializer(|_ctx, d: &[u8]| { + serde_json::from_slice::(d).map_err(|e| e.to_string()) + }) + .finish(); + }); + let (_db, runner) = builder.build().await.expect("build host db"); + tokio::spawn(runner.run()) +} + +async fn send(wr: &mut WriteHalf, v: Value) { + let mut wire = Vec::new(); + encode_frame(&serde_json::to_vec(&v).unwrap(), &mut wire); + wr.write_all(&wire).await.unwrap(); +} + +async fn next_frame(rd: &mut ReadHalf, acc: &mut FrameAccumulator) -> Value { + let mut buf = [0u8; 256]; + loop { + if let Some(Ok(f)) = acc.next_frame() { + return serde_json::from_slice(&f).expect("an AimX frame"); + } + let n = tokio::time::timeout(Duration::from_secs(3), rd.read(&mut buf)) + .await + .expect("a frame within 3s") + .unwrap(); + assert!(n > 0, "the client closed the line"); + acc.push_bytes(&buf[..n]); + } +} From f6b3a2f77e2d846eedd3a1f5b520dd70bdffa1d3 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 9 Nov 2026 17:20:24 +0000 Subject: [PATCH 2/2] fix(serial): send hello once without awaiting the reply (#443) Frame order already puts the hello ahead of every command, so waiting for the Welcome bought little and added failure modes: the handshake loop did not watch the command channel, so a silent line kept it resending after every handle was dropped, and a Welcome over the receive limit was dropped by the framer and left the client resending forever. - sends_hello with a codec hello now sends it once per connection and goes straight on. Its reply (id 0, which no ClientHandle issues) is dropped in the demux loop; a version_mismatch reply is logged as HelloRefused and the client keeps running. The Ping/Pong path is unchanged. - Drop the resend timer, HELLO_RESEND_MS, Ended::Refused and the HelloUnanswered event. - Tests: the resend and version-refusal tests become one showing the hello goes first, the subscribe follows without a Welcome, and late id-0 replies (ok and refused) leave the subscription mirroring. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01WGkDJ6wpoRhHZ8DEGbXbf3 --- aimdb-core/CHANGELOG.md | 10 +- aimdb-core/src/session/client.rs | 136 +++++------------- aimdb-core/src/session/events.rs | 10 +- aimdb-core/src/session/mod.rs | 7 +- aimdb-serial-connector/CHANGELOG.md | 3 +- aimdb-serial-connector/src/connector.rs | 6 +- aimdb-serial-connector/src/lib.rs | 7 +- .../tests/client_restart.rs | 85 +++++------ 8 files changed, 102 insertions(+), 162 deletions(-) diff --git a/aimdb-core/CHANGELOG.md b/aimdb-core/CHANGELOG.md index b7b58a03..c6f1c808 100644 --- a/aimdb-core/CHANGELOG.md +++ b/aimdb-core/CHANGELOG.md @@ -52,11 +52,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Changed - **`ClientConfig::sends_hello` sends the codec's hello (#443)** when it has - one, instead of always a `Ping`, so it starts a session on servers that - serve one long connection. Frames of the previous session that arrive - before the reply are skipped, an unanswered hello is resent every - `HELLO_RESEND_MS` (1 s), and a server that refuses the protocol version - stops the client instead of leaving it redialing. + one, instead of a `Ping`, so it starts a session on servers that serve one + long connection. The hello goes out once per connection, ahead of every + command, and its reply is not awaited: frame order is enough. A reply + refusing the protocol version is logged; the client keeps running. Codecs + without a hello keep the `Ping`/`Pong` exchange, which still waits. ### Changed (wire) diff --git a/aimdb-core/src/session/client.rs b/aimdb-core/src/session/client.rs index b38566ee..d8b109f5 100644 --- a/aimdb-core/src/session/client.rs +++ b/aimdb-core/src/session/client.rs @@ -69,17 +69,19 @@ pub struct ClientConfig { /// [`RpcError::Internal`], a dropped `subscribe` ends its stream. Defaults to /// 256; clamped to `1..=MAX_COMMAND_QUEUE_CAP`. pub max_offline_queue: usize, - /// Open every connection with a handshake and finish it before serving - /// caller commands. The handshake is the codec's - /// [`hello`](EnvelopeCodec::hello) request when it has one (AimX: `hello`), - /// else a `Ping`/`Pong` exchange. + /// Open every connection with a handshake, ahead of any caller command. + /// What that means depends on the codec: /// - /// A `hello` is what tells a server on a connection that never closes (a - /// serial line) that a new client is behind it, so it ends the previous - /// client's session. Frames of that session that arrive before the reply - /// are discarded, and an unanswered `hello` is sent again every - /// [`HELLO_RESEND_MS`] for as long as the connection stays up. A server - /// that refuses the client's protocol version stops the engine. + /// - With a [`hello`](EnvelopeCodec::hello) request (AimX: `hello`), the + /// engine sends it once and goes straight on, **without waiting for the + /// reply**. The connection keeps frame order, so the server handles the + /// `hello` before anything sent after it; that is what tells a server on + /// a connection that never closes (a serial line) that a new client is + /// behind it, so it ends the previous client's session. The reply + /// arrives later and is dropped, except that a refused protocol version + /// is logged. A `hello` lost on the way is not resent. + /// - Without one, the engine sends a `Ping` and waits for the `Pong` + /// before serving commands. pub sends_hello: bool, } @@ -97,12 +99,9 @@ impl Default for ClientConfig { } } -/// How long (ms) the client waits for the reply to its `hello` before sending -/// it again (see [`ClientConfig::sends_hello`]). -pub const HELLO_RESEND_MS: u64 = 1000; - -/// Correlation id of the handshake `hello`. [`ClientHandle`] ids start at 1, -/// so no caller's call or subscription can share it. +/// Correlation id of the handshake `hello` (see [`ClientConfig::sends_hello`]). +/// [`ClientHandle`] ids start at 1, so its reply never matches a caller's +/// call or subscription. const HELLO_ID: u64 = 0; /// Exponential backoff (ms) for the `attempt`-th redial (1-based), capped at @@ -378,9 +377,6 @@ enum Ended { Disconnected, /// Every [`ClientHandle`] was dropped — stop the engine. HandlesDropped, - /// The server refused the handshake for good (an incompatible protocol - /// version) — redialing cannot help, so stop the engine. - Refused, } /// On engine exit, close and drain the command channel so buffered/in-flight @@ -557,7 +553,7 @@ async fn client_loop( }; match drive_connection(conn, &codec, &cmd_rx, &prune_rx, &config, &ctx).await { - Ended::HandlesDropped | Ended::Refused => return, + Ended::HandlesDropped => return, Ended::Disconnected => match reconnect_after(&mut attempt, &config, &ctx).await { true => continue, false => return, @@ -609,83 +605,6 @@ fn sweep_abandoned(pending: &mut PendingCalls) { pending.retain(|_, reply| !reply.is_canceled()); } -/// Send the codec's `hello` and wait for its reply, resending it every -/// [`HELLO_RESEND_MS`] until one arrives (a frame lost on a lossy line, or a -/// peer that is not up yet). -/// -/// Every other frame is discarded: on a connection that outlived its previous -/// client, the server keeps streaming that client's events until it reads the -/// `hello`. A resent `hello` is harmless — the server handles them in order, -/// before anything this client sends next, and the extra reply that comes -/// back later matches no pending call, so the demux loop drops it. -async fn send_hello( - conn: &mut Box, - codec: &C, - out: &mut Vec, - method: &'static str, - params: Payload, - ctx: &crate::RuntimeContext, -) -> Result<(), Ended> -where - C: EnvelopeCodec + ?Sized, -{ - let mut resent = false; - loop { - out.clear(); - let hello = Inbound::Request { - id: HELLO_ID, - method: method.into(), - params: params.clone(), - }; - if codec.encode_inbound(hello, out).is_err() || conn.send(out).await.is_err() { - return Err(Ended::Disconnected); - } - let mut resend = ctx - .time() - .sleep(core::time::Duration::from_millis(HELLO_RESEND_MS)) - .fuse(); - loop { - let frame = { - let mut recv = conn.recv().fuse(); - select_biased! { - r = recv => Some(r), - _ = resend => None, - } - }; - let frame = match frame { - Some(Ok(Some(frame))) => frame, - Some(Ok(None) | Err(_)) => return Err(Ended::Disconnected), - // No reply in time: send it again. - None => break, - }; - match codec.decode_outbound(&frame) { - Ok(Outbound::Reply { - id: HELLO_ID, - result, - }) => { - return match result { - Err(RpcError::VersionMismatch) => { - report(SessionEvent::HelloRefused); - Err(Ended::Refused) - } - // Any other answer still came from a server that read - // the `hello`, which is all the handshake is for. - _ => Ok(()), - }; - } - // A frame of the session this `hello` replaces. - _ => continue, - } - } - if !resent { - report(SessionEvent::HelloUnanswered { - after_ms: HELLO_RESEND_MS, - }); - resent = true; - } - } -} - /// Drive one dialed [`Connection`]: optional handshake, then `biased` demux of /// server frames (resolve `Reply` by `id`, route `Event`/`Snapshot` to their /// subscription channels) interleaved with caller commands. Pending state is @@ -734,8 +653,16 @@ where // also starts a fresh session on the server, or else a Ping/Pong. if config.sends_hello { if let Some((method, params)) = codec.hello() { - if let Err(ended) = send_hello(&mut conn, codec, &mut out, method, params, ctx).await { - return ended; + // Sent once, not awaited: frame order puts it ahead of every + // command, and the demux loop below drops its reply. + out.clear(); + let hello = Inbound::Request { + id: HELLO_ID, + method: method.into(), + params, + }; + if codec.encode_inbound(hello, &mut out).is_err() || conn.send(&out).await.is_err() { + return Ended::Disconnected; } } else { out.clear(); @@ -805,6 +732,17 @@ where Ok(None) | Err(_) => return Ended::Disconnected, }; match codec.decode_outbound(&frame) { + // The reply to the handshake hello, which nobody awaits. + // Only a refusal is worth a word: the server keeps + // serving, but on a protocol this client may not speak. + Ok(Outbound::Reply { + id: HELLO_ID, + result, + }) => { + if result == Err(RpcError::VersionMismatch) { + report(SessionEvent::HelloRefused); + } + } Ok(Outbound::Reply { id, result }) => { if let Some(tx) = pending.remove(&id) { let _ = tx.send(result); diff --git a/aimdb-core/src/session/events.rs b/aimdb-core/src/session/events.rs index f5e02dfe..2b314306 100644 --- a/aimdb-core/src/session/events.rs +++ b/aimdb-core/src/session/events.rs @@ -60,9 +60,6 @@ pub(crate) enum SessionEvent<'a> { topic: &'a str, skipped: u64, }, - HelloUnanswered { - after_ms: u64, - }, HelloRefused, ReceiveFramingFailed, SendFrameRejected { @@ -149,14 +146,9 @@ pub(crate) fn report(event: SessionEvent<'_>) { topic, skipped ), - SessionEvent::HelloUnanswered { after_ms } => log_warn!( - target: target::CLIENT, - "no reply to hello after {} ms; resending until the server answers", - after_ms - ), SessionEvent::HelloRefused => log_error!( target: target::CLIENT, - "server refused the hello: incompatible protocol version; stopping the client" + "server refused the hello: incompatible protocol version" ), SessionEvent::ReceiveFramingFailed => { log_warn!( diff --git a/aimdb-core/src/session/mod.rs b/aimdb-core/src/session/mod.rs index 5b547139..e713bf38 100644 --- a/aimdb-core/src/session/mod.rs +++ b/aimdb-core/src/session/mod.rs @@ -42,7 +42,7 @@ mod topic_match; pub use topic_match::{is_wildcard, pattern_contains, topic_matches}; #[cfg(feature = "connector-session")] -pub use client::{pump_client, run_client, ClientConfig, ClientHandle, HELLO_RESEND_MS}; +pub use client::{pump_client, run_client, ClientConfig, ClientHandle}; #[cfg(feature = "connector-session")] pub use connector::{SessionClientConnector, SessionServerConnector}; #[cfg(feature = "connector-session")] @@ -566,8 +566,9 @@ pub trait EnvelopeCodec: Send + Sync { /// server's [`Session::starts_session`] holds, so a server serving one long /// connection (a serial line) knows the client behind it changed. /// - /// `None` (the default) falls back to a `Ping`/`Pong` handshake, which only - /// proves the link. + /// The client sends it once per connection and does not wait for the + /// reply. `None` (the default) falls back to a `Ping`/`Pong` handshake, + /// which only proves the link. fn hello(&self) -> Option<(&'static str, Payload)> { None } diff --git a/aimdb-serial-connector/CHANGELOG.md b/aimdb-serial-connector/CHANGELOG.md index 8d4c657a..787be237 100644 --- a/aimdb-serial-connector/CHANGELOG.md +++ b/aimdb-serial-connector/CHANGELOG.md @@ -14,7 +14,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 in the old session: its subscriptions reused ids the old session still held, the board refused them, and records mirrored from the board went silent until a board reset. `SerialClient::new` and `over_port` now open - every connection with a `hello` and wait for the reply before subscribing. + every connection with a `hello`, ahead of their subscriptions and writes, + without waiting for the reply. A `hello` lost on the line is not resent. `over_port` also writes a frame delimiter when it opens the port, so a frame a previous process left half-written cannot swallow the `hello`. diff --git a/aimdb-serial-connector/src/connector.rs b/aimdb-serial-connector/src/connector.rs index ce37bfa5..7e38c392 100644 --- a/aimdb-serial-connector/src/connector.rs +++ b/aimdb-serial-connector/src/connector.rs @@ -102,9 +102,9 @@ pub struct SerialClient; impl SerialClient { /// Mirror records to and from the AimX peer on `stream`, served once. /// - /// The client opens the connection with a `hello` and waits for the reply - /// before it subscribes, so a server that kept running while its previous - /// client went away ends that client's session (see the crate docs, + /// The client opens the connection with a `hello`, ahead of its + /// subscriptions and writes, so a server that kept running while its + /// previous client went away ends that client's session (see the crate docs, /// "Sessions on a line that never closes"). A caller passing its own /// config through `.with_config(...)` keeps that by leaving `sends_hello` /// on. diff --git a/aimdb-serial-connector/src/lib.rs b/aimdb-serial-connector/src/lib.rs index cfeb7142..5fa07c55 100644 --- a/aimdb-serial-connector/src/lib.rs +++ b/aimdb-serial-connector/src/lib.rs @@ -32,8 +32,11 @@ //! gets its events, however often clients come and go. //! //! Every client sends one: `SerialClient` opens each connection with a -//! `hello` and waits for the reply before it subscribes, resending it if the -//! reply does not come, and `aimdb-client` sends one when it connects. +//! `hello`, ahead of its subscriptions and writes, and `aimdb-client` sends +//! one when it connects. `SerialClient` does not wait for the reply — the +//! line keeps byte order, so the server reads the `hello` first — and does +//! not resend a `hello` lost on the line: the client then stays in the old +//! session until it reconnects. //! //! Until the next `hello`, a client that left keeps its subscriptions: they //! go on streaming into the UART and hold their buffer slots. Size the diff --git a/aimdb-serial-connector/tests/client_restart.rs b/aimdb-serial-connector/tests/client_restart.rs index bb9035a5..16aabbcf 100644 --- a/aimdb-serial-connector/tests/client_restart.rs +++ b/aimdb-serial-connector/tests/client_restart.rs @@ -170,64 +170,69 @@ async fn a_restarted_client_keeps_mirroring() { } } -/// A board that loses the first `hello`, has events of an earlier session in -/// flight, and answers the second `hello`: the client waits it out and only -/// then subscribes. A board refusing the version stops the client instead. +/// The client sends `hello` first and subscribes without waiting for the +/// `Welcome`. When the reply does arrive, late and with id `0`, it touches no +/// subscription — not even a refusal, which is only logged. #[tokio::test] -async fn the_hello_is_resent_and_stray_frames_are_skipped() { +async fn the_hello_goes_first_and_its_reply_is_not_awaited() { let (board_end, process_end) = tokio::io::duplex(64 * 1024); - let host = start_fake_host(process_end).await; + let (db, host) = start_fake_host(process_end).await; let (mut rd, mut wr) = tokio::io::split(board_end); let mut acc = FrameAccumulator::new(); - let first = next_frame(&mut rd, &mut acc).await; - assert_eq!(first["method"], "hello", "the client opens with a hello"); - assert_eq!(first["id"], 0); - // An earlier session's traffic, and no reply to the hello. - send(&mut wr, json!({"t":"pong"})).await; - send(&mut wr, json!({"t":"reply","id":7,"ok":null})).await; - - let started = tokio::time::Instant::now(); - let second = next_frame(&mut rd, &mut acc).await; - assert_eq!(second["method"], "hello", "the unanswered hello is resent"); - assert!(started.elapsed() >= Duration::from_millis(500)); - send(&mut wr, json!({"t":"reply","id":0,"ok":{}})).await; + let hello = next_frame(&mut rd, &mut acc).await; + assert_eq!( + hello["t"], "req", + "the client opens with a hello, got {hello}" + ); + assert_eq!(hello["method"], "hello"); + assert_eq!(hello["id"], 0); + assert_eq!( + hello["params"]["version"], + aimdb_core::remote::PROTOCOL_VERSION + ); + // No Welcome sent: the subscription follows anyway. let sub = next_frame(&mut rd, &mut acc).await; - assert_eq!(sub["t"], "sub", "subscribes once greeted, got {sub}"); + assert_eq!(sub["t"], "sub", "subscribes without a Welcome, got {sub}"); assert_eq!(sub["topic"], "temp"); - host.abort(); -} + let sub_id = sub["id"].as_u64().expect("a subscribe id"); + assert_ne!(sub_id, 0, "the subscription cannot share the hello's id"); + let sub_id = sub_id.to_string(); -#[tokio::test] -async fn a_refused_version_stops_the_client() { - let (board_end, process_end) = tokio::io::duplex(64 * 1024); - let host = start_fake_host(process_end).await; - let (mut rd, mut wr) = tokio::io::split(board_end); - let mut acc = FrameAccumulator::new(); + let event = + |seq: u64, c: u64| json!({"t":"event","sub":sub_id,"seq":seq,"ts":seq,"data":{"c":c}}); + send(&mut wr, event(1, 1)).await; + assert!(mirrors(&db, 1).await, "the first event reaches the mirror"); - let hello = next_frame(&mut rd, &mut acc).await; - assert_eq!(hello["method"], "hello"); + // The replies to the hello, late: a Welcome, then a refusal. + send(&mut wr, json!({"t":"reply","id":0,"ok":{}})).await; send( &mut wr, json!({"t":"reply","id":0,"err":"version_mismatch"}), ) .await; + send(&mut wr, event(2, 2)).await; + assert!( + mirrors(&db, 2).await, + "a late reply with id 0 must not end the subscription" + ); + host.abort(); +} - // Neither a subscribe nor another hello follows. - let mut buf = [0u8; 256]; - match tokio::time::timeout(Duration::from_millis(1500), rd.read(&mut buf)).await { - Err(_) | Ok(Ok(0)) => {} - Ok(r) => panic!( - "the client kept talking after a refusal: {r:?} {:?}", - &buf[..] - ), +/// Wait up to a second for the host's mirrored `temp` to read `c`. +async fn mirrors(db: &AimDb, c: u64) -> bool { + for _ in 0..50 { + if mirrored(db) == Some(c) { + return true; + } + tokio::time::sleep(Duration::from_millis(20)).await; } - host.abort(); + false } /// A host process mirroring `temp` over `stream`, which a test drives by hand. -async fn start_fake_host(stream: DuplexStream) -> JoinHandle<()> { +async fn start_fake_host(stream: DuplexStream) -> (Arc, JoinHandle<()>) { let mut builder = AimDbBuilder::new() .runtime(Arc::new(TokioAdapter)) .with_connector(SerialClient::new(TokioByteStream(stream))); @@ -240,8 +245,8 @@ async fn start_fake_host(stream: DuplexStream) -> JoinHandle<()> { }) .finish(); }); - let (_db, runner) = builder.build().await.expect("build host db"); - tokio::spawn(runner.run()) + let (db, runner) = builder.build().await.expect("build host db"); + (Arc::new(db), tokio::spawn(runner.run())) } async fn send(wr: &mut WriteHalf, v: Value) {