From ec75b8622a2764b3d133f127bc65e6fe4441a894 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Alexander=20Schn=C3=B6rch?= Date: Sun, 6 Sep 2026 06:41:49 +0000 Subject: [PATCH 1/5] feat(tcp-connector): implement runtime-neutral TCP client and server connectors with length-prefix framing --- aimdb-core/src/session/io.rs | 13 ++ aimdb-embassy-adapter/src/net.rs | 1 + aimdb-tcp-connector/Cargo.toml | 2 +- aimdb-tcp-connector/src/connector.rs | 183 ++++++++++++++++++ aimdb-tcp-connector/src/framing.rs | 57 ++++++ aimdb-tcp-connector/src/lib.rs | 6 + .../tests/connector_roundtrip.rs | 111 +++++++++++ aimdb-tokio-adapter/src/net.rs | 1 + 8 files changed, 373 insertions(+), 1 deletion(-) create mode 100644 aimdb-tcp-connector/src/connector.rs create mode 100644 aimdb-tcp-connector/tests/connector_roundtrip.rs diff --git a/aimdb-core/src/session/io.rs b/aimdb-core/src/session/io.rs index 6b3d99bb..e8caf7e4 100644 --- a/aimdb-core/src/session/io.rs +++ b/aimdb-core/src/session/io.rs @@ -273,6 +273,19 @@ pub struct FramingDialer { port: u16, } +/// `SessionClientConnector` clones its dialer per build, so a framed one must +/// clone too. +impl Clone for FramingDialer { + fn clone(&self) -> Self { + Self { + dialer: self.dialer.clone(), + framers: self.framers.clone(), + host: self.host.clone(), + port: self.port, + } + } +} + impl FramingDialer { /// Dial `host:port` through `dialer`, framing each stream with a framer /// from `framers`. diff --git a/aimdb-embassy-adapter/src/net.rs b/aimdb-embassy-adapter/src/net.rs index cac48efe..f839d206 100644 --- a/aimdb-embassy-adapter/src/net.rs +++ b/aimdb-embassy-adapter/src/net.rs @@ -209,6 +209,7 @@ impl ByteStream for EmbassyTcpStream { } /// Dials TCP connections over one caller-owned socket. +#[derive(Clone)] pub struct EmbassyTcpDialer { slot: Arc, } diff --git a/aimdb-tcp-connector/Cargo.toml b/aimdb-tcp-connector/Cargo.toml index 563f8705..c6dedd5c 100644 --- a/aimdb-tcp-connector/Cargo.toml +++ b/aimdb-tcp-connector/Cargo.toml @@ -42,7 +42,7 @@ embassy-runtime = [ tracing = ["aimdb-core/tracing"] defmt = ["aimdb-core/defmt"] -_test-tokio = ["tokio-runtime", "dep:aimdb-tokio-adapter"] +_test-tokio = ["tokio-runtime", "dep:aimdb-tokio-adapter", "aimdb-tokio-adapter/net"] # 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 diff --git a/aimdb-tcp-connector/src/connector.rs b/aimdb-tcp-connector/src/connector.rs new file mode 100644 index 00000000..f8132162 --- /dev/null +++ b/aimdb-tcp-connector/src/connector.rs @@ -0,0 +1,183 @@ +//! Runtime-neutral TCP client and server sugar. +//! +//! Both are generic over core's [`StreamDialer`] / [`StreamListener`], so the +//! socket comes from an adapter and this crate contributes only the +//! length-prefix [`LengthFramer`]. + +use alloc::boxed::Box; +use alloc::string::{String, ToString}; +use alloc::sync::Arc; +use alloc::vec::Vec; +use core::future::Future; +use core::pin::Pin; + +use aimdb_core::connector::ConnectorBuilder; +use aimdb_core::remote::{AimxConfig, SecurityPolicy}; +use aimdb_core::session::aimx::{AimxCodec, AimxDispatch}; +use aimdb_core::session::{ + Dispatch, FramingDialer, FramingListener, OneShot, SessionClientConnector, SessionConfig, + SessionLimits, SessionServerConnector, StreamDialer, StreamListener, +}; +use aimdb_core::{AimDb, DbError, DbResult}; + +use crate::framing::{LengthFramer, DEFAULT_MAX_FRAME}; +use crate::DEFAULT_SCHEME; + +type BoxFuture = Pin + Send + 'static>>; +type BuildFuture<'a> = Pin>> + Send + 'a>>; + +/// Per-`read` chunk handed to the byte stream. +pub const READ_CHUNK: usize = 1024; +/// Per-`write_all` chunk. +pub const WRITE_CHUNK: usize = 1024; + +/// The dialer half, framed. +pub type TcpFramingDialer = FramingDialer LengthFramer, READ_CHUNK, WRITE_CHUNK>; +/// The listener half, framed. +pub type TcpFramingListener = FramingListener LengthFramer, READ_CHUNK, WRITE_CHUNK>; + +/// Constructs a TCP session client connector over an adapter's dialer. +pub struct TcpClient; + +impl TcpClient { + /// Mirror records to and from an AimX peer at `host:port`. + /// + /// `dialer` comes from an adapter (`TokioNet::tcp()`, `EmbassyNet::tcp(..)`), + /// which also resolves the host. + #[allow(clippy::new_ret_no_self)] + pub fn new( + dialer: D, + host: impl Into, + port: u16, + ) -> SessionClientConnector, AimxCodec> { + SessionClientConnector::new( + FramingDialer::new( + dialer, + LengthFramer::new as fn() -> LengthFramer, + host, + port, + ), + AimxCodec, + ) + .scheme(DEFAULT_SCHEME) + } +} + +/// Accepts AimX connections over an adapter's TCP listener. +/// +/// The listener is moved in, so it is taken once at `build`; a second `build` +/// fails rather than silently serving nothing. +pub struct TcpServer { + listener: OneShot, + config: AimxConfig, + scheme: String, +} + +impl TcpServer { + /// Serve AimX on an already-bound listener. + /// + /// Prefer loopback bind addresses unless the deployment provides its own + /// network-layer protection. + pub fn new(listener: L) -> Self { + Self { + listener: OneShot::new(listener), + config: AimxConfig::uds_default(), + scheme: DEFAULT_SCHEME.to_string(), + } + } + + /// Use a prepared [`AimxConfig`] for limits and security policy. + pub fn with_config(mut self, config: AimxConfig) -> Self { + self.config = config; + self + } + + /// Set the security policy. + pub fn security_policy(mut self, policy: SecurityPolicy) -> Self { + self.config = self.config.security_policy(policy); + self + } + + /// Maximum concurrently served connections. + pub fn max_connections(mut self, max: usize) -> Self { + self.config = self.config.max_connections(max); + self + } + + /// Maximum live subscriptions per connection. + pub fn max_subs_per_connection(mut self, max: usize) -> Self { + self.config = self.config.max_subs_per_connection(max); + self + } + + /// Override the scheme this connector registers. + pub fn scheme(mut self, scheme: impl Into) -> Self { + self.scheme = scheme.into(); + self + } +} + +impl ConnectorBuilder for TcpServer +where + L: StreamListener + Send + 'static, + L::Stream: 'static, +{ + fn build<'a>(&'a self, db: &'a AimDb) -> BuildFuture<'a> { + let config = self.config.clone(); + let scheme = self.scheme.clone(); + Box::pin(async move { + // Taken on first poll, not at call time: a `build()` future dropped + // before it is polled must leave the listener where it was, or a + // later build fails having never served anything. + let listener = self + .listener + .take() + .ok_or_else(|| DbError::InvalidOperation { + operation: "TcpServer::build".to_string(), + reason: "the moved-in listener was already taken; build() ran twice" + .to_string(), + })?; + let session_config = SessionConfig { + limits: SessionLimits { + max_connections: config.max_connections, + max_subs_per_connection: config.max_subs_per_connection, + }, + reads_hello: false, + acks_subscribe: false, + }; + let framed: OneShot> = OneShot::new(FramingListener::new( + listener, + LengthFramer::new as fn() -> LengthFramer, + )); + let dispatch_config = config; + let connector = SessionServerConnector::new( + move || { + framed.take().ok_or_else(|| DbError::InvalidOperation { + operation: "TcpServer::build".to_string(), + reason: "the moved-in listener was already taken".to_string(), + }) + }, + AimxCodec, + move |db: &AimDb| -> Arc { + crate::apply_writable(db, &dispatch_config); + Arc::new(AimxDispatch::new( + Arc::new(db.clone()), + dispatch_config.clone(), + )) + }, + session_config, + ) + .scheme(scheme); + connector.build(db).await + }) + } + + fn scheme(&self) -> &str { + &self.scheme + } +} + +/// The default payload bound a caller gets when it does not set one. +pub const fn default_max_frame() -> usize { + DEFAULT_MAX_FRAME +} diff --git a/aimdb-tcp-connector/src/framing.rs b/aimdb-tcp-connector/src/framing.rs index 97e8630c..d02fd5cd 100644 --- a/aimdb-tcp-connector/src/framing.rs +++ b/aimdb-tcp-connector/src/framing.rs @@ -91,3 +91,60 @@ impl FrameAccumulator { Some(Ok(self.buf.drain(..len).collect())) } } + +/// Length-prefix framing against core's [`Framer`](aimdb_core::session::Framer), +/// so one framer serves both runtimes. +/// +/// Unlike a self-synchronizing format, a length prefix has no delimiter to +/// resync on, so a framing error is fatal: `next_frame` reports it once and the +/// accumulator is left empty rather than pretending the stream is still +/// aligned. +#[cfg(any(feature = "tokio-runtime", feature = "embassy-runtime"))] +pub struct LengthFramer { + acc: FrameAccumulator, + max_frame: usize, +} + +#[cfg(any(feature = "tokio-runtime", feature = "embassy-runtime"))] +impl LengthFramer { + /// A framer bounded by [`DEFAULT_MAX_FRAME`]. + pub fn new() -> Self { + Self::with_max_frame(DEFAULT_MAX_FRAME) + } + + /// A framer bounded by `max_frame` payload bytes. + pub fn with_max_frame(max_frame: usize) -> Self { + Self { + acc: FrameAccumulator::with_max_frame(max_frame), + max_frame, + } + } +} + +#[cfg(any(feature = "tokio-runtime", feature = "embassy-runtime"))] +impl Default for LengthFramer { + fn default() -> Self { + Self::new() + } +} + +#[cfg(any(feature = "tokio-runtime", feature = "embassy-runtime"))] +impl aimdb_core::session::Framer for LengthFramer { + fn encode(&self, frame: &[u8], out: &mut Vec) { + // Core's `Framer::encode` is infallible, so an oversized frame is + // dropped here rather than written half-encoded: the peer would read a + // length prefix with no payload behind it and desync permanently. + if frame.len() > self.max_frame { + return; + } + let _ = encode_frame(frame, out); + } + + fn push_bytes(&mut self, bytes: &[u8]) { + self.acc.push_bytes(bytes); + } + + fn next_frame(&mut self) -> Option, ()>> { + self.acc.next_frame().map(|r| r.map_err(|_| ())) + } +} diff --git a/aimdb-tcp-connector/src/lib.rs b/aimdb-tcp-connector/src/lib.rs index 82b14495..d144cb3d 100644 --- a/aimdb-tcp-connector/src/lib.rs +++ b/aimdb-tcp-connector/src/lib.rs @@ -15,6 +15,12 @@ extern crate alloc; pub mod framing; +// Runtime-neutral `TcpClient`/`TcpServer` over an adapter's stream transports. +#[cfg(any(feature = "tokio-runtime", feature = "embassy-runtime"))] +pub mod connector; + +// Superseded by `connector` over the adapters' stream transports; both are +// deleted once the tests and examples move across. #[cfg(feature = "tokio-runtime")] pub mod tokio_transport; diff --git a/aimdb-tcp-connector/tests/connector_roundtrip.rs b/aimdb-tcp-connector/tests/connector_roundtrip.rs new file mode 100644 index 00000000..ef373da2 --- /dev/null +++ b/aimdb-tcp-connector/tests/connector_roundtrip.rs @@ -0,0 +1,111 @@ +//! `TcpClient`/`TcpServer` over the adapter's stream transports, end to end +//! through a real `AimDb`. +//! +//! The socket comes from `TokioNet`; this crate supplies only the length-prefix +//! framer. Mirrors `tokio_roundtrip.rs`, which still drives the runtime-specific +//! transports. +#![cfg(feature = "_test-tokio")] + +use std::sync::Arc; +use std::time::Duration; + +use aimdb_core::buffer::BufferCfg; +use aimdb_core::connector::ConnectorBuilder; +use aimdb_core::AimDbBuilder; +use aimdb_tcp_connector::connector::{TcpClient, TcpServer}; +use aimdb_tokio_adapter::net::TokioNet; +use aimdb_tokio_adapter::{TokioAdapter, TokioRecordRegistrarExt}; +use serde::{Deserialize, Serialize}; + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +struct Setting { + level: u64, +} + +async fn db() -> Arc { + let mut builder = AimDbBuilder::new().runtime(Arc::new(TokioAdapter)); + builder.configure::("setting", |reg| { + reg.buffer(BufferCfg::SingleLatest).with_remote_access(); + }); + let (db, _runner) = builder.build().await.expect("build db"); + Arc::new(db) +} + +/// The server accepts on an adapter listener and serves AimX over it. +#[tokio::test] +async fn server_serves_over_an_adapter_listener() { + let listener = TokioNet::listen("127.0.0.1:0").await.expect("bind"); + let addr = listener.local_addr().expect("bound addr"); + + let db = db().await; + let server = TcpServer::new(listener); + let futures = server.build(&db).await.expect("build server"); + assert_eq!(futures.len(), 1, "one serve future"); + let serving = tokio::spawn(async move { + for f in futures { + f.await; + } + }); + + // A bare TCP connect proves the listener is live and accepting. + let peer = tokio::time::timeout(Duration::from_secs(5), tokio::net::TcpStream::connect(addr)) + .await + .expect("connect timed out") + .expect("connect"); + assert!(peer.peer_addr().is_ok()); + + serving.abort(); +} + +/// The client registers under the TCP scheme and builds its pump futures. +#[tokio::test] +async fn client_builds_over_an_adapter_dialer() { + let listener = TokioNet::listen("127.0.0.1:0").await.expect("bind"); + let port = listener.local_addr().expect("bound addr").port(); + + let db = db().await; + let client = TcpClient::new(TokioNet::tcp(), "127.0.0.1", port); + assert_eq!(ConnectorBuilder::scheme(&client), "tcp"); + + let futures = client.build(&db).await.expect("build client"); + assert!( + !futures.is_empty(), + "client contributes at least one future" + ); +} + +/// The listener is moved in, so a second `build` is refused rather than +/// silently serving nothing. +#[tokio::test] +async fn a_second_build_is_refused() { + let listener = TokioNet::listen("127.0.0.1:0").await.expect("bind"); + let db = db().await; + let server = TcpServer::new(listener); + + server.build(&db).await.expect("first build"); + let Err(err) = server.build(&db).await else { + panic!("a second build must fail"); + }; + assert!( + format!("{err}").contains("already taken"), + "unexpected error: {err}" + ); +} + +/// A `build()` future dropped before it is polled must not consume the +/// listener — otherwise a lost `select!` arm or an unrelated builder error +/// leaves the server permanently unbuildable. +#[tokio::test] +async fn an_unpolled_build_leaves_the_listener_in_place() { + let listener = TokioNet::listen("127.0.0.1:0").await.expect("bind"); + let db = db().await; + let server = TcpServer::new(listener); + + drop(server.build(&db)); + + let futures = server + .build(&db) + .await + .expect("the listener must survive an unpolled build"); + assert_eq!(futures.len(), 1); +} diff --git a/aimdb-tokio-adapter/src/net.rs b/aimdb-tokio-adapter/src/net.rs index 574d6f09..1b370e48 100644 --- a/aimdb-tokio-adapter/src/net.rs +++ b/aimdb-tokio-adapter/src/net.rs @@ -78,6 +78,7 @@ where } /// Dials TCP connections. +#[derive(Clone, Copy, Default)] pub struct TokioTcpDialer; impl StreamDialer for TokioTcpDialer { From 34730f2fa8e3fb4637f21e8a46c447bfc5bdfaac Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Alexander=20Schn=C3=B6rch?= Date: Sun, 6 Sep 2026 07:14:08 +0000 Subject: [PATCH 2/5] feat(tcp-connector): refactor TCP connectors to use framed dialer and listener for improved framing support --- aimdb-tcp-connector/examples/tcp_demo.rs | 19 ++- aimdb-tcp-connector/src/connector.rs | 31 +++-- aimdb-tcp-connector/tests/embassy_loopback.rs | 124 +++++++++++------- aimdb-tcp-connector/tests/tokio_roundtrip.rs | 11 +- 4 files changed, 117 insertions(+), 68 deletions(-) diff --git a/aimdb-tcp-connector/examples/tcp_demo.rs b/aimdb-tcp-connector/examples/tcp_demo.rs index c66014c0..a3d6af7e 100644 --- a/aimdb-tcp-connector/examples/tcp_demo.rs +++ b/aimdb-tcp-connector/examples/tcp_demo.rs @@ -25,7 +25,8 @@ use aimdb_core::remote::{AimxConfig, SecurityPolicy}; use aimdb_core::session::aimx::AimxCodec; use aimdb_core::session::{run_client, ClientConfig, Payload}; use aimdb_core::AimDbBuilder; -use aimdb_tcp_connector::tokio_transport::{TcpDialer, TcpServer}; +use aimdb_tcp_connector::connector::{framed_dialer, TcpServer}; +use aimdb_tokio_adapter::net::TokioNet; use aimdb_tokio_adapter::{TokioAdapter, TokioRecordRegistrarExt}; use serde::{Deserialize, Serialize}; use serde_json::json; @@ -73,9 +74,12 @@ async fn run_server(bind_addr: String) { .max_connections(8) .max_subs_per_connection(32); + let listener = TokioNet::listen(&bind_addr) + .await + .expect("bind the TCP listener"); let mut builder = AimDbBuilder::new() .runtime(Arc::new(TokioAdapter)) - .with_connector(TcpServer::new(bind_addr).with_config(config)); + .with_connector(TcpServer::new(listener).with_config(config)); builder.configure::("counter", |reg| { reg.buffer(BufferCfg::SingleLatest).with_remote_access(); }); @@ -147,8 +151,9 @@ async fn run_set_mode(endpoint: String, level: u64) { } fn connect(endpoint: String) -> aimdb_core::session::ClientHandle { + let (host, port) = split_endpoint(&endpoint); let (handle, engine) = run_client( - TcpDialer::new(endpoint), + framed_dialer(TokioNet::tcp(), host, port), AimxCodec, ClientConfig { sends_hello: false, @@ -159,3 +164,11 @@ fn connect(endpoint: String) -> aimdb_core::session::ClientHandle { tokio::spawn(engine); handle } + +/// Split `host:port`, defaulting to the AimX TCP port when none is given. +fn split_endpoint(endpoint: &str) -> (String, u16) { + match endpoint.rsplit_once(':') { + Some((host, port)) => (host.to_string(), port.parse().unwrap_or(7001)), + None => (endpoint.to_string(), 7001), + } +} diff --git a/aimdb-tcp-connector/src/connector.rs b/aimdb-tcp-connector/src/connector.rs index f8132162..9717556b 100644 --- a/aimdb-tcp-connector/src/connector.rs +++ b/aimdb-tcp-connector/src/connector.rs @@ -20,7 +20,7 @@ use aimdb_core::session::{ }; use aimdb_core::{AimDb, DbError, DbResult}; -use crate::framing::{LengthFramer, DEFAULT_MAX_FRAME}; +use crate::framing::LengthFramer; use crate::DEFAULT_SCHEME; type BoxFuture = Pin + Send + 'static>>; @@ -36,6 +36,25 @@ pub type TcpFramingDialer = FramingDialer LengthFramer, READ_CHUNK /// The listener half, framed. pub type TcpFramingListener = FramingListener LengthFramer, READ_CHUNK, WRITE_CHUNK>; +/// Frame an adapter's dialer for `host:port` with length-prefix framing. +pub fn framed_dialer( + dialer: D, + host: impl Into, + port: u16, +) -> TcpFramingDialer { + FramingDialer::new( + dialer, + LengthFramer::new as fn() -> LengthFramer, + host, + port, + ) +} + +/// Frame an adapter's listener with length-prefix framing. +pub fn framed_listener(listener: L) -> TcpFramingListener { + FramingListener::new(listener, LengthFramer::new as fn() -> LengthFramer) +} + /// Constructs a TCP session client connector over an adapter's dialer. pub struct TcpClient; @@ -145,10 +164,7 @@ where reads_hello: false, acks_subscribe: false, }; - let framed: OneShot> = OneShot::new(FramingListener::new( - listener, - LengthFramer::new as fn() -> LengthFramer, - )); + let framed = OneShot::new(framed_listener(listener)); let dispatch_config = config; let connector = SessionServerConnector::new( move || { @@ -176,8 +192,3 @@ where &self.scheme } } - -/// The default payload bound a caller gets when it does not set one. -pub const fn default_max_frame() -> usize { - DEFAULT_MAX_FRAME -} diff --git a/aimdb-tcp-connector/tests/embassy_loopback.rs b/aimdb-tcp-connector/tests/embassy_loopback.rs index a8986afc..2f36e60b 100644 --- a/aimdb-tcp-connector/tests/embassy_loopback.rs +++ b/aimdb-tcp-connector/tests/embassy_loopback.rs @@ -1,15 +1,13 @@ -//! Runtime smoke for the Embassy TCP half (feature `_test-embassy-loopback`). +//! Runtime smoke for the Embassy TCP path (feature `_test-embassy-loopback`). //! -//! The transport is welded to a concrete `embassy_net::tcp::TcpSocket` with no -//! seam for a fake, so socket recycling and waker handoff can only be exercised -//! over a real stack. Two `embassy-net` stacks wired by an in-memory -//! `embassy-net-driver-channel` crossover drive the real -//! `TcpListener`/`TcpDialer`/`TcpConnection` triple under `block_on`: +//! Socket recycling and waker handoff can only be exercised over a real stack, +//! so two `embassy-net` stacks wired by an in-memory +//! `embassy-net-driver-channel` crossover drive the adapter's transports under +//! the connector's framing, via `block_on`: //! //! - recycle: accept -> exchange -> drop -> re-accept (`recycle_then_reaccept`); -//! - concurrency: one pooled `N = 2` listener keeps both sockets in `accept()` -//! on a single port (via the test-only `accept_on`, one accept per index) while -//! two clients dial it (`two_concurrent_sessions`); +//! - concurrency: one pooled `N = 2` listener serves two clients on a single +//! port across sequential accepts (`two_concurrent_sessions`); //! - redial: after a failed connect and after a dropped link //! (`dialer_redials_after_failure_and_drop`); //! - cancellation: a cancelled accept returns its socket to the slot @@ -21,8 +19,9 @@ extern crate alloc; use core::future::Future; use aimdb_core::session::{Connection, Dialer, Listener}; -use aimdb_tcp_connector::{TcpDialer, TcpListener}; -use embassy_net::{Config, IpAddress, IpEndpoint, Ipv4Address, Ipv4Cidr, Stack, StaticConfigV4}; +use aimdb_embassy_adapter::net::EmbassyNet; +use aimdb_tcp_connector::connector::{framed_dialer, framed_listener}; +use embassy_net::{Config, Ipv4Address, Ipv4Cidr, Stack, StaticConfigV4}; use embassy_net_driver_channel as ch; use embassy_net_driver_channel::driver::{HardwareAddress, LinkState}; @@ -180,9 +179,8 @@ where }); } -fn endpoint(port: u16) -> IpEndpoint { - IpEndpoint::new(IpAddress::Ipv4(SERVER_IP), port) -} +/// As `StreamDialer::connect` takes it: a host string the adapter resolves. +const SERVER_HOST: &str = "192.168.0.1"; /// Exchange one framed request + reply over an already-connected pair, asserting /// the framing round-trips both ways. @@ -232,8 +230,16 @@ async fn send_and_verify(client: &mut dyn Connection, tag: &[u8]) { #[test] fn recycle_then_reaccept() { drive(|server_stack, client_stack| async move { - let mut listener = TcpListener::new(server_stack, 7000u16, buf(), buf()); - let dialer = TcpDialer::new(client_stack, endpoint(7000), buf(), buf()); + let mut listener = framed_listener(EmbassyNet::listen::<1>( + server_stack, + 7000u16, + [(buf(), buf())], + )); + let dialer = framed_dialer( + EmbassyNet::tcp(client_stack, buf(), buf()), + SERVER_HOST, + 7000, + ); // First connection over the single pooled socket. let (accepted, connected) = futures::join!(listener.accept(), dialer.connect()); @@ -255,34 +261,40 @@ fn recycle_then_reaccept() { }); } -/// One pooled `N = 2` listener keeps both of its sockets in `accept()` on a -/// single port while two clients dial that port at once — the same-port fan-out -/// and pooled worker creation that `TcpListener::::with_buffers` exists for. -/// Each client lands on a distinct pooled slot; a broken pool (only one socket -/// accepting, or both racing to the same slot) would hang the second session and -/// trip the watchdog. (Wiring the pool into the AimX session engine via -/// `TcpServer` is intentionally outside this transport-level smoke test.) +/// One pooled `N = 2` listener serves two clients on a single port. A broken +/// pool — only one socket listening, or both racing the same slot — would hang +/// the second session and trip the watchdog. +/// +/// `accept_pool.rs` covers the sharper property (every slot stays in `LISTEN` +/// *between* accepts, with a negative control); this adds framing on top. #[test] fn two_concurrent_sessions() { drive(|server_stack, client_stack| async move { // One pooled listener, two sockets, both bound to port 7001. - let listener = - TcpListener::<2>::with_buffers(server_stack, 7001u16, [buf(), buf()], [buf(), buf()]); - let dialer_a = TcpDialer::new(client_stack, endpoint(7001), buf(), buf()); - let dialer_b = TcpDialer::new(client_stack, endpoint(7001), buf(), buf()); - - // Drive the pooled sockets directly via the test-only `accept_on`, one - // accept per index (its single-caller-per-index contract): both sockets - // accept on 7001 while both clients dial it, each landing on its own slot. - let (a_srv, b_srv, a_cli, b_cli) = futures::join!( - listener.accept_on(0), - listener.accept_on(1), - dialer_a.connect(), - dialer_b.connect(), + let mut listener = framed_listener(EmbassyNet::listen::<2>( + server_stack, + 7001u16, + [(buf(), buf()), (buf(), buf())], + )); + let dialer_a = framed_dialer( + EmbassyNet::tcp(client_stack, buf(), buf()), + SERVER_HOST, + 7001, ); - let mut a_srv = a_srv.expect("accept slot 0"); - let mut b_srv = b_srv.expect("accept slot 1"); + let dialer_b = framed_dialer( + EmbassyNet::tcp(client_stack, buf(), buf()), + SERVER_HOST, + 7001, + ); + + // No per-index hook any more: the pool keeps every slot listening across + // calls, so two sequential accepts serve both clients. + let (a_srv, a_cli) = futures::join!(listener.accept(), dialer_a.connect()); + let mut a_srv = a_srv.expect("accept A"); let mut a_cli = a_cli.expect("connect A"); + + let (b_srv, b_cli) = futures::join!(listener.accept(), dialer_b.connect()); + let mut b_srv = b_srv.expect("accept B"); let mut b_cli = b_cli.expect("connect B"); // Drive both sessions at once. Servers echo (the stack picks the pairing); @@ -302,7 +314,11 @@ fn two_concurrent_sessions() { #[test] fn dialer_redials_after_failure_and_drop() { drive(|server_stack, client_stack| async move { - let dialer = TcpDialer::new(client_stack, endpoint(7003), buf(), buf()); + let dialer = framed_dialer( + EmbassyNet::tcp(client_stack, buf(), buf()), + SERVER_HOST, + 7003, + ); // No socket is listening on 7003 yet -> the server stack RSTs the SYN -> // connect fails. The dialer must recycle its socket for a redial. @@ -312,7 +328,11 @@ fn dialer_redials_after_failure_and_drop() { ); // Bring a listener up; the recycled dialer socket now connects. - let mut listener = TcpListener::new(server_stack, 7003u16, buf(), buf()); + let mut listener = framed_listener(EmbassyNet::listen::<1>( + server_stack, + 7003u16, + [(buf(), buf())], + )); let (accepted, connected) = futures::join!(listener.accept(), dialer.connect()); let mut server = accepted.expect("accept after listener up"); let mut client = connected.expect("redial after failed connect"); @@ -329,17 +349,25 @@ fn dialer_redials_after_failure_and_drop() { } /// A cancelled accept — its future dropped mid-`accept()`, as a `select!` timeout -/// or shutdown branch would drop it — must return the pooled socket to its slot. -/// Without the drop guard the socket is dropped instead of recycled, the slot -/// stays empty, and the follow-up accept below would hang until the watchdog. +/// or shutdown branch would drop it — must leave the pool able to accept again. +/// The stored accepts survive the outer future's cancellation; a slot leaked +/// instead would hang the follow-up accept until the watchdog. #[test] fn cancelled_accept_recycles_socket() { use futures::future::{ready, select, Either}; use futures::pin_mut; drive(|server_stack, client_stack| async move { - let mut listener = TcpListener::new(server_stack, 7005u16, buf(), buf()); - let dialer = TcpDialer::new(client_stack, endpoint(7005), buf(), buf()); + let mut listener = framed_listener(EmbassyNet::listen::<1>( + server_stack, + 7005u16, + [(buf(), buf())], + )); + let dialer = framed_dialer( + EmbassyNet::tcp(client_stack, buf(), buf()), + SERVER_HOST, + 7005, + ); // No client dials 7005, so this accept takes the pooled socket and then // parks in `TcpSocket::accept`. Cancel it by letting a ready future win a @@ -362,7 +390,5 @@ fn cancelled_accept_recycles_socket() { }); } -// Note: there is no shipped public multi-slot accept to misuse — the pooled path -// is `TcpServer` (one worker per slot) and the `accept_on` used above is a -// test-only, single-caller-per-index hook. So the one-waiter-per-slot invariant -// is upheld by construction and there is nothing to assert at runtime. +// The pool exposes a single `accept()`, so there is no per-slot entry point to +// misuse and the one-waiter-per-slot invariant holds by construction. diff --git a/aimdb-tcp-connector/tests/tokio_roundtrip.rs b/aimdb-tcp-connector/tests/tokio_roundtrip.rs index 1755362e..048b256d 100644 --- a/aimdb-tcp-connector/tests/tokio_roundtrip.rs +++ b/aimdb-tcp-connector/tests/tokio_roundtrip.rs @@ -10,7 +10,8 @@ use aimdb_core::session::{ run_client, serve, ClientConfig, Dispatch, Payload, SessionConfig, SessionLimits, }; use aimdb_core::AimDbBuilder; -use aimdb_tcp_connector::tokio_transport::{TcpDialer, TcpListener}; +use aimdb_tcp_connector::connector::{framed_dialer, framed_listener}; +use aimdb_tokio_adapter::net::TokioNet; use aimdb_tokio_adapter::{TokioAdapter, TokioRecordRegistrarExt}; use serde::{Deserialize, Serialize}; use serde_json::json; @@ -32,9 +33,7 @@ async fn aimx_roundtrips_over_tcp_loopback() { db.set_record_from_json("setting", json!({ "level": 42 })) .expect("seed setting"); - let listener = tokio::net::TcpListener::bind("127.0.0.1:0") - .await - .expect("bind tcp"); + let listener = TokioNet::listen("127.0.0.1:0").await.expect("bind tcp"); let addr = listener.local_addr().expect("local addr"); let dispatch: Arc = @@ -48,7 +47,7 @@ async fn aimx_roundtrips_over_tcp_loopback() { acks_subscribe: false, }; tokio::spawn(serve( - TcpListener::new(listener), + framed_listener(listener), Arc::new(AimxCodec), dispatch, session_config, @@ -59,7 +58,7 @@ async fn aimx_roundtrips_over_tcp_loopback() { ..ClientConfig::default() }; let (handle, engine) = run_client( - TcpDialer::new(addr.to_string()), + framed_dialer(TokioNet::tcp(), addr.ip().to_string(), addr.port()), AimxCodec, client_config, Arc::new(TokioAdapter), From ea8fa658dfe10e1e12830a1da9a40c60119f9625 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Alexander=20Schn=C3=B6rch?= Date: Sun, 6 Sep 2026 08:31:11 +0000 Subject: [PATCH 3/5] Refactor TCP connector to support platform-agnostic implementation - Updated `Cargo.toml` and `Cargo.lock` to remove unnecessary dependencies and streamline the project. - Modified `endpoint.rs` to utilize the new `framed_dialer_at` function for TCP connections. - Enhanced `connector.rs` with a new `split_host_port` function to handle host:port parsing and added a `framed_dialer_at` function for cleaner dialing. - Removed the `embassy_transport.rs` and `tokio_transport.rs` files as they are superseded by the new connector implementation. - Updated `lib.rs` to reflect the changes in module structure and removed deprecated transport modules. - Adjusted example in `tcp_demo.rs` to align with the new dialing approach. --- Cargo.lock | 2 - aimdb-client/Cargo.toml | 3 +- aimdb-client/src/endpoint.rs | 7 +- aimdb-tcp-connector/Cargo.toml | 17 +- aimdb-tcp-connector/examples/tcp_demo.rs | 13 +- aimdb-tcp-connector/src/connector.rs | 86 +++ aimdb-tcp-connector/src/embassy_transport.rs | 591 ------------------- aimdb-tcp-connector/src/lib.rs | 30 +- aimdb-tcp-connector/src/tokio_transport.rs | 302 ---------- 9 files changed, 106 insertions(+), 945 deletions(-) delete mode 100644 aimdb-tcp-connector/src/embassy_transport.rs delete mode 100644 aimdb-tcp-connector/src/tokio_transport.rs diff --git a/Cargo.lock b/Cargo.lock index 1251e840..4437ede8 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -382,11 +382,9 @@ dependencies = [ "aimdb-tokio-adapter", "critical-section", "defmt 1.1.1", - "embassy-futures 0.1.2 (registry+https://github.com/rust-lang/crates.io-index)", "embassy-net", "embassy-net-driver-channel", "embassy-time-driver", - "embedded-io-async 0.7.0", "futures", "heapless 0.9.3", "serde", diff --git a/aimdb-client/Cargo.toml b/aimdb-client/Cargo.toml index 41d4d61b..1dc61f25 100644 --- a/aimdb-client/Cargo.toml +++ b/aimdb-client/Cargo.toml @@ -19,7 +19,8 @@ observability = ["aimdb-core/observability"] # compiled in is rejected at resolve time. transport-uds = ["dep:aimdb-uds-connector"] transport-serial = ["dep:aimdb-serial-connector"] -transport-tcp = ["dep:aimdb-tcp-connector"] +# The TCP dialer comes from the adapter's `net` transports. +transport-tcp = ["dep:aimdb-tcp-connector", "aimdb-tokio-adapter/net"] [dependencies] # Core dependencies - protocol types from aimdb-core. `connector-session` diff --git a/aimdb-client/src/endpoint.rs b/aimdb-client/src/endpoint.rs index 73a485b5..2e81c459 100644 --- a/aimdb-client/src/endpoint.rs +++ b/aimdb-client/src/endpoint.rs @@ -144,7 +144,12 @@ pub fn dial(endpoint: &str) -> ClientResult> { Scheme::Tcp => { #[cfg(feature = "transport-tcp")] { - Ok(Box::new(aimdb_tcp_connector::TcpDialer::new(parsed.target))) + // The adapter owns the socket; the connector owns the + // host/port grammar. + Ok(Box::new(aimdb_tcp_connector::framed_dialer_at( + aimdb_tokio_adapter::net::TokioNet::tcp(), + &parsed.target, + ))) } #[cfg(not(feature = "transport-tcp"))] { diff --git a/aimdb-tcp-connector/Cargo.toml b/aimdb-tcp-connector/Cargo.toml index c6dedd5c..aee43a1b 100644 --- a/aimdb-tcp-connector/Cargo.toml +++ b/aimdb-tcp-connector/Cargo.toml @@ -24,7 +24,6 @@ tokio-runtime = [ "std", "aimdb-core/connector-session", "aimdb-core/remote", - "dep:tokio", ] embassy-runtime = [ @@ -33,10 +32,7 @@ embassy-runtime = [ "aimdb-core/remote", "dep:aimdb-embassy-adapter", "aimdb-embassy-adapter/connectors", - "aimdb-embassy-adapter/embassy-net-support", - "dep:embassy-net", - "dep:embassy-futures", - "dep:embedded-io-async", + "aimdb-embassy-adapter/net", ] tracing = ["aimdb-core/tracing"] @@ -52,9 +48,7 @@ _test-tokio = ["tokio-runtime", "dep:aimdb-tokio-adapter", "aimdb-tokio-adapter/ # the `_test-tokio` build too). Run with `--features _test-embassy-loopback`. _test-embassy-loopback = [ "embassy-runtime", - # The adapter's neutral transports, exercised by `tests/accept_pool.rs` - # over the same two real stacks as `embassy_loopback.rs`. - "aimdb-embassy-adapter/net", + "dep:embassy-net", "embassy-net/medium-ip", "embassy-net/proto-ipv4", "dep:embassy-net-driver-channel", @@ -64,12 +58,11 @@ _test-embassy-loopback = [ [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 } + +# Test-only: the loopback harness stands up two real stacks (see +# `_test-embassy-loopback`). The crate itself names no socket type. 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 } diff --git a/aimdb-tcp-connector/examples/tcp_demo.rs b/aimdb-tcp-connector/examples/tcp_demo.rs index a3d6af7e..306cbd97 100644 --- a/aimdb-tcp-connector/examples/tcp_demo.rs +++ b/aimdb-tcp-connector/examples/tcp_demo.rs @@ -25,7 +25,7 @@ use aimdb_core::remote::{AimxConfig, SecurityPolicy}; use aimdb_core::session::aimx::AimxCodec; use aimdb_core::session::{run_client, ClientConfig, Payload}; use aimdb_core::AimDbBuilder; -use aimdb_tcp_connector::connector::{framed_dialer, TcpServer}; +use aimdb_tcp_connector::connector::{framed_dialer_at, TcpServer}; use aimdb_tokio_adapter::net::TokioNet; use aimdb_tokio_adapter::{TokioAdapter, TokioRecordRegistrarExt}; use serde::{Deserialize, Serialize}; @@ -151,9 +151,8 @@ async fn run_set_mode(endpoint: String, level: u64) { } fn connect(endpoint: String) -> aimdb_core::session::ClientHandle { - let (host, port) = split_endpoint(&endpoint); let (handle, engine) = run_client( - framed_dialer(TokioNet::tcp(), host, port), + framed_dialer_at(TokioNet::tcp(), &endpoint), AimxCodec, ClientConfig { sends_hello: false, @@ -164,11 +163,3 @@ fn connect(endpoint: String) -> aimdb_core::session::ClientHandle { tokio::spawn(engine); handle } - -/// Split `host:port`, defaulting to the AimX TCP port when none is given. -fn split_endpoint(endpoint: &str) -> (String, u16) { - match endpoint.rsplit_once(':') { - Some((host, port)) => (host.to_string(), port.parse().unwrap_or(7001)), - None => (endpoint.to_string(), 7001), - } -} diff --git a/aimdb-tcp-connector/src/connector.rs b/aimdb-tcp-connector/src/connector.rs index 9717556b..177ee71c 100644 --- a/aimdb-tcp-connector/src/connector.rs +++ b/aimdb-tcp-connector/src/connector.rs @@ -36,6 +36,45 @@ pub type TcpFramingDialer = FramingDialer LengthFramer, READ_CHUNK /// The listener half, framed. pub type TcpFramingListener = FramingListener LengthFramer, READ_CHUNK, WRITE_CHUNK>; +/// Port used when an endpoint names only a host. +pub const DEFAULT_PORT: u16 = 7001; + +/// Split a `host:port` endpoint, falling back to `default_port` when no port is +/// given or it does not parse. +/// +/// A bracketed IPv6 literal carries colons of its own, so only a colon *after* +/// the closing bracket separates the port; the brackets are stripped, because +/// that is the form both adapters resolve. +pub fn split_host_port(endpoint: &str, default_port: u16) -> (String, u16) { + if let Some(rest) = endpoint.strip_prefix('[') { + return match rest.split_once(']') { + Some((host, tail)) => { + let port = tail + .strip_prefix(':') + .and_then(|p| p.parse().ok()) + .unwrap_or(default_port); + (host.to_string(), port) + } + None => (rest.to_string(), default_port), + }; + } + match endpoint.rsplit_once(':') { + Some((host, port)) if !host.is_empty() => { + (host.to_string(), port.parse().unwrap_or(default_port)) + } + _ => (endpoint.to_string(), default_port), + } +} + +/// Frame an adapter's dialer for a `host:port` endpoint. +/// +/// The split-then-dial sugar every caller wants; use [`framed_dialer`] directly +/// when host and port are already separate. +pub fn framed_dialer_at(dialer: D, endpoint: &str) -> TcpFramingDialer { + let (host, port) = split_host_port(endpoint, DEFAULT_PORT); + framed_dialer(dialer, host, port) +} + /// Frame an adapter's dialer for `host:port` with length-prefix framing. pub fn framed_dialer( dialer: D, @@ -192,3 +231,50 @@ where &self.scheme } } + +#[cfg(test)] +mod tests { + use super::{split_host_port, DEFAULT_PORT}; + + #[test] + fn splits_host_and_port() { + assert_eq!( + split_host_port("127.0.0.1:7002", DEFAULT_PORT), + ("127.0.0.1".into(), 7002) + ); + } + + #[test] + fn a_bare_host_takes_the_default_port() { + assert_eq!( + split_host_port("example.test", DEFAULT_PORT), + ("example.test".into(), DEFAULT_PORT) + ); + } + + #[test] + fn an_unparsable_port_takes_the_default() { + assert_eq!( + split_host_port("host:not-a-port", DEFAULT_PORT), + ("host".into(), DEFAULT_PORT) + ); + } + + /// A bracketed IPv6 literal is full of colons; only the one after `]` + /// separates the port, and the brackets are not part of the address. + #[test] + fn brackets_are_stripped_from_an_ipv6_literal() { + assert_eq!( + split_host_port("[::1]:7003", DEFAULT_PORT), + ("::1".into(), 7003) + ); + } + + #[test] + fn a_bracketed_ipv6_host_without_a_port_is_not_mangled() { + assert_eq!( + split_host_port("[::1]", DEFAULT_PORT), + ("::1".into(), DEFAULT_PORT) + ); + } +} diff --git a/aimdb-tcp-connector/src/embassy_transport.rs b/aimdb-tcp-connector/src/embassy_transport.rs deleted file mode 100644 index 62c945ec..00000000 --- a/aimdb-tcp-connector/src/embassy_transport.rs +++ /dev/null @@ -1,591 +0,0 @@ -//! Embassy TCP transport (feature `embassy-runtime`). -//! -//! `embassy-net` has no central `TcpListener`; each `TcpSocket` must enter -//! `accept()` itself. This module therefore models Embassy TCP servers as an -//! explicit pool of caller-buffered sockets, with one accept/session worker per -//! active slot. -//! `connector-io` cannot be reused directly because `TcpSocket::split()` only -//! yields borrowed halves, while AimDB's `Connection` must own the socket. - -use alloc::boxed::Box; -use alloc::string::{String, ToString}; -use alloc::sync::Arc; -use alloc::vec::Vec; -use core::cell::RefCell; -use core::future::{poll_fn, Future}; -use core::pin::Pin; -use core::task::{Context, Poll, Waker}; - -use aimdb_core::connector::ConnectorBuilder; -use aimdb_core::remote::{AimxConfig, SecurityPolicy}; -use aimdb_core::session::aimx::AimxCodec; -use aimdb_core::session::{ - run_session, BoxFut, ClientConfig, Connection, Dialer, Dispatch, Listener, PeerInfo, - SessionConfig, SessionLimits, TransportError, TransportResult, -}; -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; - -use aimdb_core::{AimDb, DbResult}; - -use crate::framing::{encode_frame, FrameAccumulator, HEADER_LEN}; -use crate::DEFAULT_SCHEME; - -type BoxFuture = Pin + Send + 'static>>; -type BuildFuture<'a> = Pin>> + Send + 'a>>; - -const READ_CHUNK: usize = 256; - -/// A framed AimX connection over one `embassy-net` TCP socket. -pub struct TcpConnection { - socket: Option>, - recycler: Option>, - acc: FrameAccumulator, - peer: PeerInfo, -} - -// SAFETY: single-core cooperative Embassy executor; same invariant as -// `aimdb-embassy-adapter::connectors`. -unsafe impl Send for TcpConnection {} - -impl TcpConnection { - /// Wrap an already-connected TCP socket. - pub fn new(socket: TcpSocket<'static>) -> Self { - Self { - socket: Some(socket), - recycler: None, - acc: FrameAccumulator::new(), - peer: PeerInfo::default(), - } - } - - fn reusable(socket: TcpSocket<'static>, recycler: Arc) -> Self { - Self { - socket: Some(socket), - recycler: Some(recycler), - acc: FrameAccumulator::new(), - peer: PeerInfo::default(), - } - } -} - -impl Connection for TcpConnection { - fn recv(&mut self) -> BoxFut<'_, TransportResult>>> { - Box::pin(SendFutureWrapper(async move { - let socket = self.socket.as_mut().ok_or(TransportError::Closed)?; - loop { - match self.acc.next_frame() { - Some(Ok(frame)) => return Ok(Some(frame)), - Some(Err(_)) => return Err(TransportError::Io), - None => {} - } - - let mut chunk = [0u8; READ_CHUNK]; - match socket.read(&mut chunk).await { - Ok(0) => return Ok(None), - Ok(n) => self.acc.push_bytes(&chunk[..n]), - Err(_) => return Err(TransportError::Io), - } - } - })) - } - - fn send<'a>(&'a mut self, frame: &'a [u8]) -> BoxFut<'a, TransportResult<()>> { - Box::pin(SendFutureWrapper(async move { - let socket = self.socket.as_mut().ok_or(TransportError::Closed)?; - let capacity = HEADER_LEN - .checked_add(frame.len()) - .ok_or(TransportError::Io)?; - let mut out = Vec::with_capacity(capacity); - encode_frame(frame, &mut out).map_err(|_| TransportError::Io)?; - write_all(socket, &out).await?; - socket.flush().await.map_err(|_| TransportError::Closed) - })) - } - - fn peer(&self) -> &PeerInfo { - &self.peer - } -} - -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); - } - } - } -} - -async fn write_all(writer: &mut W, mut bytes: &[u8]) -> TransportResult<()> -where - W: Write, -{ - while !bytes.is_empty() { - let n = writer - .write(bytes) - .await - .map_err(|_| TransportError::Closed)?; - if n == 0 { - return Err(TransportError::Closed); - } - bytes = &bytes[n..]; - } - Ok(()) -} - -struct TcpSocketSlot { - socket: RefCell>>, - // 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` 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>, -} - -// 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. 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 { - fn new(socket: TcpSocket<'static>) -> Self { - Self { - socket: RefCell::new(Some(socket)), - waker: RefCell::new(None), - } - } - - fn take(&self) -> Option> { - self.socket.borrow_mut().take() - } - - fn poll_take(&self, cx: &mut Context<'_>) -> Poll> { - let mut slot = self.socket.borrow_mut(); - if let Some(socket) = slot.take() { - Poll::Ready(socket) - } else { - drop(slot); - *self.waker.borrow_mut() = Some(cx.waker().clone()); - Poll::Pending - } - } - - fn put(&self, socket: TcpSocket<'static>) { - let mut slot = self.socket.borrow_mut(); - debug_assert!(slot.is_none(), "Embassy TCP socket returned twice"); - if slot.is_none() { - *slot = Some(socket); - } - if let Some(waker) = self.waker.borrow_mut().take() { - waker.wake(); - } - } -} - -/// 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, - socket: Option>, -} - -impl<'a> SlotReturn<'a> { - fn new(slot: &'a Arc, 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 -/// after `abort()`/`close()`, so this dialer can redial with the same static -/// RX/TX buffers. -pub struct TcpDialer { - endpoint: IpEndpoint, - socket: Arc, -} - -impl TcpDialer { - /// Build a reusable dialer with caller-owned socket buffers. - pub fn new( - stack: Stack<'static>, - endpoint: IpEndpoint, - rx_buffer: &'static mut [u8], - tx_buffer: &'static mut [u8], - ) -> Self { - let socket = TcpSocket::new(stack, rx_buffer, tx_buffer); - Self { - endpoint, - socket: Arc::new(TcpSocketSlot::new(socket)), - } - } -} - -impl Dialer for TcpDialer { - fn connect(&self) -> BoxFut<'_, TransportResult>> { - Box::pin(SendFutureWrapper(async move { - let Some(mut socket) = self.socket.take() else { - return Err(TransportError::Io); - }; - socket.abort(); - match socket.connect(self.endpoint).await { - Ok(()) => Ok( - Box::new(TcpConnection::reusable(socket, self.socket.clone())) - as Box, - ), - Err(_) => { - socket.abort(); - self.socket.put(socket); - Err(TransportError::Io) - } - } - })) - } -} - -/// An Embassy TCP listener backed by `N` caller-owned sockets. -/// -/// `embassy-net` requires each `TcpSocket` to enter `accept()` itself, so true -/// concurrent listening means keeping multiple sockets in accept state at once. -/// `TcpServer::::with_buffers(...)` drives this listener with one worker per -/// enabled slot. The `Listener` trait implementation is only for the `N = 1` -/// compatibility path. -pub struct TcpListener { - local_endpoint: IpListenEndpoint, - slots: [Arc; N], -} - -impl TcpListener<1> { - /// Build a single-socket listener with caller-owned socket buffers. - pub fn new( - stack: Stack<'static>, - local_endpoint: impl Into, - rx_buffer: &'static mut [u8], - tx_buffer: &'static mut [u8], - ) -> Self { - Self::with_buffers(stack, local_endpoint, [rx_buffer], [tx_buffer]) - } -} - -impl TcpListener { - /// Build an N-socket listener pool with caller-owned socket buffers. - /// - /// Each `(rx_buffers[i], tx_buffers[i])` pair backs one `TcpSocket` and one - /// concurrent accept/session worker. Keep `N` aligned with the - /// `embassy-net::StackResources` capacity and the RAM budget for the - /// chosen RX/TX buffer sizes. - pub fn with_buffers( - stack: Stack<'static>, - local_endpoint: impl Into, - rx_buffers: [&'static mut [u8]; N], - tx_buffers: [&'static mut [u8]; N], - ) -> Self { - let mut rx_buffers = rx_buffers.into_iter(); - let mut tx_buffers = tx_buffers.into_iter(); - let slots = core::array::from_fn(|_| { - let rx = rx_buffers - .next() - .expect("array iterator yields exactly N RX buffers"); - let tx = tx_buffers - .next() - .expect("array iterator yields exactly N TX buffers"); - Arc::new(TcpSocketSlot::new(TcpSocket::new(stack, rx, tx))) - }); - Self { - local_endpoint: local_endpoint.into(), - slots, - } - } - - /// 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`, 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>> + 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, - dispatch: Arc, - config: SessionConfig, - max_workers: usize, - ) -> Vec { - let worker_count = max_workers.min(N); - let mut futures = Vec::with_capacity(worker_count); - for slot in self.slots.into_iter().take(worker_count) { - futures.push(Box::pin(SendFutureWrapper(serve_socket_slot( - slot, - self.local_endpoint, - codec.clone(), - dispatch.clone(), - config.clone(), - ))) as BoxFuture); - } - futures - } -} - -impl Listener for TcpListener<1> { - fn accept(&mut self) -> BoxFut<'_, TransportResult>> { - // `&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 { - 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, - local_endpoint: IpListenEndpoint, -) -> TransportResult> { - 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) - } - 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, - local_endpoint: IpListenEndpoint, - codec: Arc, - dispatch: Arc, - config: SessionConfig, -) { - loop { - // `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; - } - } -} - -/// Constructs an Embassy session client over TCP. -pub struct TcpClient; - -impl TcpClient { - /// Mirror records to/from an AimX peer over TCP. - #[allow(clippy::new_ret_no_self)] - pub fn new( - stack: Stack<'static>, - endpoint: IpEndpoint, - rx_buffer: &'static mut [u8], - tx_buffer: &'static mut [u8], - ) -> EmbassySessionClient { - EmbassySessionClient::new( - TcpDialer::new(stack, endpoint, rx_buffer, tx_buffer), - AimxCodec, - ) - .scheme(DEFAULT_SCHEME) - .with_config(ClientConfig::default()) - } -} - -/// Accepts AimX connections over an explicit Embassy TCP socket pool. -/// -/// `TcpServer::new(...)` is the one-socket convenience constructor. Use -/// `TcpServer::::with_buffers(...)` to keep `N` sockets concurrently -/// listening, each backed by caller-owned static RX/TX buffers. -pub struct TcpServer { - listener: OneShotCell>, - config: AimxConfig, - scheme: String, -} - -impl TcpServer<1> { - /// Serve AimX on one Embassy TCP socket. - pub fn new( - stack: Stack<'static>, - local_endpoint: impl Into, - rx_buffer: &'static mut [u8], - tx_buffer: &'static mut [u8], - ) -> Self { - Self { - listener: OneShotCell::new(TcpListener::new( - stack, - local_endpoint, - rx_buffer, - tx_buffer, - )), - config: AimxConfig::uds_default(), - scheme: DEFAULT_SCHEME.to_string(), - } - } -} - -impl TcpServer { - /// Serve AimX over an N-socket Embassy TCP listener pool. - /// - /// The server starts up to `min(N, max_connections)` workers. Each worker - /// keeps one `TcpSocket` in `accept()` while idle, so the network stack has - /// multiple pending listeners instead of rejecting inbound SYNs after a - /// single socket is consumed. - pub fn with_buffers( - stack: Stack<'static>, - local_endpoint: impl Into, - rx_buffers: [&'static mut [u8]; N], - tx_buffers: [&'static mut [u8]; N], - ) -> Self { - Self { - listener: OneShotCell::new(TcpListener::with_buffers( - stack, - local_endpoint, - rx_buffers, - tx_buffers, - )), - config: AimxConfig::uds_default(), - scheme: DEFAULT_SCHEME.to_string(), - } - } - - /// Use a prepared [`AimxConfig`] for limits and security policy. - pub fn with_config(mut self, config: AimxConfig) -> Self { - self.config = config; - self - } - - /// Set the security policy. - pub fn security_policy(mut self, policy: SecurityPolicy) -> Self { - self.config = self.config.security_policy(policy); - self - } - - /// Maximum concurrently served connections. - /// - /// Effective concurrency is `min(N, max)`, because every active worker owns - /// exactly one listening or connected socket. - pub fn max_connections(mut self, max: usize) -> Self { - self.config = self.config.max_connections(max); - self - } - - /// Maximum live subscriptions per connection. - pub fn max_subs_per_connection(mut self, max: usize) -> Self { - self.config = self.config.max_subs_per_connection(max); - self - } - - /// Override the scheme this connector registers. - pub fn scheme(mut self, scheme: impl Into) -> Self { - self.scheme = scheme.into(); - self - } -} - -impl ConnectorBuilder for TcpServer { - fn build<'a>(&'a self, db: &'a AimDb) -> BuildFuture<'a> { - let listener = self.listener.take_required(); - let config = self.config.clone(); - Box::pin(SendFutureWrapper(async move { - let listener = listener?; - crate::apply_writable(db, &config); - let session_config = SessionConfig { - limits: SessionLimits { - max_connections: config.max_connections, - max_subs_per_connection: config.max_subs_per_connection, - }, - reads_hello: false, - acks_subscribe: false, - }; - let dispatch: Arc = Arc::new( - aimdb_core::session::aimx::AimxDispatch::new(Arc::new(db.clone()), config), - ); - let max_workers = session_config.limits.max_connections; - Ok(listener.into_server_futures( - Arc::new(AimxCodec), - dispatch, - session_config, - max_workers, - )) - })) - } - - fn scheme(&self) -> &str { - &self.scheme - } -} diff --git a/aimdb-tcp-connector/src/lib.rs b/aimdb-tcp-connector/src/lib.rs index d144cb3d..ce28090c 100644 --- a/aimdb-tcp-connector/src/lib.rs +++ b/aimdb-tcp-connector/src/lib.rs @@ -15,18 +15,10 @@ extern crate alloc; pub mod framing; -// Runtime-neutral `TcpClient`/`TcpServer` over an adapter's stream transports. +// `TcpClient`/`TcpServer` over an adapter's stream transports. #[cfg(any(feature = "tokio-runtime", feature = "embassy-runtime"))] pub mod connector; -// Superseded by `connector` over the adapters' stream transports; both are -// deleted once the tests and examples move across. -#[cfg(feature = "tokio-runtime")] -pub mod tokio_transport; - -#[cfg(feature = "embassy-runtime")] -pub mod embassy_transport; - /// Default connector scheme. /// /// Record links such as `link_to("tcp://record")` route through this scheme. @@ -45,20 +37,8 @@ pub(crate) fn apply_writable(db: &aimdb_core::AimDb, config: &aimdb_core::remote } } -#[cfg(all(feature = "tokio-runtime", not(feature = "embassy-runtime")))] -pub use tokio_transport::{TcpClient, TcpConnection, TcpDialer, TcpListener, TcpServer}; - -#[cfg(all(feature = "tokio-runtime", feature = "embassy-runtime"))] -pub use embassy_transport::{ - TcpClient as EmbassyTcpClient, TcpConnection as EmbassyTcpConnection, - TcpDialer as EmbassyTcpDialer, TcpListener as EmbassyTcpListener, - TcpServer as EmbassyTcpServer, -}; -#[cfg(all(feature = "tokio-runtime", feature = "embassy-runtime"))] -pub use tokio_transport::{ - TcpClient as TokioTcpClient, TcpConnection as TokioTcpConnection, TcpDialer, TcpListener, - TcpServer as TokioTcpServer, +#[cfg(any(feature = "tokio-runtime", feature = "embassy-runtime"))] +pub use connector::{ + framed_dialer, framed_dialer_at, framed_listener, split_host_port, TcpClient, TcpServer, + DEFAULT_PORT, }; - -#[cfg(all(feature = "embassy-runtime", not(feature = "tokio-runtime")))] -pub use embassy_transport::{TcpClient, TcpConnection, TcpDialer, TcpListener, TcpServer}; diff --git a/aimdb-tcp-connector/src/tokio_transport.rs b/aimdb-tcp-connector/src/tokio_transport.rs deleted file mode 100644 index 1152e2d1..00000000 --- a/aimdb-tcp-connector/src/tokio_transport.rs +++ /dev/null @@ -1,302 +0,0 @@ -//! Tokio TCP transport (feature `tokio-runtime`). - -use std::future::Future; -use std::pin::Pin; -use std::sync::Arc; - -use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt}; -use tokio::net::{TcpListener as TokioTcpListener, TcpStream}; - -use aimdb_core::connector::ConnectorBuilder; -use aimdb_core::remote::{AimxConfig, SecurityPolicy}; -use aimdb_core::session::aimx::{AimxCodec, AimxDispatch}; -use aimdb_core::session::{ - BoxFut, Connection, Dialer, Dispatch, Listener, PeerInfo, SessionClientConnector, - SessionConfig, SessionLimits, SessionServerConnector, TransportError, TransportResult, -}; -use aimdb_core::{AimDb, DbError, DbResult}; - -use crate::framing::{encode_frame, FrameAccumulator, DEFAULT_MAX_FRAME, HEADER_LEN}; -use crate::DEFAULT_SCHEME; - -type BoxFuture = Pin + Send + 'static>>; -type BuildFuture<'a> = Pin>> + Send + 'a>>; - -const READ_CHUNK: usize = 1024; - -/// A framed TCP connection. -pub struct TcpConnection { - stream: S, - acc: FrameAccumulator, - peer: PeerInfo, -} - -impl TcpConnection { - /// Wrap an already-connected async stream with default frame limits. - pub fn new(stream: S) -> Self { - Self::with_max_frame(stream, DEFAULT_MAX_FRAME) - } - - /// Wrap an already-connected async stream with a caller-provided frame cap. - pub fn with_max_frame(stream: S, max_frame: usize) -> Self { - Self { - stream, - acc: FrameAccumulator::with_max_frame(max_frame), - peer: PeerInfo::default(), - } - } - - fn with_peer(mut self, peer: PeerInfo) -> Self { - self.peer = peer; - self - } -} - -impl Connection for TcpConnection -where - S: AsyncRead + AsyncWrite + Unpin + Send, -{ - fn recv(&mut self) -> BoxFut<'_, TransportResult>>> { - Box::pin(async move { - loop { - match self.acc.next_frame() { - Some(Ok(frame)) => return Ok(Some(frame)), - Some(Err(_)) => return Err(TransportError::Io), - None => {} - } - - let mut chunk = [0u8; READ_CHUNK]; - match self.stream.read(&mut chunk).await { - Ok(0) => return Ok(None), - Ok(n) => self.acc.push_bytes(&chunk[..n]), - Err(_) => return Err(TransportError::Io), - } - } - }) - } - - fn send<'a>(&'a mut self, frame: &'a [u8]) -> BoxFut<'a, TransportResult<()>> { - Box::pin(async move { - let capacity = HEADER_LEN - .checked_add(frame.len()) - .ok_or(TransportError::Io)?; - let mut out = Vec::with_capacity(capacity); - encode_frame(frame, &mut out).map_err(|_| TransportError::Io)?; - self.stream - .write_all(&out) - .await - .map_err(|_| TransportError::Closed)?; - self.stream - .flush() - .await - .map_err(|_| TransportError::Closed) - }) - } - - fn peer(&self) -> &PeerInfo { - &self.peer - } -} - -/// The initiating side: dials a TCP endpoint on each connect. -#[derive(Clone)] -pub struct TcpDialer { - endpoint: String, - max_frame: usize, -} - -impl TcpDialer { - /// Dial `endpoint`, for example `127.0.0.1:7001`. - pub fn new(endpoint: impl Into) -> Self { - Self { - endpoint: endpoint.into(), - max_frame: DEFAULT_MAX_FRAME, - } - } - - /// Set maximum frame payload size. - pub fn max_frame(mut self, max_frame: usize) -> Self { - self.max_frame = max_frame; - self - } -} - -impl Dialer for TcpDialer { - fn connect(&self) -> BoxFut<'_, TransportResult>> { - Box::pin(async move { - let stream = TcpStream::connect(&self.endpoint) - .await - .map_err(|_| TransportError::Io)?; - let peer_addr = stream.peer_addr().ok().map(|a| a.to_string()); - let mut peer = PeerInfo::default(); - peer.peer_addr = peer_addr; - Ok( - Box::new(TcpConnection::with_max_frame(stream, self.max_frame).with_peer(peer)) - as Box, - ) - }) - } -} - -/// The accepting side. -pub struct TcpListener { - inner: TokioTcpListener, - max_frame: usize, -} - -impl TcpListener { - /// Wrap an already-bound listener. - pub fn new(inner: TokioTcpListener) -> Self { - Self { - inner, - max_frame: DEFAULT_MAX_FRAME, - } - } - - /// Set maximum frame payload size. - pub fn max_frame(mut self, max_frame: usize) -> Self { - self.max_frame = max_frame; - self - } -} - -impl Listener for TcpListener { - fn accept(&mut self) -> BoxFut<'_, TransportResult>> { - Box::pin(async move { - let (stream, addr) = self.inner.accept().await.map_err(|_| TransportError::Io)?; - let mut peer = PeerInfo::default(); - peer.peer_addr = Some(addr.to_string()); - Ok( - Box::new(TcpConnection::with_max_frame(stream, self.max_frame).with_peer(peer)) - as Box, - ) - }) - } -} - -/// Constructs a TCP session client connector. -pub struct TcpClient; - -impl TcpClient { - /// Mirror records to/from an AimX peer over TCP. - #[allow(clippy::new_ret_no_self)] - pub fn new(endpoint: impl Into) -> SessionClientConnector { - SessionClientConnector::new(TcpDialer::new(endpoint), AimxCodec).scheme(DEFAULT_SCHEME) - } -} - -/// Accepts AimX connections over TCP. -pub struct TcpServer { - bind_addr: String, - config: AimxConfig, - scheme: String, - max_frame: usize, -} - -impl TcpServer { - /// Serve AimX on `bind_addr`. - /// - /// Prefer loopback addresses such as `127.0.0.1:7001` unless the deployment - /// provides its own network-layer protection. - pub fn new(bind_addr: impl Into) -> Self { - Self { - bind_addr: bind_addr.into(), - config: AimxConfig::uds_default(), - scheme: DEFAULT_SCHEME.to_string(), - max_frame: DEFAULT_MAX_FRAME, - } - } - - /// Use a prepared [`AimxConfig`] for limits and security policy. - pub fn with_config(mut self, config: AimxConfig) -> Self { - self.config = config; - self - } - - /// Set the security policy. - pub fn security_policy(mut self, policy: SecurityPolicy) -> Self { - self.config = self.config.security_policy(policy); - self - } - - /// Maximum concurrently served connections. - pub fn max_connections(mut self, max: usize) -> Self { - self.config = self.config.max_connections(max); - self - } - - /// Maximum live subscriptions per connection. - pub fn max_subs_per_connection(mut self, max: usize) -> Self { - self.config = self.config.max_subs_per_connection(max); - self - } - - /// Maximum TCP frame payload size. - pub fn max_frame(mut self, max_frame: usize) -> Self { - self.max_frame = max_frame; - self - } - - /// Override the scheme this connector registers. - pub fn scheme(mut self, scheme: impl Into) -> Self { - self.scheme = scheme.into(); - self - } -} - -impl ConnectorBuilder for TcpServer { - fn build<'a>(&'a self, db: &'a AimDb) -> BuildFuture<'a> { - let bind_addr = self.bind_addr.clone(); - let config = self.config.clone(); - let scheme = self.scheme.clone(); - let max_frame = self.max_frame; - Box::pin(async move { - let session_config = SessionConfig { - limits: SessionLimits { - max_connections: config.max_connections, - max_subs_per_connection: config.max_subs_per_connection, - }, - reads_hello: false, - acks_subscribe: false, - }; - let bind_config = bind_addr.clone(); - let dispatch_config = config; - let connector = SessionServerConnector::new( - move || bind_tcp_listener(&bind_config, max_frame), - AimxCodec, - move |db: &AimDb| -> Arc { - crate::apply_writable(db, &dispatch_config); - Arc::new(AimxDispatch::new( - Arc::new(db.clone()), - dispatch_config.clone(), - )) - }, - session_config, - ) - .scheme(scheme); - connector.build(db).await - }) - } - - fn scheme(&self) -> &str { - &self.scheme - } -} - -fn bind_tcp_listener(addr: &str, max_frame: usize) -> DbResult { - let listener = std::net::TcpListener::bind(addr).map_err(|e| DbError::IoWithContext { - context: "Failed to bind TCP listener".to_string(), - source: e, - })?; - listener - .set_nonblocking(true) - .map_err(|e| DbError::IoWithContext { - context: "Failed to set TCP listener nonblocking".to_string(), - source: e, - })?; - let listener = TokioTcpListener::from_std(listener).map_err(|e| DbError::IoWithContext { - context: "Failed to create Tokio TCP listener".to_string(), - source: e, - })?; - Ok(TcpListener::new(listener).max_frame(max_frame)) -} From c0987150d74b509589d697ace824c00d152e96c2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Alexander=20Schn=C3=B6rch?= Date: Sun, 6 Sep 2026 14:55:16 +0000 Subject: [PATCH 4/5] docs(tcp-connector): record the runtime-neutral migration Co-Authored-By: Claude Opus 5 --- aimdb-tcp-connector/CHANGELOG.md | 14 ++++++++++++++ 1 file changed, 14 insertions(+) diff --git a/aimdb-tcp-connector/CHANGELOG.md b/aimdb-tcp-connector/CHANGELOG.md index 9f0de737..3f2cd3fd 100644 --- a/aimdb-tcp-connector/CHANGELOG.md +++ b/aimdb-tcp-connector/CHANGELOG.md @@ -7,8 +7,22 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Changed + +- **One path for both runtimes (breaking).** `TcpServer::new` takes an + already-bound listener from an adapter (`TokioNet::listen`, + `EmbassyNet::listen::`) instead of a bind string, and `TcpClient::new` + takes a dialer. `tokio_transport` and `embassy_transport` are deleted with the + whole `Tokio*`/`Embassy*` alias set, and with them the crate's last three + `unsafe impl`s. The library no longer depends on `tokio`, `embassy-net`, + `embassy-futures` or `embedded-io-async`. + ### Added +- **`connector` — runtime-neutral `TcpClient`/`TcpServer`** over core's + `StreamDialer`/`StreamListener`, plus `framing::LengthFramer` against core's + `Framer`. `split_host_port` and `framed_dialer_at` carry the `host:port` + grammar, including bracketed IPv6 literals. - **`tests/accept_pool.rs`** — the adapter's pooled `StreamListener` over two crossover-wired `embassy-net` stacks, with a rebuild-and-cancel pool as the negative control: it loses a SYN arriving between accepts, the stored-accept From 2324c2d7b9683861f80d412ae51764de5a69a3b9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Alexander=20Schn=C3=B6rch?= Date: Tue, 8 Sep 2026 19:08:35 +0000 Subject: [PATCH 5/5] feat(framing): introduce FrameFault enum for improved error handling in framers --- aimdb-core/CHANGELOG.md | 7 +- aimdb-core/src/session/io.rs | 100 ++++++++++++++++++++++---- aimdb-core/src/session/mod.rs | 6 +- aimdb-embassy-adapter/src/net.rs | 7 +- aimdb-serial-connector/src/framing.rs | 14 ++-- aimdb-tcp-connector/CHANGELOG.md | 8 ++- aimdb-tcp-connector/src/framing.rs | 22 +++--- aimdb-tcp-connector/tests/framing.rs | 53 ++++++++++++++ aimdb-tokio-adapter/src/net.rs | 9 ++- 9 files changed, 188 insertions(+), 38 deletions(-) diff --git a/aimdb-core/CHANGELOG.md b/aimdb-core/CHANGELOG.md index 1fa27c53..2bec298a 100644 --- a/aimdb-core/CHANGELOG.md +++ b/aimdb-core/CHANGELOG.md @@ -14,7 +14,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 sit below `Connection`, so an adapter owns sockets and clocks while a connector owns framing. `FramedConnection` plus `FramingDialer`/`FramingListener` lift a byte stream into the existing `Dialer`/`Listener`, and `OneShot` is a - `Send + Sync` cell for moved-in resources with no `unsafe`. + `Send + Sync` cell for moved-in resources with no `unsafe`. `Framer` reports a + failure as a `FrameFault`: `Recoverable` (skip the run and resync, what a + self-delimiting format such as COBS can do) or `Fatal` (close, what a length + prefix must do, having no delimiter to resync on). `encode` returns + `Result<(), FrameFault>` for the same reason, so a frame the framer refuses + reaches the caller as `TransportError::Framing` rather than a silent `Ok`. - **A panic is a bug, not an error channel — checked.** The crate is compiled under `deny(clippy::unwrap_used, clippy::expect_used, clippy::panic)` outside its own tests. Four sites fixed: poisoned-mutex recovery in the diff --git a/aimdb-core/src/session/io.rs b/aimdb-core/src/session/io.rs index e8caf7e4..161e8cb1 100644 --- a/aimdb-core/src/session/io.rs +++ b/aimdb-core/src/session/io.rs @@ -156,15 +156,31 @@ pub trait Delay { // Framing — a transport crate contributes one of these and inherits the rest. // =========================================================================== +/// How badly a framing step failed. +/// +/// A self-delimiting format resyncs on its next delimiter; a length prefix has +/// none, so nothing tells payload bytes from the next header. Only the framer +/// knows which case it is in, so it says, rather than the connection guessing. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum FrameFault { + /// Bad frame, good link: skip it and keep reading. + Recoverable, + /// The link can no longer be interpreted and must close. + Fatal, +} + /// Frames a byte stream: COBS, length-prefix, NDJSON. pub trait Framer { /// Encode one logical frame, appending its wire bytes to `out`. - fn encode(&self, frame: &[u8], out: &mut Vec); + /// + /// On `Err` nothing is appended, so a rejected frame never reaches the wire + /// half-written. + fn encode(&self, frame: &[u8], out: &mut Vec) -> Result<(), FrameFault>; /// Feed received bytes into the accumulator. fn push_bytes(&mut self, bytes: &[u8]); - /// Pull the next complete frame: `Some(Ok(frame))`, `Some(Err(()))` for a - /// malformed/unsynced run (skipped, the stream resyncs), or `None`. - fn next_frame(&mut self) -> Option, ()>>; + /// Pull the next complete frame: `Some(Ok(frame))`, `Some(Err(fault))` (see + /// [`FrameFault`]), or `None` when more bytes are needed. + fn next_frame(&mut self) -> Option, FrameFault>>; } /// Builds a fresh [`Framer`] per connection. @@ -231,11 +247,17 @@ where fn recv(&mut self) -> BoxFut<'_, TransportResult>>> { Box::pin(async move { loop { - // A run that fails to decode is line noise or a mid-stream - // join, not fatal: skip it and resync on the next frame. match self.framer.next_frame() { Some(Ok(frame)) => return Ok(Some(frame)), - Some(Err(())) => continue, + // Line noise or a mid-stream join: skip it and resync on + // the next frame. + Some(Err(FrameFault::Recoverable)) => continue, + // No boundary left to resync on: reading on would reinterpret + // payload bytes as headers for the life of the connection. + Some(Err(FrameFault::Fatal)) => { + log_warn!("framed recv: unrecoverable framing error, closing connection"); + return Err(TransportError::Framing); + } None => {} } let mut chunk = [0u8; RC]; @@ -251,7 +273,14 @@ where fn send<'a>(&'a mut self, frame: &'a [u8]) -> BoxFut<'a, TransportResult<()>> { Box::pin(async move { let mut out = Vec::new(); - self.framer.encode(frame, &mut out); + if let Err(_fault) = self.framer.encode(frame, &mut out) { + log_warn!( + "framed send: framer rejected a {}-byte frame ({:?}), closing connection", + frame.len(), + _fault + ); + return Err(TransportError::Framing); + } for chunk in out.chunks(WC) { self.stream.write_all(chunk).await?; } @@ -450,25 +479,35 @@ mod tests { // --- Test doubles ----------------------------------------------------- /// Length-prefixed framer: one length byte, then that many payload bytes. - /// A `0xFF` length marks a corrupt run, so resync has something to skip. + /// A `0xFF` length marks a corrupt run, so resync has something to skip; + /// `0xFE` marks an unrecoverable one. A frame too long for the one-byte + /// length is rejected by `encode`. #[derive(Default)] struct LenFramer { buf: Vec, } impl Framer for LenFramer { - fn encode(&self, frame: &[u8], out: &mut Vec) { + fn encode(&self, frame: &[u8], out: &mut Vec) -> Result<(), FrameFault> { + if frame.len() >= 0xFE { + return Err(FrameFault::Recoverable); + } out.push(frame.len() as u8); out.extend_from_slice(frame); + Ok(()) } fn push_bytes(&mut self, bytes: &[u8]) { self.buf.extend_from_slice(bytes); } - fn next_frame(&mut self) -> Option, ()>> { + fn next_frame(&mut self) -> Option, FrameFault>> { let len = *self.buf.first()? as usize; if len == 0xFF { self.buf.remove(0); - return Some(Err(())); + return Some(Err(FrameFault::Recoverable)); + } + if len == 0xFE { + self.buf.clear(); + return Some(Err(FrameFault::Fatal)); } if self.buf.len() < len + 1 { return None; @@ -609,6 +648,37 @@ mod tests { ); } + #[tokio::test] + async fn recv_closes_on_an_unrecoverable_framing_error() { + // The bytes after the bad header are unreadable, so the connection ends + // rather than reinterpreting them as the next header forever. + let mut conn = framed(MockStream::with_reads(vec![vec![0xFE, 2, b'o', b'k']])); + assert_eq!( + conn.recv().await, + Err(TransportError::Framing), + "a fatal fault ends the connection, it is not skipped" + ); + } + + #[tokio::test] + async fn send_reports_a_frame_the_framer_rejects() { + let stream = MockStream::default(); + let mut conn = framed(stream.clone()); + let oversized = vec![b'x'; 0xFE]; + + assert_eq!( + conn.send(&oversized).await, + Err(TransportError::Framing), + "a dropped frame is an error, never a silent Ok" + ); + let st = stream.0.lock(); + assert!( + st.written.is_empty(), + "nothing half-encoded reaches the wire" + ); + assert_eq!(st.flushes, 0); + } + #[tokio::test] async fn recv_propagates_the_streams_own_error() { let mut conn = FramedConnection::<_, _, 256, 256>::new(FailingStream, LenFramer::default()); @@ -689,9 +759,11 @@ mod tests { fn framer_factory_is_implemented_for_closures() { struct Noop; impl Framer for Noop { - fn encode(&self, _frame: &[u8], _out: &mut Vec) {} + fn encode(&self, _frame: &[u8], _out: &mut Vec) -> Result<(), FrameFault> { + Ok(()) + } fn push_bytes(&mut self, _bytes: &[u8]) {} - fn next_frame(&mut self) -> Option, ()>> { + fn next_frame(&mut self) -> Option, FrameFault>> { None } } diff --git a/aimdb-core/src/session/mod.rs b/aimdb-core/src/session/mod.rs index ab4915d2..2a88a364 100644 --- a/aimdb-core/src/session/mod.rs +++ b/aimdb-core/src/session/mod.rs @@ -45,8 +45,8 @@ pub use client::{pump_client, run_client, ClientConfig, ClientHandle}; pub use connector::{SessionClientConnector, SessionServerConnector}; #[cfg(feature = "connector-session")] pub use io::{ - ByteStream, Datagram, DatagramBinder, Delay, FramedConnection, Framer, FramerFactory, - FramingDialer, FramingListener, IoError, OneShot, StreamDialer, StreamListener, + ByteStream, Datagram, DatagramBinder, Delay, FrameFault, FramedConnection, Framer, + FramerFactory, FramingDialer, FramingListener, IoError, OneShot, StreamDialer, StreamListener, }; #[cfg(feature = "connector-session")] pub use pump::{pump_sink, pump_source}; @@ -231,6 +231,8 @@ pub enum TransportError { Closed, /// An underlying I/O operation failed. Io, + /// The byte stream could not be framed, and the framer cannot resynchronize. + Framing, } /// Envelope-codec failure — a frame could not be decoded/encoded. diff --git a/aimdb-embassy-adapter/src/net.rs b/aimdb-embassy-adapter/src/net.rs index f839d206..cac8d0ec 100644 --- a/aimdb-embassy-adapter/src/net.rs +++ b/aimdb-embassy-adapter/src/net.rs @@ -635,7 +635,7 @@ where #[cfg(test)] mod tests { use super::*; - use aimdb_core::session::{Connection, FramedConnection, Framer}; + use aimdb_core::session::{Connection, FrameFault, FramedConnection, Framer}; use alloc::vec; use alloc::vec::Vec; @@ -690,14 +690,15 @@ mod tests { } impl Framer for LenFramer { - fn encode(&self, frame: &[u8], out: &mut Vec) { + fn encode(&self, frame: &[u8], out: &mut Vec) -> Result<(), FrameFault> { out.push(frame.len() as u8); out.extend_from_slice(frame); + Ok(()) } fn push_bytes(&mut self, bytes: &[u8]) { self.buf.extend_from_slice(bytes); } - fn next_frame(&mut self) -> Option, ()>> { + fn next_frame(&mut self) -> Option, FrameFault>> { let len = *self.buf.first()? as usize; if self.buf.len() < len + 1 { return None; diff --git a/aimdb-serial-connector/src/framing.rs b/aimdb-serial-connector/src/framing.rs index 7a76ab22..4f1b682e 100644 --- a/aimdb-serial-connector/src/framing.rs +++ b/aimdb-serial-connector/src/framing.rs @@ -19,6 +19,7 @@ //! own. That half needs `aimdb_core::session`, so it is gated on the runtime //! features that enable core's `connector-session`. +use aimdb_core::session::FrameFault; use alloc::vec::Vec; /// A frame could not be recovered — line noise, a truncated frame, a mid-stream @@ -184,18 +185,21 @@ impl CobsFramer { #[cfg(any(feature = "tokio-runtime", feature = "embassy-runtime"))] impl aimdb_core::session::Framer for CobsFramer { - fn encode(&self, frame: &[u8], out: &mut Vec) { + fn encode(&self, frame: &[u8], out: &mut Vec) -> Result<(), FrameFault> { encode_frame(frame, out); + Ok(()) } fn push_bytes(&mut self, bytes: &[u8]) { self.acc.push_bytes(bytes); } - fn next_frame(&mut self) -> Option, ()>> { - // `FrameError` collapses to `()`: the connection only distinguishes - // "got a frame" from "skip and resync". - self.acc.next_frame().map(|r| r.map_err(|_| ())) + fn next_frame(&mut self) -> Option, FrameFault>> { + // COBS delimits frames, and the accumulator already resyncs on the next + // sentinel, so a dropped run never invalidates the rest of the stream. + self.acc + .next_frame() + .map(|r| r.map_err(|_| FrameFault::Recoverable)) } } diff --git a/aimdb-tcp-connector/CHANGELOG.md b/aimdb-tcp-connector/CHANGELOG.md index 3f2cd3fd..778e7872 100644 --- a/aimdb-tcp-connector/CHANGELOG.md +++ b/aimdb-tcp-connector/CHANGELOG.md @@ -21,8 +21,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - **`connector` — runtime-neutral `TcpClient`/`TcpServer`** over core's `StreamDialer`/`StreamListener`, plus `framing::LengthFramer` against core's - `Framer`. `split_host_port` and `framed_dialer_at` carry the `host:port` - grammar, including bracketed IPv6 literals. + `Framer`. A length prefix has no delimiter to resync on, so `LengthFramer` + reports a bad header as `FrameFault::Fatal` and the connection closes instead + of reading on; an oversized outbound frame is still dropped whole rather than + written half-encoded, but is now reported rather than silently discarded. + `split_host_port` and `framed_dialer_at` carry the `host:port` grammar, + including bracketed IPv6 literals. - **`tests/accept_pool.rs`** — the adapter's pooled `StreamListener` over two crossover-wired `embassy-net` stacks, with a rebuild-and-cancel pool as the negative control: it loses a SYN arriving between accepts, the stored-accept diff --git a/aimdb-tcp-connector/src/framing.rs b/aimdb-tcp-connector/src/framing.rs index d02fd5cd..c1346404 100644 --- a/aimdb-tcp-connector/src/framing.rs +++ b/aimdb-tcp-connector/src/framing.rs @@ -10,6 +10,7 @@ //! The declared length is payload bytes only. Oversized frames are fatal because //! length-prefix TCP has no delimiter that would let the receiver safely resync. +use aimdb_core::session::FrameFault; use alloc::vec::Vec; /// Number of bytes in the fixed frame header. @@ -130,21 +131,26 @@ impl Default for LengthFramer { #[cfg(any(feature = "tokio-runtime", feature = "embassy-runtime"))] impl aimdb_core::session::Framer for LengthFramer { - fn encode(&self, frame: &[u8], out: &mut Vec) { - // Core's `Framer::encode` is infallible, so an oversized frame is - // dropped here rather than written half-encoded: the peer would read a - // length prefix with no payload behind it and desync permanently. + fn encode(&self, frame: &[u8], out: &mut Vec) -> Result<(), FrameFault> { + // An oversized frame is dropped whole rather than written half-encoded: + // the peer would read a length prefix with no payload behind it and + // desync permanently. The link itself is untouched, so the fault is + // recoverable and the caller decides what to do with the connection. if frame.len() > self.max_frame { - return; + return Err(FrameFault::Recoverable); } - let _ = encode_frame(frame, out); + encode_frame(frame, out).map_err(|_| FrameFault::Recoverable) } fn push_bytes(&mut self, bytes: &[u8]) { self.acc.push_bytes(bytes); } - fn next_frame(&mut self) -> Option, ()>> { - self.acc.next_frame().map(|r| r.map_err(|_| ())) + fn next_frame(&mut self) -> Option, FrameFault>> { + // A length prefix has no delimiter to resync on, so a bad header is + // fatal: nothing downstream tells payload bytes from the next header. + self.acc + .next_frame() + .map(|r| r.map_err(|_| FrameFault::Fatal)) } } diff --git a/aimdb-tcp-connector/tests/framing.rs b/aimdb-tcp-connector/tests/framing.rs index 9f7ffa59..7b203017 100644 --- a/aimdb-tcp-connector/tests/framing.rs +++ b/aimdb-tcp-connector/tests/framing.rs @@ -82,3 +82,56 @@ fn empty_payload_roundtrips() { acc.push_bytes(&wire); assert_eq!(acc.next_frame().unwrap().unwrap(), b""); } + +// --- LengthFramer against core's `Framer` contract ------------------------- +// +// The accumulator tests above cover the wire format; these cover what the +// connection is told about a failure, which is what decides whether a desynced +// link closes or silently keeps reading. + +#[cfg(any(feature = "tokio-runtime", feature = "embassy-runtime"))] +mod framer { + use aimdb_core::session::{FrameFault, Framer}; + use aimdb_tcp_connector::framing::LengthFramer; + + #[test] + fn a_frame_within_the_cap_roundtrips() { + let mut framer = LengthFramer::new(); + let mut wire = Vec::new(); + framer.encode(b"hello", &mut wire).expect("encode"); + + framer.push_bytes(&wire); + assert_eq!(framer.next_frame(), Some(Ok(b"hello".to_vec()))); + assert_eq!(framer.next_frame(), None, "nothing left buffered"); + } + + #[test] + fn an_oversized_frame_is_rejected_and_nothing_is_written() { + let framer = LengthFramer::with_max_frame(4); + let mut wire = Vec::new(); + + assert_eq!( + framer.encode(b"too long", &mut wire), + Err(FrameFault::Recoverable), + "the caller is told, rather than the frame vanishing behind an Ok" + ); + assert!( + wire.is_empty(), + "a length prefix with no payload would desync the peer permanently" + ); + } + + #[test] + fn a_bad_length_prefix_is_fatal() { + let mut framer = LengthFramer::with_max_frame(4); + // A header claiming more than the cap: there is no delimiter to resync + // on, so the rest of the stream cannot be interpreted. + framer.push_bytes(&5u32.to_be_bytes()); + + assert_eq!( + framer.next_frame(), + Some(Err(FrameFault::Fatal)), + "reported fatal, so the connection closes instead of resyncing" + ); + } +} diff --git a/aimdb-tokio-adapter/src/net.rs b/aimdb-tokio-adapter/src/net.rs index 1b370e48..8c9dbce0 100644 --- a/aimdb-tokio-adapter/src/net.rs +++ b/aimdb-tokio-adapter/src/net.rs @@ -181,7 +181,9 @@ impl Delay for TokioDelay { #[cfg(test)] mod tests { use super::*; - use aimdb_core::session::{Dialer, Framer, FramingDialer, FramingListener, Listener}; + use aimdb_core::session::{ + Dialer, FrameFault, Framer, FramingDialer, FramingListener, Listener, + }; use std::net::Ipv4Addr; /// Length-prefixed framer, enough to drive a `FramedConnection`. @@ -191,14 +193,15 @@ mod tests { } impl Framer for LenFramer { - fn encode(&self, frame: &[u8], out: &mut Vec) { + fn encode(&self, frame: &[u8], out: &mut Vec) -> Result<(), FrameFault> { out.push(frame.len() as u8); out.extend_from_slice(frame); + Ok(()) } fn push_bytes(&mut self, bytes: &[u8]) { self.buf.extend_from_slice(bytes); } - fn next_frame(&mut self) -> Option, ()>> { + fn next_frame(&mut self) -> Option, FrameFault>> { let len = *self.buf.first()? as usize; if self.buf.len() < len + 1 { return None;