Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 12 additions & 0 deletions aimdb-core/CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -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)

Expand Down
10 changes: 10 additions & 0 deletions aimdb-core/src/session/aimx/codec.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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)]
Expand Down
69 changes: 56 additions & 13 deletions aimdb-core/src/session/client.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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,
}

Expand All @@ -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 {
Expand Down Expand Up @@ -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,
}
}
}

Expand Down Expand Up @@ -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);
Expand Down
5 changes: 5 additions & 0 deletions aimdb-core/src/session/events.rs
Original file line number Diff line number Diff line change
Expand Up @@ -60,6 +60,7 @@ pub(crate) enum SessionEvent<'a> {
topic: &'a str,
skipped: u64,
},
HelloRefused,
ReceiveFramingFailed,
SendFrameRejected {
bytes: usize,
Expand Down Expand Up @@ -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,
Expand Down
13 changes: 13 additions & 0 deletions aimdb-core/src/session/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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<Outbound<'a>, 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
}
}

// ===========================================================================
Expand Down
12 changes: 12 additions & 0 deletions aimdb-serial-connector/CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
25 changes: 23 additions & 2 deletions aimdb-serial-connector/src/connector.rs
Original file line number Diff line number Diff line change
Expand Up @@ -72,14 +72,22 @@ impl Dialer for SerialPortDialer {
fn connect(&self) -> BoxFut<'_, TransportResult<Box<dyn Connection>>> {
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;
// otherwise the first frame is a stale leftover that fails to decode
// 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<dyn Connection>,
Expand All @@ -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
Expand All @@ -114,6 +129,7 @@ impl SerialClient {
.scheme(DEFAULT_SCHEME)
.with_config(ClientConfig {
reconnect: false,
sends_hello: true,
..ClientConfig::default()
})
}
Expand All @@ -122,14 +138,19 @@ 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<String>,
baud: u32,
) -> SessionClientConnector<SerialPortDialer, AimxCodec> {
SessionClientConnector::new(SerialPortDialer::new(path, baud), AimxCodec)
.scheme(DEFAULT_SCHEME)
.with_config(ClientConfig {
sends_hello: true,
..ClientConfig::default()
})
}
}

Expand Down
7 changes: 7 additions & 0 deletions aimdb-serial-connector/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
Loading
Loading