diff --git a/aimdb-core/CHANGELOG.md b/aimdb-core/CHANGELOG.md index 635c3753..c6f1c808 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 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/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..d8b109f5 100644 --- a/aimdb-core/src/session/client.rs +++ b/aimdb-core/src/session/client.rs @@ -69,8 +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, - /// 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, ahead of any caller command. + /// What that means depends on the codec: + /// + /// - 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, } @@ -88,6 +99,11 @@ impl Default for ClientConfig { } } +/// 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 /// [`ClientConfig::max_reconnect_delay`]. fn backoff_delay(config: &ClientConfig, attempt: usize) -> u64 { @@ -633,19 +649,35 @@ 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() { + // 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(); + 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, + } } } @@ -700,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 e76b4e72..2b314306 100644 --- a/aimdb-core/src/session/events.rs +++ b/aimdb-core/src/session/events.rs @@ -60,6 +60,7 @@ pub(crate) enum SessionEvent<'a> { topic: &'a str, skipped: u64, }, + HelloRefused, ReceiveFramingFailed, SendFrameRejected { bytes: usize, @@ -145,6 +146,10 @@ pub(crate) fn report(event: SessionEvent<'_>) { topic, skipped ), + SessionEvent::HelloRefused => log_error!( + target: target::CLIENT, + "server refused the hello: incompatible protocol version" + ), 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..e713bf38 100644 --- a/aimdb-core/src/session/mod.rs +++ b/aimdb-core/src/session/mod.rs @@ -559,6 +559,19 @@ 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. + /// + /// 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 c1f1a0ca..787be237 100644 --- a/aimdb-serial-connector/CHANGELOG.md +++ b/aimdb-serial-connector/CHANGELOG.md @@ -7,6 +7,18 @@ 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`, 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`. + ## [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..7e38c392 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`, 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. + /// /// 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..5fa07c55 100644 --- a/aimdb-serial-connector/src/lib.rs +++ b/aimdb-serial-connector/src/lib.rs @@ -31,6 +31,13 @@ //! 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`, 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 //! 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..16aabbcf --- /dev/null +++ b/aimdb-serial-connector/tests/client_restart.rs @@ -0,0 +1,271 @@ +//! 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; + } +} + +/// 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_goes_first_and_its_reply_is_not_awaited() { + let (board_end, process_end) = tokio::io::duplex(64 * 1024); + 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 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 without a Welcome, got {sub}"); + assert_eq!(sub["topic"], "temp"); + 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(); + + 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"); + + // 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(); +} + +/// 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; + } + false +} + +/// A host process mirroring `temp` over `stream`, which a test drives by hand. +async fn start_fake_host(stream: DuplexStream) -> (Arc, 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"); + (Arc::new(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]); + } +}