Feat/tcp connecotr hardening

CHANGELOG.md

@@ -18,6 +18,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 > - [aimdb-websocket-connector/CHANGELOG.md](aimdb-websocket-connector/CHANGELOG.md)
 > - [aimdb-uds-connector/CHANGELOG.md](aimdb-uds-connector/CHANGELOG.md)
 > - [aimdb-serial-connector/CHANGELOG.md](aimdb-serial-connector/CHANGELOG.md)
+> - [aimdb-tcp-connector/CHANGELOG.md](aimdb-tcp-connector/CHANGELOG.md)
 > - [aimdb-ws-protocol/CHANGELOG.md](aimdb-ws-protocol/CHANGELOG.md)
 > - [aimdb-wasm-adapter/CHANGELOG.md](aimdb-wasm-adapter/CHANGELOG.md)
 > - [aimdb-sync/CHANGELOG.md](aimdb-sync/CHANGELOG.md)

Makefile

@@ -179,6 +179,8 @@ test:
 	cargo test --package aimdb-serial-connector --no-default-features --features "embassy-runtime"
 	@printf "$(YELLOW)  → Testing TCP connector (tokio: length-prefix framing + AimX loopback)$(NC)\n"
 	cargo test --package aimdb-tcp-connector --no-default-features --features "_test-tokio"
+	@printf "$(YELLOW)  → Testing TCP connector (embassy: socket recycle + concurrent slots + redial over an embassy-net loopback)$(NC)\n"
+	cargo test --package aimdb-tcp-connector --no-default-features --features "_test-embassy-loopback" --test embassy_loopback
 
 fmt:
 	@printf "$(GREEN)Formatting code (workspace members only)...$(NC)\n"
@@ -280,6 +282,8 @@ clippy:
 	cargo clippy --package aimdb-tcp-connector --target thumbv7em-none-eabihf --target-dir $(EMBEDDED_CHECK_TARGET_DIR) --no-default-features --features "embassy-runtime" -- -D warnings
 	@printf "$(YELLOW)  → Clippy on TCP connector (embassy + defmt)$(NC)\n"
 	cargo clippy --package aimdb-tcp-connector --target thumbv7em-none-eabihf --target-dir $(EMBEDDED_CHECK_TARGET_DIR) --no-default-features --features "embassy-runtime,defmt" -- -D warnings
+	@printf "$(YELLOW)  → Clippy on TCP connector (embassy-net loopback smoke, host)$(NC)\n"
+	cargo clippy --package aimdb-tcp-connector --no-default-features --features "_test-embassy-loopback" --test embassy_loopback -- -D warnings
 	@printf "$(YELLOW)  → Clippy on WASM adapter$(NC)\n"
 	cargo clippy --package aimdb-wasm-adapter --target wasm32-unknown-unknown --features "wasm-runtime" -- -D warnings
 	@printf "$(YELLOW)  → Clippy on benchmarking infrastructure (host-only, incl. benches)$(NC)\n"

aimdb-tcp-connector/CHANGELOG.md

