fix(session): AimX over serial — every reconnect gets its events (#328)

CHANGELOG.md

@@ -76,6 +76,14 @@ is unchanged. ([aimdb-core](aimdb-core/CHANGELOG.md))
 
 ### Fixed
 
+- **AimX over serial: every reconnect gets its events.** A client's `hello`
+  on a live connection starts a new session, an ended subscription no longer
+  cancels a newer one with the same id, and a subscription whose source ended
+  is reported instead of staying silent (#328).
+  ([aimdb-core](aimdb-core/CHANGELOG.md),
+  [aimdb-serial-connector](aimdb-serial-connector/CHANGELOG.md),
+  [aimdb-client](aimdb-client/CHANGELOG.md))
+
 - **The embedded MQTT backend advertises the largest packet it receives**
   (`Maximum Packet Size` 3,328 in CONNECT), so an oversized retained message is
   withheld by the broker instead of reconnecting the client forever.

aimdb-client/CHANGELOG.md

@@ -7,6 +7,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Fixed
+
+- **A subscription the server ended no longer reads as a transport failure
+  (#328).** A subscription stream's terminal `Internal` error now says "the
+  server ended the subscription (its source closed or could not be opened)";
+  it used to say "engine/transport failure", although a dropped connection
+  ends a subscription stream without an error. Calls keep the old text, which
+  fits them: for a call, the client engine raises `Internal` itself when the
+  connection drops.
+
 ## [0.7.0] - 2026-09-18
 
 ### Changed

aimdb-client/src/engine.rs

@@ -399,7 +399,7 @@ fn decode_updates(
             }),
             Err(e) => {
                 *terminated = true;
-                Err(rpc_err(e))
+                Err(sub_err(e))
             }
         };
         futures::future::ready(Some(out))
@@ -434,6 +434,24 @@ fn rpc_err(e: RpcError) -> ClientError {
     }
 }
 
+/// Map the terminal error of a subscription stream onto a [`ClientError`].
+///
+/// Unlike a call's error it always comes from the server: a dropped
+/// connection ends the stream without one. So `Internal` here is not a
+/// transport failure but the server ending the subscription, because its
+/// source closed or could not be opened (a record buffer with no subscriber
+/// slot left, for example).
+fn sub_err(e: RpcError) -> ClientError {
+    match e {
+        RpcError::NotFound | RpcError::Denied => rpc_err(e),
+        _ => ClientError::server_error(
+            "internal",
+            "the server ended the subscription (its source closed or could not be opened)",
+            None,
+        ),
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -542,4 +560,42 @@ mod tests {
             "unexpected terminal error: {err}"
         );
     }
+
+    /// A subscription the server ended (`Internal` as its terminal item) says
+    /// so, rather than blaming the transport: a dropped connection would have
+    /// ended the stream without an error.
+    #[tokio::test]
+    async fn a_subscription_the_server_ended_says_so() {
+        let raw = raw_stream(vec![ok_update("1", 0), Err(RpcError::Internal)]);
+        let out: Vec<_> = decode_updates(raw).collect().await;
+
+        assert_eq!(out.len(), 2);
+        let err = out[1].as_ref().expect_err("the end reaches the caller");
+        assert!(
+            matches!(
+                err,
+                ClientError::ServerError { code, message, .. }
+                    if code == "internal" && message.contains("ended the subscription")
+            ),
+            "unexpected terminal error: {err}"
+        );
+    }
+
+    /// A call's `Internal` keeps naming the engine or transport: for calls the
+    /// client engine raises it itself when the connection drops.
+    #[test]
+    fn a_calls_internal_error_still_names_the_transport() {
+        let err = rpc_err(RpcError::Internal);
+        assert!(
+            matches!(
+                &err,
+                ClientError::ServerError { message, .. } if message == "engine/transport failure"
+            ),
+            "{err}"
+        );
+        assert!(matches!(
+            sub_err(RpcError::Denied),
+            ClientError::ServerError { code, .. } if code == "denied"
+        ));
+    }
 }

aimdb-core/CHANGELOG.md

@@ -129,6 +129,36 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   never got a value, a join ran without that input, and with default
   features nothing was logged.
 
+### Fixed
+
+- **Every client on a serial line gets its events, not every other one (#328).**
+  A UART never closes, so the AimX session engine saw every client behind it
+  as one session: a departed client's subscriptions kept streaming and held
+  their buffer subscriber slots, and the next client's subscribe collided with
+  them by id. Three changes in `run_session`:
+  - **A second handshake starts a new session.** `Session::starts_session`
+    (new, default `false`; `AimxSession`: `hello`) names a client's handshake.
+    When one arrives on a connection that already had one, the engine drops
+    every subscription pump (releasing its buffer subscriber), discards their
+    queued events and opens a fresh session before answering. This applies
+    to every transport; elsewhere a second `hello` on one connection does not
+    happen with a well-behaved client.
+  - **An ended pump only removes its own subscription.** Subscriptions carry
+    a generation, so a pump ending late (after `unsub`, then `sub` with the
+    same id) no longer cancels the newer subscription. A `sub` whose id is
+    still live is refused with `Denied` instead of silently replacing it.
+  - **A subscription whose stream ends on its own is reported.** When its
+    source closes, or never opened (an Embassy SPMC ring with no subscriber
+    slot left), the client gets the remaining events and then an `Internal`
+    error under the subscribe id, which it turns into a terminal error,
+    instead of an acknowledged subscription that never delivers. An
+    unsubscribed subscription, or the previous client's after a new
+    handshake, still ends in silence.
+
+  A client that leaves without another one connecting is still invisible:
+  its subscriptions run until the next handshake, since the engine has no
+  timer to detect a silent peer.
+
 ## [2.0.0] - 2026-09-18
 
 ### Added

aimdb-core/src/session/aimx/dispatch.rs

@@ -102,6 +102,12 @@ impl DispatchReply {
 }
 
 impl Session for AimxSession {
+    /// `hello` opens every AimX session; a second one on a connection comes
+    /// from a new client (a serial line never closes between clients).
+    fn starts_session(&self, method: &str) -> bool {
+        method == "hello"
+    }
+
     fn call<'a>(
         &'a mut self,
         method: &'a str,

aimdb-core/src/session/mod.rs

@@ -475,6 +475,20 @@ pub trait Session: Send {
         Vec::new()
     }
 
+    /// Whether a request for `method` is a client's handshake, which starts a
+    /// new session. Defaulted to `false`.
+    ///
+    /// When a handshake arrives on a connection that already had one,
+    /// [`run_session`] treats it as a new client: it
+    /// cancels every subscription, discards their queued events and opens a
+    /// fresh session before answering. A transport that carries one peer for
+    /// its whole life (a serial line) has no other way to tell that the client
+    /// behind it changed.
+    fn starts_session(&self, method: &str) -> bool {
+        let _ = method;
+        false
+    }
+
     /// Fire-and-forget write: no reply. Routes through the producer/arbiter path,
     /// so single-writer-per-key stays intact.
     fn write<'a>(