@@ -0,0 +1,15 @@
+# Changelog - aimdb-tcp-connector
+
+All notable changes to the `aimdb-tcp-connector` crate will be documented in this file.
+
+The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
+and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
+
+## [Unreleased]
+
+### Added
+
+- **New crate — the length-prefixed TCP transport for AimDB remote access (AimX over TCP, refs #121).** Contributes the `Dialer`/`Listener`/`Connection` transport triple plus thin `TcpClient`/`TcpServer` sugar; the AimX codec + dispatch and the runtime-neutral session engines (`run_client`/`serve`) are reused from `aimdb-core`. Every AimX envelope is framed as a `u32` big-endian length prefix. Two runtime halves:
+  - **`tokio-runtime`** (std, host/gateway) — TCP transport over `tokio::net`.
+  - **`embassy-runtime`** (`no_std + alloc`, MCU) — an explicit pool of caller-buffered `embassy-net` sockets, one accept/session worker per slot (`TcpServer::<N>::with_buffers`), with socket recycling across reconnects. A synchronous `accept()` failure (e.g. a port-0 endpoint rejected as `InvalidPort`) yields instead of spinning the cooperative executor.
+- **Embassy TCP runtime smoke test** (`_test-embassy-loopback`) — exercises the socket pool over two real `embassy-net` stacks wired by an in-memory driver-channel crossover: recycle → re-accept, concurrent accept slots, and dialer redial after a failed connect / dropped link.

aimdb-tcp-connector/Cargo.toml

@@ -34,6 +34,7 @@ embassy-runtime = [
     "aimdb-embassy-adapter/connectors",
     "aimdb-embassy-adapter/embassy-net-support",
     "dep:embassy-net",
+    "dep:embassy-futures",
     "dep:embedded-io-async",
 ]
 
@@ -42,21 +43,48 @@ defmt = ["aimdb-core/defmt"]
 
 _test-tokio = ["tokio-runtime", "dep:aimdb-tokio-adapter"]
 
+# Internal: the Embassy TCP half's runtime smoke (`tests/embassy_loopback.rs`)
+# stands up two real `embassy-net` stacks wired by an in-memory driver-channel
+# crossover, so it needs `medium-ip`/`proto-ipv4` and the host-only loopback
+# deps. Kept off `embassy-runtime` (production never pulls a network device or a
+# critical-section impl) and out of `[dev-dependencies]` (they would compile into
+# the `_test-tokio` build too). Run with `--features _test-embassy-loopback`.
+_test-embassy-loopback = [
+    "embassy-runtime",
+    "embassy-net/medium-ip",
+    "embassy-net/proto-ipv4",
+    "dep:embassy-net-driver-channel",
+    "dep:critical-section",
+]
+
 [dependencies]
 aimdb-core = { version = "1.1.0", path = "../aimdb-core", default-features = false }
 
 tokio = { workspace = true, optional = true, features = ["net", "io-util"] }
 
 aimdb-embassy-adapter = { version = "0.6.0", path = "../aimdb-embassy-adapter", default-features = false, optional = true }
 embassy-net = { workspace = true, optional = true }
+embassy-futures = { workspace = true, optional = true }
 embedded-io-async = { workspace = true, optional = true }
 
 aimdb-tokio-adapter = { version = "0.6.0", path = "../aimdb-tokio-adapter", optional = true }
 
+# test-only (see `_test-embassy-loopback`)
+embassy-net-driver-channel = { version = "0.4.0", path = "../_external/embassy/embassy-net-driver-channel", optional = true }
+critical-section = { version = "1.1", features = ["std"], optional = true }
+
 [dev-dependencies]
 tokio = { version = "1", features = ["rt-multi-thread", "macros", "time", "io-util", "net"] }
 serde = { workspace = true }
 serde_json = { workspace = true }
+# The Embassy loopback smoke expands `host_test_stubs!()` (no-op defmt logger +
+# host time driver) and drives the stack with `futures::executor::block_on`; the
+# macro references `defmt`/`embassy-time-driver` at the call site.
+futures = { workspace = true }
+defmt = { workspace = true }
+embassy-time-driver = { path = "../_external/embassy/embassy-time-driver" }
+# `StaticConfigV4.dns_servers` is a heapless 0.9 `Vec` (embassy-net's version).
+heapless = "0.9"
 
 [[example]]
 name = "tcp_demo"

aimdb-tcp-connector/src/embassy_transport.rs

@@ -25,6 +25,7 @@ use aimdb_core::session::{
 };
 use aimdb_embassy_adapter::connectors::{EmbassySessionClient, OneShotCell};
 use aimdb_embassy_adapter::SendFutureWrapper;
+use embassy_futures::yield_now;
 use embassy_net::tcp::TcpSocket;
 use embassy_net::{IpEndpoint, IpListenEndpoint, Stack};
 use embedded_io_async::Write;
@@ -114,6 +115,8 @@ impl Connection for TcpConnection {
 impl Drop for TcpConnection {
     fn drop(&mut self) {
         if let Some(mut socket) = self.socket.take() {
+            // Double abort is intentional: this one resets the link promptly on
+            // drop; the next taker re-aborts before reuse for a clean socket.
             socket.abort();
             if let Some(recycler) = &self.recycler {
                 recycler.put(socket);
@@ -141,14 +144,21 @@ where
 
 struct TcpSocketSlot {
     socket: RefCell<Option<TcpSocket<'static>>>,
+    // Single `Waker`: at most one accept ever waits on a given slot. Every shipped
+    // accept path holds one waiter per slot by construction — the `&mut self`
+    // `Listener` impl serializes accepts on the sole slot, and `TcpServer<N>` runs
+    // exactly one worker per slot. The pooled per-slot accept is test-only (see
+    // `accept_on`), and its single caller drives one accept per index. So this
+    // waker is never clobbered.
     waker: RefCell<Option<Waker>>,
 }
 
 // SAFETY: single-core cooperative Embassy executor; the socket and stack stay on
 // the same executor task set, and `RefCell` is never borrowed from another core.
 unsafe impl Send for TcpSocketSlot {}
-// SAFETY: same invariant. This is only shared so the dropped connection can
-// return its socket to the dialer-owned slot.
+// SAFETY: same invariant. Shared only so a dropped connection can return its
+// socket to the slot — owned by a dialer, an accept/session worker, or the
+// `Listener` compat path, never touched from more than one at a time.
 unsafe impl Sync for TcpSocketSlot {}
 
 impl TcpSocketSlot {
@@ -186,6 +196,47 @@ impl TcpSocketSlot {
     }
 }
 
+/// Owns a socket taken out of its slot for the duration of an `accept()`, and
+/// returns it to the slot if dropped before the accept succeeds. Without this,
+/// an accept future dropped mid-`accept` (a `select!` timeout or shutdown branch
+/// winning the race) would drop the socket instead of recycling it, leaving the
+/// slot permanently empty so every later accept on that index waits forever.
+/// [`SlotReturn::into_socket`] defuses it once the socket moves into a
+/// [`TcpConnection`] on the success path.
+struct SlotReturn<'a> {
+    slot: &'a Arc<TcpSocketSlot>,
+    socket: Option<TcpSocket<'static>>,
+}
+
+impl<'a> SlotReturn<'a> {
+    fn new(slot: &'a Arc<TcpSocketSlot>, socket: TcpSocket<'static>) -> Self {
+        Self {
+            slot,
+            socket: Some(socket),
+        }
+    }
+
+    fn socket_mut(&mut self) -> &mut TcpSocket<'static> {
+        self.socket
+            .as_mut()
+            .expect("socket present until into_socket")
+    }
+
+    /// Take the socket back, defusing the guard so its `Drop` becomes a no-op.
+    fn into_socket(mut self) -> TcpSocket<'static> {
+        self.socket.take().expect("socket taken exactly once")
+    }
+}
+
+impl Drop for SlotReturn<'_> {
+    fn drop(&mut self) {
+        if let Some(mut socket) = self.socket.take() {
+            socket.abort();
+            self.slot.put(socket);
+        }
+    }
+}
+
 /// A reusable Embassy TCP dialer backed by one caller-owned socket.
 ///
 /// Unlike moved-in UART peripherals, `embassy-net` TCP sockets can be reused
@@ -288,6 +339,26 @@ impl<const N: usize> TcpListener<N> {
         }
     }
 
+    /// Accept one connection on pooled socket `index`, recycling that socket back
+    /// into its slot when the returned connection drops. **Test-only** (gated on
+    /// `_test-embassy-loopback`): the shipped pooled path is `TcpServer<N>`, which
+    /// runs one worker per slot; this lets the transport-level loopback test drive
+    /// the same-port fan-out directly, without the session engine.
+    ///
+    /// The slot stores a single `Waker`, so this takes `&self` on the contract
+    /// that the caller drives **at most one accept per `index`** (the test uses
+    /// distinct indices). Panics if `index >= N`.
+    #[cfg(feature = "_test-embassy-loopback")]
+    #[doc(hidden)]
+    pub fn accept_on(
+        &self,
+        index: usize,
+    ) -> impl Future<Output = TransportResult<Box<dyn Connection>>> + Send + '_ {
+        let slot = self.slots[index].clone();
+        let local_endpoint = self.local_endpoint;
+        SendFutureWrapper(async move { accept_on_slot(&slot, local_endpoint).await })
+    }
+
     fn into_server_futures(
         self,
         codec: Arc<AimxCodec>,
@@ -312,25 +383,52 @@ impl<const N: usize> TcpListener<N> {
 
 impl Listener for TcpListener<1> {
     fn accept(&mut self) -> BoxFut<'_, TransportResult<Box<dyn Connection>>> {
+        // `&mut self` already serializes accepts on the sole slot, so the
+        // one-waiter-per-slot invariant holds here.
+        let slot = self.slots[0].clone();
+        let local_endpoint = self.local_endpoint;
         Box::pin(SendFutureWrapper(async move {
-            let slot = &self.slots[0];
-            let mut socket = poll_fn(|cx| slot.poll_take(cx)).await;
-            socket.abort();
-            match socket.accept(self.local_endpoint).await {
-                Ok(()) => {
-                    Ok(Box::new(TcpConnection::reusable(socket, slot.clone()))
-                        as Box<dyn Connection>)
-                }
-                Err(_) => {
-                    socket.abort();
-                    slot.put(socket);
-                    Err(TransportError::Io)
-                }
-            }
+            accept_on_slot(&slot, local_endpoint).await
         }))
     }
 }
 
+/// Take the socket from `slot`, accept one inbound connection on it, and hand
+/// back a recyclable [`TcpConnection`]. Shared by the [`Listener`] impl, the
+/// per-slot server workers, and the test-only `accept_on` so all pooled accept
+/// paths behave identically.
+async fn accept_on_slot(
+    slot: &Arc<TcpSocketSlot>,
+    local_endpoint: IpListenEndpoint,
+) -> TransportResult<Box<dyn Connection>> {
+    let socket = poll_fn(|cx| slot.poll_take(cx)).await;
+    // The socket lives in this guard until it either moves into a `TcpConnection`
+    // (success) or is returned to the slot. If the whole future is dropped while
+    // `accept()` is still pending, the guard's `Drop` recycles the socket so the
+    // slot is never left permanently empty.
+    let mut guard = SlotReturn::new(slot, socket);
+    guard.socket_mut().abort();
+    // Bind the result before matching so the `accept()` future's borrow of
+    // `guard` ends here, freeing `guard` for `into_socket` / `drop` below.
+    let accepted = guard.socket_mut().accept(local_endpoint).await;
+    match accepted {
+        Ok(()) => {
+            let socket = guard.into_socket();
+            Ok(Box::new(TcpConnection::reusable(socket, slot.clone())) as Box<dyn Connection>)
+        }
+        Err(_) => {
+            // Dropping `guard` aborts the socket and returns it to the slot.
+            drop(guard);
+            // `accept()` can fail synchronously (e.g. port-0 `InvalidPort`);
+            // without this await the caller re-enters `accept()` immediately with
+            // no yield point, starving the executor. Yield so a misconfig
+            // warn-loops instead of hanging.
+            yield_now().await;
+            Err(TransportError::Io)
+        }
+    }
+}
+
 async fn serve_socket_slot(
     slot: Arc<TcpSocketSlot>,
     local_endpoint: IpListenEndpoint,
@@ -339,18 +437,10 @@ async fn serve_socket_slot(
     config: SessionConfig,
 ) {
     loop {
-        let mut socket = poll_fn(|cx| slot.poll_take(cx)).await;
-        socket.abort();
-        match socket.accept(local_endpoint).await {
-            Ok(()) => {
-                let conn =
-                    Box::new(TcpConnection::reusable(socket, slot.clone())) as Box<dyn Connection>;
-                run_session(conn, codec.as_ref(), dispatch.as_ref(), &config).await;
-            }
-            Err(_) => {
-                socket.abort();
-                slot.put(socket);
-            }
+        // `accept_on_slot` already yields on the synchronous-failure path, so a
+        // misconfig warn-loops here instead of starving the executor.
+        if let Ok(conn) = accept_on_slot(&slot, local_endpoint).await {
+            run_session(conn, codec.as_ref(), dispatch.as_ref(), &config).await;
         }
     }
 }