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/CHANGELOG.md b/aimdb-client/CHANGELOG.md index 588af779..123fb25f 100644 --- a/aimdb-client/CHANGELOG.md +++ b/aimdb-client/CHANGELOG.md @@ -7,6 +7,17 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Changed + +- **`tcp://` endpoint validation shares one grammar with the connector.** + `require_tcp_target` had grown a 56-line private copy of the `host:port` + grammar; it now calls `aimdb_core::session::split_host_port_opt` and keeps + only its own policy on top — a URL must name its port, where a connector + constructor may default one. Rejection messages are unchanged and now pinned + by a test, including the two that are easy to confuse: an unbracketed IPv6 + literal is told to add brackets, a bracketed one missing its port is told to + add a port. + ### Changed (breaking) — Design 047 - **One subscription API: `AimxConnection::subscribe`.** It now yields 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..b12e3dbc 100644 --- a/aimdb-client/src/endpoint.rs +++ b/aimdb-client/src/endpoint.rs @@ -144,7 +144,14 @@ 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. + let dialer = aimdb_tcp_connector::framed_dialer_at( + aimdb_tokio_adapter::net::TokioNet::tcp(), + &parsed.target, + ) + .map_err(|e| ClientError::unsupported_endpoint(endpoint, e.to_string()))?; + Ok(Box::new(dialer)) } #[cfg(not(feature = "transport-tcp"))] { @@ -167,60 +174,38 @@ fn require_nonempty(endpoint: &str, target: &str) -> ClientResult<()> { } /// Validate `tcp://host:port`. +/// +/// The grammar itself is [`split_host_port_opt`] — one implementation, shared +/// with the connector, reachable here even when `transport-tcp` is off. What +/// this adds is the client's own policy: a URL must name its port, where a +/// connector constructor may default one. fn require_tcp_target(endpoint: &str, target: &str) -> ClientResult<()> { require_nonempty(endpoint, target)?; - let (host, port) = if let Some(rest) = target.strip_prefix('[') { - let Some((host, after_host)) = rest.split_once(']') else { - return Err(ClientError::unsupported_endpoint( - endpoint, - "missing closing bracket for IPv6 TCP host", - )); - }; - let Some(port) = after_host.strip_prefix(':') else { - return Err(ClientError::unsupported_endpoint( - endpoint, - "missing TCP port", - )); - }; - (host, port) - } else { - if target.matches(':').count() > 1 { - return Err(ClientError::unsupported_endpoint( - endpoint, - "IPv6 TCP hosts must be bracketed, e.g. tcp://[::1]:7001", - )); - } - let Some((host, port)) = target.split_once(':') else { - return Err(ClientError::unsupported_endpoint( - endpoint, - "missing TCP port", - )); - }; - if host.contains(['[', ']']) { - return Err(ClientError::unsupported_endpoint( - endpoint, - "malformed TCP host", - )); - } - (host, port) - }; + let (host, port) = aimdb_core::session::split_host_port_opt(target) + .map_err(|e| ClientError::unsupported_endpoint(endpoint, e.to_string()))?; - if host.is_empty() { + // Brackets are stripped from a well-formed literal, so any left over came + // from the middle of a host: `tcp://foo[bar:7001`. + if host.contains(['[', ']']) { return Err(ClientError::unsupported_endpoint( endpoint, - "missing TCP host", + "malformed TCP host", )); } - if port.is_empty() { - return Err(ClientError::unsupported_endpoint( - endpoint, - "missing TCP port", - )); + + if port.is_none() { + // An unbracketed IPv6 literal cannot carry a port, so "add one" is the + // wrong advice — say what actually has to change. Keyed on the original + // target, since a bracketed host arrives here already stripped. + let reason = if !target.starts_with('[') && host.contains(':') { + "IPv6 TCP hosts must be bracketed, e.g. tcp://[::1]:7001" + } else { + "missing TCP port" + }; + return Err(ClientError::unsupported_endpoint(endpoint, reason)); } - port.parse::().map_err(|_| { - ClientError::unsupported_endpoint(endpoint, format!("invalid TCP port {port:?}")) - })?; + Ok(()) } @@ -308,6 +293,25 @@ mod tests { assert_eq!(p.target, "[fe80::1]:7001"); } + /// Each rejection says what has to change. Asserting the reasons, not just + /// `is_err`, is what stops the grammar move from silently degrading them — + /// an unbracketed IPv6 literal needs brackets, not a port appended. + #[test] + fn a_rejected_tcp_endpoint_says_what_is_wrong() { + let reason = |ep: &str| match parse_endpoint(ep) { + Err(ClientError::UnsupportedEndpoint { reason, .. }) => reason, + other => panic!("{ep} should be rejected as an endpoint, got {other:?}"), + }; + + assert!(reason("tcp://host").contains("missing TCP port")); + assert!(reason("tcp://[fe80::1]").contains("missing TCP port")); + assert!(reason("tcp://fe80::1").contains("must be bracketed")); + assert!(reason("tcp://:1234").contains("no host")); + assert!(reason("tcp://[]:7001").contains("no host")); + assert!(reason("tcp://host:fast").contains("not a number")); + assert!(reason("tcp://[fe80::1:7001").contains("closing bracket")); + } + #[test] fn malformed_endpoints_are_rejected() { // Malformed TCP. diff --git a/aimdb-client/src/engine.rs b/aimdb-client/src/engine.rs index 87535dec..ed3cf29f 100644 --- a/aimdb-client/src/engine.rs +++ b/aimdb-client/src/engine.rs @@ -1,6 +1,7 @@ //! Engine-based AimX client. //! -//! The client rides the shared session engine: a [`UdsDialer`] + the symmetric +//! The client rides the shared session engine: a `UdsDialer` (unlinked — it +//! exists only behind `transport-uds`) + the symmetric //! [`AimxCodec`] drive [`run_client`], which owns the wire, the request-id //! demux, and (optionally) reconnect. The public surface is the cheap-clone //! [`ClientHandle`] plus typed convenience wrappers and per-subscription diff --git a/aimdb-core/CHANGELOG.md b/aimdb-core/CHANGELOG.md index 1fa27c53..1589e0c6 100644 --- a/aimdb-core/CHANGELOG.md +++ b/aimdb-core/CHANGELOG.md @@ -14,7 +14,21 @@ 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`. + `TransportError` also gains `Busy`, for a transport whose one endpoint + resource is already in use — distinguishable from `Io`, and retried like it. +- **`session::endpoint` — the `host:port` grammar, in one place.** + `split_host_port_opt` / `split_host_port` and `EndpointError`, moved up from + `aimdb-tcp-connector` because more than one crate needs them and they do not + all depend on each other: `aimdb-client` resolves `tcp://` URLs whether or not + that connector is compiled in, so it had grown a 56-line copy of the same + grammar. `_opt` reports whether a port was written at all, so one caller can + default it while another rejects an endpoint that omits it. - **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/client.rs b/aimdb-core/src/session/client.rs index 54c59aca..ff16b0d8 100644 --- a/aimdb-core/src/session/client.rs +++ b/aimdb-core/src/session/client.rs @@ -501,7 +501,7 @@ async fn client_loop( // `Closed` is terminal — the dialer signals it will never succeed // again (e.g. the caller stopped the bridge), so retrying would // spin a permanently-failing redial forever. Only transient - // failures (`Io`) earn a backoff+retry. Other transports' dialers + // failures (`Io`, `Busy`) earn a backoff+retry. Other transports' dialers // map connect failures to `Io`, never `Closed`, so this is safe. if e == TransportError::Closed { return; diff --git a/aimdb-core/src/session/endpoint.rs b/aimdb-core/src/session/endpoint.rs new file mode 100644 index 00000000..fe985ce3 --- /dev/null +++ b/aimdb-core/src/session/endpoint.rs @@ -0,0 +1,161 @@ +//! `host:port` endpoint grammar, shared by every transport that speaks it. +//! +//! It lives here rather than in a connector because more than one crate needs +//! it and they do not all depend on each other: `aimdb-client` resolves +//! `scheme://` URLs whether or not the matching connector is compiled in. +//! One grammar, one implementation, one set of tests. +//! +//! Callers layer their own policy on top: [`split_host_port_opt`] reports +//! whether a port was written at all, so one caller can default it while +//! another rejects an endpoint that omits it. + +use alloc::string::{String, ToString}; + +/// Why an endpoint is not a `host:port`. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum EndpointError { + /// The endpoint names no host. + EmptyHost, + /// A bracketed IPv6 literal with no closing `]`. + UnclosedBracket, + /// A port was written, and is not one. + BadPort, +} + +impl core::fmt::Display for EndpointError { + fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result { + f.write_str(match self { + Self::EmptyHost => "endpoint names no host", + Self::UnclosedBracket => "missing closing bracket for IPv6 host", + Self::BadPort => "port is not a number in 0..=65535", + }) + } +} + +/// Split a `host:port` endpoint, leaving the port `None` when none was written. +/// +/// 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 the adapters resolve. An **unbracketed** IPv6 literal +/// cannot carry a port at all — brackets are what add one — so every colon in +/// it belongs to the address. +/// +/// A port that is written but is not one is an error, never a silent fallback: +/// dialing a different service is worse than not dialing. +pub fn split_host_port_opt(endpoint: &str) -> Result<(String, Option), EndpointError> { + if let Some(rest) = endpoint.strip_prefix('[') { + let Some((host, tail)) = rest.split_once(']') else { + return Err(EndpointError::UnclosedBracket); + }; + if host.is_empty() { + return Err(EndpointError::EmptyHost); + } + let port = if tail.is_empty() { + None + } else { + // Anything but `:port` after `]` is junk, including junk with a port + // behind it — `[::1]oops:7003` names no reachable service. + Some( + tail.strip_prefix(':') + .ok_or(EndpointError::BadPort)? + .parse() + .map_err(|_| EndpointError::BadPort)?, + ) + }; + return Ok((host.to_string(), port)); + } + match endpoint.rsplit_once(':') { + // Another colon before the last one: an unbracketed IPv6 literal, whose + // trailing group is part of the address rather than a port. + Some((head, _)) if head.contains(':') => Ok((endpoint.to_string(), None)), + Some(("", _)) => Err(EndpointError::EmptyHost), + Some((host, port)) => Ok(( + host.to_string(), + Some(port.parse().map_err(|_| EndpointError::BadPort)?), + )), + None if endpoint.is_empty() => Err(EndpointError::EmptyHost), + None => Ok((endpoint.to_string(), None)), + } +} + +/// As [`split_host_port_opt`], substituting `default_port` when the endpoint +/// names only a host. +pub fn split_host_port(endpoint: &str, default_port: u16) -> Result<(String, u16), EndpointError> { + let (host, port) = split_host_port_opt(endpoint)?; + Ok((host, port.unwrap_or(default_port))) +} + +#[cfg(test)] +mod tests { + use super::*; + + const DEFAULT: u16 = 7001; + + fn split(endpoint: &str) -> Result<(String, u16), EndpointError> { + split_host_port(endpoint, DEFAULT) + } + + #[test] + fn splits_host_and_port() { + assert_eq!(split("127.0.0.1:7002"), Ok(("127.0.0.1".into(), 7002))); + } + + #[test] + fn a_bare_host_takes_the_default_port() { + assert_eq!(split("example.test"), Ok(("example.test".into(), DEFAULT))); + assert_eq!( + split_host_port_opt("example.test"), + Ok(("example.test".into(), None)), + "the caller can tell a defaulted port from a written one" + ); + } + + /// A written port that is not one names no service, so it is an error + /// rather than a silent fallback to the default — that would dial a + /// different, possibly live, server. + #[test] + fn an_unparsable_port_is_rejected() { + assert_eq!(split("host:not-a-port"), Err(EndpointError::BadPort)); + assert_eq!(split("10.0.0.5:8080x"), Err(EndpointError::BadPort)); + assert_eq!(split("host:99999"), Err(EndpointError::BadPort)); + assert_eq!(split("host:"), Err(EndpointError::BadPort)); + } + + /// 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("[::1]:7003"), Ok(("::1".into(), 7003))); + } + + #[test] + fn a_bracketed_ipv6_host_without_a_port_is_not_mangled() { + assert_eq!(split("[::1]"), Ok(("::1".into(), DEFAULT))); + } + + /// Brackets are what let an IPv6 literal carry a port, so without them + /// every colon belongs to the address and no port was written. + #[test] + fn an_unbracketed_ipv6_literal_keeps_all_its_colons() { + assert_eq!(split("::1"), Ok(("::1".into(), DEFAULT))); + assert_eq!(split("fe80::1"), Ok(("fe80::1".into(), DEFAULT))); + assert_eq!( + split("2001:db8::dead:beef"), + Ok(("2001:db8::dead:beef".into(), DEFAULT)), + "the trailing group is address, not a port" + ); + } + + #[test] + fn a_malformed_endpoint_is_rejected() { + assert_eq!(split(":99999"), Err(EndpointError::EmptyHost)); + assert_eq!(split(""), Err(EndpointError::EmptyHost)); + assert_eq!(split("[]:7001"), Err(EndpointError::EmptyHost)); + assert_eq!(split("[::1"), Err(EndpointError::UnclosedBracket)); + assert_eq!( + split("[::1]oops:7003"), + Err(EndpointError::BadPort), + "an explicit port behind junk is not silently honoured" + ); + } +} diff --git a/aimdb-core/src/session/io.rs b/aimdb-core/src/session/io.rs index 6b3d99bb..999c8595 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?; } @@ -266,6 +295,9 @@ where /// Lifts a [`StreamDialer`] and a [`FramerFactory`] into a [`Dialer`], so /// `run_client` drives an adapter transport unchanged. +/// +/// `Clone` because `SessionClientConnector` clones its dialer per build. +#[derive(Clone)] pub struct FramingDialer { dialer: D, framers: FF, @@ -437,25 +469,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; @@ -596,6 +638,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()); @@ -676,9 +749,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..85c5f170 100644 --- a/aimdb-core/src/session/mod.rs +++ b/aimdb-core/src/session/mod.rs @@ -24,6 +24,8 @@ mod client; #[cfg(feature = "connector-session")] mod connector; #[cfg(feature = "connector-session")] +mod endpoint; +#[cfg(feature = "connector-session")] mod io; #[cfg(feature = "connector-session")] mod pump; @@ -44,9 +46,11 @@ pub use client::{pump_client, run_client, ClientConfig, ClientHandle}; #[cfg(feature = "connector-session")] pub use connector::{SessionClientConnector, SessionServerConnector}; #[cfg(feature = "connector-session")] +pub use endpoint::{split_host_port, split_host_port_opt, EndpointError}; +#[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 +235,11 @@ pub enum TransportError { Closed, /// An underlying I/O operation failed. Io, + /// The byte stream could not be framed, and the framer cannot resynchronize. + Framing, + /// The transport's one endpoint resource is already in use — a second dial + /// on a single-socket transport while the first connection is live. + Busy, } /// Envelope-codec failure — a frame could not be decoded/encoded. diff --git a/aimdb-data-contracts/src/lib.rs b/aimdb-data-contracts/src/lib.rs index 48f17ddd..0139b474 100644 --- a/aimdb-data-contracts/src/lib.rs +++ b/aimdb-data-contracts/src/lib.rs @@ -179,7 +179,8 @@ pub trait Settable: SchemaType { /// Project a schema type onto a numeric domain signal. /// /// The trait's kernel is the numeric projection: implement it, call -/// [`ObservableRegistrarExt::observe`], +/// `ObservableRegistrarExt::observe` (unlinked — it exists only behind the +/// `observable` feature), /// and the signal is folded into live last/min/max/mean statistics that surface /// on `record.list` / `record.get` and stage profiling. The signal can also feed /// threshold checks, alerting, and aggregation. diff --git a/aimdb-embassy-adapter/CHANGELOG.md b/aimdb-embassy-adapter/CHANGELOG.md index ec61f9fe..65ecc83d 100644 --- a/aimdb-embassy-adapter/CHANGELOG.md +++ b/aimdb-embassy-adapter/CHANGELOG.md @@ -23,7 +23,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 `accept()` returns it to its slot instead of leaving the slot permanently empty, and both yield before reporting a synchronously-failing attempt so a misconfigured endpoint (port-0 `InvalidPort`) warn-loops rather than spinning - the single-core executor. + the single-core executor. `EmbassyTcpDialer` is `Clone` so a framed dialer can + meet `SessionClientConnector`'s bound, but a clone **shares** the one socket + rather than duplicating it: a clone dialing while another handle holds the + link gets `TransportError::Busy`, not `Io`, so the client engine's dial-failed + warning names the real cause instead of looking like a dead peer. A second + concurrent connection needs a second `EmbassyNet::tcp` with its own buffers. - **`RuntimeOps` implemented for `EmbassyAdapter` (Issue #130, design 034 Phase 2).** The dyn-safe capability surface from `aimdb-executor`, gated on `embassy-time` like `TimeOps`: `now_nanos()` is boot-anchored uptime at microsecond granularity (the portable lower bound), `sleep` boxes `embassy_time::Timer::after`, `unix_time` rides the `set_unix_time` anchor, `log` forwards to the defmt-backed `Logger`. Covered by the shared contract test on the host (the test time driver now wakes immediately on `schedule_wake`, so already-expired timers complete; non-zero sleeps remain unusable on the pinned-at-0 host clock). - **M17 — centralized Embassy connector spines: the one audited home for the single-core `unsafe` ([Design 033](../docs/design/033-M17-unify-connectors-drop-send.md)).** New `connectors` module (features `connectors` / `connector-io`) collecting the force-`Send` plumbing every Embassy connector used to hand-roll, so a connector crate carries **no `unsafe` and no `SendFutureWrapper`**: - **Session spine** — `EmbassySessionClient` / `EmbassySessionServer` (the Embassy duals of core's `SessionClientConnector` / `SessionServerConnector`), the one-shot `OneShotDialer` / `OneShotListener` over a moved-in peripheral connection (the listener parks forever after the first accept — point-to-point), and the force-`Send + Sync` `OneShotCell` for builders holding a moved-in value. `EmbassySessionClient::new` defaults to `reconnect: false` (unlike `ClientConfig::default`): a one-shot dialer can't redial, so the engine would otherwise spin on `TransportError::Io` forever; a re-dialable transport opts back in via `with_config`. diff --git a/aimdb-embassy-adapter/src/net.rs b/aimdb-embassy-adapter/src/net.rs index cac48efe..ad8d3a1b 100644 --- a/aimdb-embassy-adapter/src/net.rs +++ b/aimdb-embassy-adapter/src/net.rs @@ -209,6 +209,14 @@ impl ByteStream for EmbassyTcpStream { } /// Dials TCP connections over one caller-owned socket. +/// +/// `Clone` shares that socket rather than duplicating it — it exists so a +/// framed dialer can satisfy `SessionClientConnector`'s `Clone` bound. A clone +/// dialing while another handle holds the connection gets +/// [`TransportError::Busy`]. For a second *concurrent* connection call +/// [`EmbassyNet::tcp`] again with its own buffers, which is the only way to get +/// a second socket. +#[derive(Clone)] pub struct EmbassyTcpDialer { slot: Arc, } @@ -228,7 +236,7 @@ impl StreamDialer for EmbassyTcpDialer { let endpoint = IpEndpoint::new(addr.into(), port); let Some(socket) = self.slot.take() else { - return Err(TransportError::Io); + return Err(TransportError::Busy); }; // The guard owns the socket for the whole dial: on success it is // defused and the socket moves into the stream, on failure *or @@ -634,7 +642,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; @@ -689,14 +697,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..d28ddeff 100644 --- a/aimdb-serial-connector/src/framing.rs +++ b/aimdb-serial-connector/src/framing.rs @@ -11,14 +11,19 @@ //! the round-trip is unit-tested on the host without any transport. //! //! Two layers live here. [`encode_frame`] and [`FrameAccumulator`] are the COBS -//! codec itself, with no dependency on the session substrate. [`CobsFramer`] -//! below is that codec behind core's `Framer` trait, plus the +//! codec itself, with no dependency on the session substrate. `CobsFramer` +//! below — unlinked, as it exists only behind a runtime feature — is that codec +//! behind core's `Framer` trait, plus the //! `FramedConnection` aliases it forms with each adapter's byte source — the //! whole of what this crate contributes to a session, since the byte sources //! come from the adapters and this crate names no socket or UART type of its //! own. That half needs `aimdb_core::session`, so it is gated on the runtime //! features that enable core's `connector-session`. +// Gated with the items that use it: the accumulator below is `alloc`-only and +// builds without core's session layer. +#[cfg(any(feature = "tokio-runtime", feature = "embassy-runtime"))] +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 +189,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-serial-connector/src/lib.rs b/aimdb-serial-connector/src/lib.rs index 56e68aaf..4dc517b8 100644 --- a/aimdb-serial-connector/src/lib.rs +++ b/aimdb-serial-connector/src/lib.rs @@ -3,10 +3,13 @@ //! //! A thin, swappable transport crate (the serial sibling of `aimdb-uds-connector`): //! it contributes only the `Dialer`/`Listener`/`Connection` triple plus thin -//! sugar; the AimX codec ([`AimxCodec`](aimdb_core::session::aimx::AimxCodec)), -//! dispatch ([`AimxDispatch`](aimdb_core::session::aimx::AimxDispatch)), and the +//! sugar; the AimX codec (`AimxCodec`), dispatch (`AimxDispatch`), and the //! runtime-neutral session engines are reused verbatim from `aimdb-core`. //! +//! Core's session items are named unlinked throughout these docs: they exist +//! only when a runtime feature pulls in `aimdb-core/connector-session`, and a +//! link to them fails `cargo doc` on a build without one. +//! //! The wire is the same compact AimX JSON as UDS, but framed with **COBS** //! (Consistent Overhead Byte Stuffing) and a `0x00` delimiter instead of a //! newline — self-synchronizing on a lossy/unframed serial medium. See @@ -15,8 +18,7 @@ //! # Two halves //! //! - **`tokio-runtime`** (std, host/gateway): real serial via `tokio-serial`, -//! riding the generic [`SessionClientConnector`](aimdb_core::session::SessionClientConnector) -//! / [`SessionServerConnector`](aimdb_core::session::SessionServerConnector). +//! riding the generic `SessionClientConnector` / `SessionServerConnector`. //! See `tokio_transport`. //! - **`embassy-runtime`** (`no_std + alloc`, MCU): generic over //! `embedded-io-async` UART halves; the COBS `Framer` plus thin sugar over the diff --git a/aimdb-tcp-connector/CHANGELOG.md b/aimdb-tcp-connector/CHANGELOG.md index 9f0de737..1c2cde7b 100644 --- a/aimdb-tcp-connector/CHANGELOG.md +++ b/aimdb-tcp-connector/CHANGELOG.md @@ -7,8 +7,54 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Changed + +- **Features name what the code needs, not an executor.** New `connector` + feature gates everything the session layer backs; `std` becomes orthogonal, + since `src/` contains no `std::` and is `alloc`-only either way. The eleven + `any(tokio-runtime, embassy-runtime)` gates — never once naming a single + runtime, because after the migration there is no per-runtime code — collapse to + one `cfg(feature = "connector")`. `tokio-runtime` and `embassy-runtime` remain + as aliases and will be removed after a release. A third runtime (FreeRTOS with + lwIP) now enables `connector` and supplies its own transport, rather than + enabling a feature named after an executor it does not use. +- **`embassy-runtime` no longer pulls `aimdb-embassy-adapter`.** `src/` never + named it; only the loopback harness does, so it moves to + `_test-embassy-loopback` — which is where the Tokio side already had its + adapter. The Embassy *library* graph is now `aimdb-core` alone. +- **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`. 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. + The frame cap is settable again: `LengthFramers` is a `FramerFactory` holding + `max_frame`, where the `fn() -> LengthFramer` it replaces was stateless and + could only ever produce `DEFAULT_MAX_FRAME`. Reach it through + `TcpServer::max_frame(n)`, `TcpClient::bounded(..)`, `framed_dialer_bounded` + or `framed_listener_bounded`; the un-suffixed constructors keep the 64 KiB + default. It bounds what one connection can make the receiver buffer, which is + a memory limit on an MCU and a DoS limit on an exposed port. + `split_host_port` and `framed_dialer_at` carry the `host:port` grammar and + are fallible, returning `EndpointError`. The grammar itself now lives in + `aimdb_core::session::endpoint` and is re-exported here unchanged — same + paths, same behaviour — so `aimdb-client` can share it without depending on + this crate. Brackets are what let an IPv6 + literal carry a port, so an unbracketed one (`fe80::1`, `2001:db8::dead:beef`) + keeps every colon as address and takes the default port rather than having its + last group read as one. A port that is written but is not a number in + `0..=65535` is rejected instead of falling back to `DEFAULT_PORT`, which would + dial a different — possibly live — service. - **`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/Cargo.toml b/aimdb-tcp-connector/Cargo.toml index 563f8705..34ac3ad6 100644 --- a/aimdb-tcp-connector/Cargo.toml +++ b/aimdb-tcp-connector/Cargo.toml @@ -13,36 +13,32 @@ categories = ["network-programming", "embedded", "no-std"] [features] default = ["aimdb-core/alloc"] -std = [ - "aimdb-core/std", +# The connector needs core's session layer, and nothing else: `src/` names no +# adapter and no `std::`, so one gate covers every target. A runtime is chosen by +# passing an adapter's dialer/listener — a dependency the caller adds, not a +# feature here — which is what lets a third runtime work with no edit to this +# crate. +connector = [ "aimdb-core/alloc", "aimdb-core/connector-session", "aimdb-core/remote", ] -tokio-runtime = [ - "std", - "aimdb-core/connector-session", - "aimdb-core/remote", - "dep:tokio", -] +# Orthogonal to `connector`: the code is `alloc`-only either way, so this only +# lifts `no_std` and forwards core's own `std`. +std = ["connector", "aimdb-core/std"] -embassy-runtime = [ - "aimdb-core/alloc", - "aimdb-core/connector-session", - "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", -] +# Deprecated aliases. They name executors this crate never mentions, which blocks +# a FreeRTOS/lwIP build from enabling the connector without claiming to be +# Embassy. Kept so existing consumers and CI legs keep working; remove after a +# release. +tokio-runtime = ["std"] +embassy-runtime = ["connector"] 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 @@ -52,9 +48,10 @@ _test-tokio = ["tokio-runtime", "dep: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`. + "dep:aimdb-embassy-adapter", + "aimdb-embassy-adapter/connectors", "aimdb-embassy-adapter/net", + "dep:embassy-net", "embassy-net/medium-ip", "embassy-net/proto-ipv4", "dep:embassy-net-driver-channel", @@ -64,12 +61,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 c66014c0..42a556f1 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_at, 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,10 @@ async fn run_set_mode(endpoint: String, level: u64) { } fn connect(endpoint: String) -> aimdb_core::session::ClientHandle { + let dialer = framed_dialer_at(TokioNet::tcp(), &endpoint) + .unwrap_or_else(|e| panic!("invalid endpoint {endpoint:?}: {e}")); let (handle, engine) = run_client( - TcpDialer::new(endpoint), + dialer, AimxCodec, ClientConfig { sends_hello: false, diff --git a/aimdb-tcp-connector/src/connector.rs b/aimdb-tcp-connector/src/connector.rs new file mode 100644 index 00000000..dea0db87 --- /dev/null +++ b/aimdb-tcp-connector/src/connector.rs @@ -0,0 +1,248 @@ +//! 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`](crate::framing::LengthFramer), bounded by a +//! [`LengthFramers`] factory so the cap stays settable. + +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}; + +// The `host:port` grammar lives in core: `aimdb-client` resolves `tcp://` +// URLs whether or not this crate is compiled in, so both need it and +// neither can depend on the other. Re-exported here as the crate's own API. +pub use aimdb_core::session::{split_host_port, split_host_port_opt, EndpointError}; + +use crate::framing::{LengthFramers, 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; +/// The listener half, framed. +pub type TcpFramingListener = FramingListener; + +/// Port used when an endpoint names only a host. +pub const DEFAULT_PORT: u16 = 7001; + +/// 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, +) -> Result, EndpointError> { + let (host, port) = split_host_port(endpoint, DEFAULT_PORT)?; + Ok(framed_dialer(dialer, host, port)) +} + +/// Frame an adapter's dialer for `host:port` with length-prefix framing, +/// bounded by [`DEFAULT_MAX_FRAME`]. +pub fn framed_dialer( + dialer: D, + host: impl Into, + port: u16, +) -> TcpFramingDialer { + framed_dialer_bounded(dialer, host, port, DEFAULT_MAX_FRAME) +} + +/// As [`framed_dialer`], capping an inbound frame at `max_frame` payload bytes. +pub fn framed_dialer_bounded( + dialer: D, + host: impl Into, + port: u16, + max_frame: usize, +) -> TcpFramingDialer { + FramingDialer::new(dialer, LengthFramers::new(max_frame), host, port) +} + +/// Frame an adapter's listener with length-prefix framing, bounded by +/// [`DEFAULT_MAX_FRAME`]. +pub fn framed_listener(listener: L) -> TcpFramingListener { + framed_listener_bounded(listener, DEFAULT_MAX_FRAME) +} + +/// As [`framed_listener`], capping an inbound frame at `max_frame` payload bytes. +pub fn framed_listener_bounded( + listener: L, + max_frame: usize, +) -> TcpFramingListener { + FramingListener::new(listener, LengthFramers::new(max_frame)) +} + +/// 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> { + Self::bounded(dialer, host, port, DEFAULT_MAX_FRAME) + } + + /// As [`new`](Self::new), capping an inbound frame at `max_frame` payload + /// bytes rather than [`DEFAULT_MAX_FRAME`]. + pub fn bounded( + dialer: D, + host: impl Into, + port: u16, + max_frame: usize, + ) -> SessionClientConnector, AimxCodec> { + SessionClientConnector::new( + framed_dialer_bounded(dialer, host, port, max_frame), + 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, + max_frame: usize, +} + +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(), + max_frame: DEFAULT_MAX_FRAME, + } + } + + /// Cap an inbound frame at `max_frame` payload bytes. + /// + /// The default is [`DEFAULT_MAX_FRAME`]. Lower it on a memory-constrained + /// target, or on a port reachable by peers you do not control: it bounds + /// what one connection can make the receiver buffer. + pub fn max_frame(mut self, max_frame: usize) -> Self { + self.max_frame = max_frame; + self + } + + /// 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::new(framed_listener_bounded(listener, self.max_frame)); + 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 + } +} 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/framing.rs b/aimdb-tcp-connector/src/framing.rs index 97e8630c..03939dd5 100644 --- a/aimdb-tcp-connector/src/framing.rs +++ b/aimdb-tcp-connector/src/framing.rs @@ -10,6 +10,10 @@ //! 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. +// Gated with the items that use it: the accumulator below is `alloc`-only and +// builds without core's session layer. +#[cfg(feature = "connector")] +use aimdb_core::session::FrameFault; use alloc::vec::Vec; /// Number of bytes in the fixed frame header. @@ -91,3 +95,102 @@ 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(feature = "connector")] +pub struct LengthFramer { + acc: FrameAccumulator, + max_frame: usize, +} + +#[cfg(feature = "connector")] +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(feature = "connector")] +impl Default for LengthFramer { + fn default() -> Self { + Self::new() + } +} + +/// Builds one [`LengthFramer`] per connection, bounded by `max_frame`. +/// +/// A `fn() -> LengthFramer` is nameable but stateless, so it can only ever +/// produce [`DEFAULT_MAX_FRAME`]. Carrying the bound in a factory keeps the +/// framed type aliases nameable *and* lets a deployment choose the cap — on a +/// constrained target it is a memory bound, on an exposed port a limit on what +/// a peer can make the receiver buffer. +#[cfg(feature = "connector")] +#[derive(Debug, Clone, Copy)] +pub struct LengthFramers { + max_frame: usize, +} + +#[cfg(feature = "connector")] +impl LengthFramers { + /// Framers bounded by `max_frame` payload bytes. + pub fn new(max_frame: usize) -> Self { + Self { max_frame } + } +} + +#[cfg(feature = "connector")] +impl Default for LengthFramers { + fn default() -> Self { + Self::new(DEFAULT_MAX_FRAME) + } +} + +#[cfg(feature = "connector")] +impl aimdb_core::session::FramerFactory for LengthFramers { + type Framer = LengthFramer; + + fn framer(&self) -> LengthFramer { + LengthFramer::with_max_frame(self.max_frame) + } +} + +#[cfg(feature = "connector")] +impl aimdb_core::session::Framer for LengthFramer { + 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 Err(FrameFault::Recoverable); + } + 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, 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/src/lib.rs b/aimdb-tcp-connector/src/lib.rs index 82b14495..2dfc5c2d 100644 --- a/aimdb-tcp-connector/src/lib.rs +++ b/aimdb-tcp-connector/src/lib.rs @@ -1,10 +1,14 @@ //! Length-prefixed TCP transport connector for AimDB remote access. //! -//! This crate contributes only the TCP transport triple plus thin -//! [`TcpClient`]/[`TcpServer`] sugar. AimX protocol bytes still come from -//! [`AimxCodec`](aimdb_core::session::aimx::AimxCodec), and the session engines +//! This crate contributes only the length-prefix framing plus thin +//! `TcpClient`/`TcpServer` sugar; the socket comes from an adapter +//! (`TokioNet`, `EmbassyNet`) through core's `StreamDialer`/`StreamListener`. +//! AimX protocol bytes still come from `AimxCodec`, and the session engines //! still live in `aimdb-core`. //! +//! Names above are unlinked on purpose: they exist only behind the `connector` +//! feature, and a link to them fails `cargo doc` on a build without it. +//! //! TCP is a byte stream, so the transport frames every AimX envelope as //! `u32` big-endian payload length followed by the payload bytes. See //! [`framing`]. @@ -15,11 +19,9 @@ extern crate alloc; pub mod framing; -#[cfg(feature = "tokio-runtime")] -pub mod tokio_transport; - -#[cfg(feature = "embassy-runtime")] -pub mod embassy_transport; +// `TcpClient`/`TcpServer` over an adapter's stream transports. +#[cfg(feature = "connector")] +pub mod connector; /// Default connector scheme. /// @@ -28,7 +30,7 @@ pub const DEFAULT_SCHEME: &str = "tcp"; /// Mark each record named in the policy's writable set as writable, so /// `record.list` advertises the writable flag. The dispatch also enforces it. -#[cfg(any(feature = "tokio-runtime", feature = "embassy-runtime"))] +#[cfg(feature = "connector")] pub(crate) fn apply_writable(db: &aimdb_core::AimDb, config: &aimdb_core::remote::AimxConfig) { for key in config.security_policy.writable_records() { if let Some(id) = db.inner().resolve_str(&key) { @@ -39,20 +41,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(feature = "connector")] +pub use connector::{ + framed_dialer, framed_dialer_at, framed_dialer_bounded, framed_listener, + framed_listener_bounded, split_host_port, EndpointError, TcpClient, TcpServer, DEFAULT_PORT, }; -#[cfg(all(feature = "tokio-runtime", feature = "embassy-runtime"))] -pub use tokio_transport::{ - TcpClient as TokioTcpClient, TcpConnection as TokioTcpConnection, TcpDialer, TcpListener, - TcpServer as TokioTcpServer, -}; - -#[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)) -} diff --git a/aimdb-tcp-connector/tests/accept_pool.rs b/aimdb-tcp-connector/tests/accept_pool.rs index f0005fde..b5d3e74c 100644 --- a/aimdb-tcp-connector/tests/accept_pool.rs +++ b/aimdb-tcp-connector/tests/accept_pool.rs @@ -61,8 +61,11 @@ const MTU: usize = 1514; const SERVER_IP: Ipv4Address = Ipv4Address::new(192, 168, 0, 1); const CLIENT_IP: Ipv4Address = Ipv4Address::new(192, 168, 0, 2); -/// As `StreamDialer::connect` takes it: a host string the adapter resolves. -const SERVER_HOST: &str = "192.168.0.1"; +/// As `StreamDialer::connect` takes it, derived from [`SERVER_IP`] so a second +/// literal cannot drift from the address the stack is configured with. +fn server_host() -> alloc::string::String { + alloc::format!("{SERVER_IP}") +} type ChState = ch::State; @@ -256,7 +259,12 @@ fn pool_keeps_every_slot_listening_between_accepts() { // Accept #1 arms both slots, returns when A lands. let (accepted_a, mut client_a) = futures::join!( async { listener.accept().await.expect("accept A") }, - async { dialer_a.connect(SERVER_HOST, 7101).await.expect("dial A") }, + async { + dialer_a + .connect(&server_host(), 7101) + .await + .expect("dial A") + }, ); let (mut server_a, peer_a) = accepted_a; assert!( @@ -268,7 +276,7 @@ fn pool_keeps_every_slot_listening_between_accepts() { roundtrip(&mut server_a, &mut client_a, b"aaa").await; // Slot 1 must still be in LISTEN. - let mut client_b = dialer_b.connect(SERVER_HOST, 7101).await.expect( + let mut client_b = dialer_b.connect(&server_host(), 7101).await.expect( "second SYN was refused: the pool did not keep slot 1 listening between accepts", ); @@ -300,11 +308,16 @@ fn naive_pool_loses_the_syn_that_arrives_between_accepts() { let (socket_a, _client_a) = futures::join!( async { listener.accept().await.expect("accept A") }, - async { dialer_a.connect(SERVER_HOST, 7102).await.expect("dial A") }, + async { + dialer_a + .connect(&server_host(), 7102) + .await + .expect("dial A") + }, ); let _keep_a = socket_a; - let refused = dialer_b.connect(SERVER_HOST, 7102).await; + let refused = dialer_b.connect(&server_host(), 7102).await; assert_eq!( refused.err(), Some(TransportError::Io), diff --git a/aimdb-tcp-connector/tests/connector_roundtrip.rs b/aimdb-tcp-connector/tests/connector_roundtrip.rs new file mode 100644 index 00000000..893c1ddc --- /dev/null +++ b/aimdb-tcp-connector/tests/connector_roundtrip.rs @@ -0,0 +1,207 @@ +//! `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. Complements `tokio_roundtrip.rs`, which drives the same framed +//! transports straight through the session engines: that file covers the wire +//! path, this one covers the connector builders sitting on top of it. +#![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::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, 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); +} + +/// The builder's own wiring, which neither existing suite observes. +/// +/// `tokio_roundtrip.rs` round-trips AimX but hand-builds its `SessionConfig` +/// and dispatch; the tests above drive `TcpServer::build` but only check that a +/// socket accepts. Between them sits everything `build` actually *does*, and two +/// distinct pieces of it are asserted here — both verified by mutation, because +/// a test that cannot fail is worse than none: +/// +/// - **the config reaches the dispatch**: dropping it in `with_config` costs the +/// security policy and the write comes back `Denied`. +/// - **`apply_writable` runs**: it has exactly one caller and marks record +/// storage writable from that policy. It does *not* gate writes — the policy +/// does, in `ensure_writable` — so it is only visible in the metadata +/// `record.list` returns, which is what the last assertion reads. +#[tokio::test] +async fn a_policy_allowed_write_lands_through_the_built_server() { + let listener = TokioNet::listen("127.0.0.1:0").await.expect("bind"); + let addr = listener.local_addr().expect("bound addr"); + + let mut policy = SecurityPolicy::read_write(); + policy.allow_write_key("setting"); + + let mut builder = AimDbBuilder::new() + .runtime(Arc::new(TokioAdapter)) + .with_connector( + TcpServer::new(listener).with_config(AimxConfig::uds_default().security_policy(policy)), + ); + builder.configure::("setting", |reg| { + reg.buffer(BufferCfg::SingleLatest).with_remote_access(); + }); + let (db, runner) = builder.build().await.expect("build db"); + db.set_record_from_json("setting", serde_json::json!({ "level": 1 })) + .expect("seed setting"); + tokio::spawn(runner.run()); + + let (handle, engine) = run_client( + framed_dialer(TokioNet::tcp(), addr.ip().to_string(), addr.port()), + AimxCodec, + ClientConfig { + sends_hello: false, + ..ClientConfig::default() + }, + Arc::new(TokioAdapter), + ); + tokio::spawn(engine); + + let set: Payload = serde_json::to_vec(&serde_json::json!({ + "name": "setting", + "value": { "level": 7 } + })) + .unwrap() + .into(); + tokio::time::timeout(Duration::from_secs(5), handle.call("record.set", set)) + .await + .expect("record.set within timeout") + .expect("the policy marks 'setting' writable, so the write must be allowed"); + + // Read back over the wire: an ack alone would not prove the value landed. + let get: Payload = serde_json::to_vec(&serde_json::json!({ "name": "setting" })) + .unwrap() + .into(); + let reply = tokio::time::timeout(Duration::from_secs(5), handle.call("record.get", get)) + .await + .expect("record.get within timeout") + .expect("record.get ok"); + let value: serde_json::Value = serde_json::from_slice(&reply).expect("json reply"); + assert_eq!(value, serde_json::json!({ "level": 7 })); + + // `apply_writable` marks storage from the policy, and `record.list` is the + // only place that marking surfaces. Without it a client cannot tell which + // records it may write. + let list = tokio::time::timeout( + Duration::from_secs(5), + handle.call("record.list", Payload::from(&b"{}"[..])), + ) + .await + .expect("record.list within timeout") + .expect("record.list ok"); + let records: serde_json::Value = serde_json::from_slice(&list).expect("json reply"); + let setting = records + .as_array() + .expect("record.list returns an array") + .iter() + .find(|r| r["record_key"] == "setting") + .expect("'setting' is listed"); + assert_eq!( + setting["writable"], + serde_json::json!(true), + "the policy's writable marking must reach record metadata" + ); +} diff --git a/aimdb-tcp-connector/tests/embassy_loopback.rs b/aimdb-tcp-connector/tests/embassy_loopback.rs index a8986afc..0b1ac519 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 @@ -20,9 +18,10 @@ 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_core::session::{Connection, Dialer, Listener, TransportError}; +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,8 +179,14 @@ where }); } -fn endpoint(port: u16) -> IpEndpoint { - IpEndpoint::new(IpAddress::Ipv4(SERVER_IP), port) +/// As `StreamDialer::connect` takes it: a host string the adapter resolves. +/// +/// Derived from [`SERVER_IP`] rather than written out again — a second literal +/// can drift from the address the stack is actually configured with, and the +/// symptom is every test parking until the watchdog rather than an assertion +/// naming the mismatch. +fn server_host() -> alloc::string::String { + alloc::format!("{SERVER_IP}") } /// Exchange one framed request + reply over an already-connected pair, asserting @@ -232,8 +237,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 +268,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 dialer_b = 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"); + + // 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); @@ -296,13 +315,50 @@ fn two_concurrent_sessions() { }); } +/// `Clone` on the dialer shares its one socket rather than duplicating it — the +/// derive exists only to satisfy `SessionClientConnector`'s bound, which clones +/// per build. A clone dialing while the original holds the link must say so: +/// a bare `Io` is indistinguishable from the peer being down, and the client +/// engine would retry-loop forever without ever naming the real cause. +#[test] +fn a_cloned_dialer_reports_a_busy_socket() { + drive(|server_stack, client_stack| async move { + let mut listener = framed_listener(EmbassyNet::listen::<1>( + server_stack, + 7001u16, + [(buf(), buf())], + )); + let dialer = framed_dialer( + EmbassyNet::tcp(client_stack, buf(), buf()), + server_host(), + 7001, + ); + let clone = dialer.clone(); + + let (srv, cli) = futures::join!(listener.accept(), dialer.connect()); + let _srv = srv.expect("accept"); + let _cli = cli.expect("connect"); + + // `_cli` still holds the only socket. + assert_eq!( + clone.connect().await.err(), + Some(TransportError::Busy), + "a clone shares the socket, so the second dial is Busy, not Io" + ); + }); +} + /// Dialer reuses its single socket: a connect to a port with no listener fails /// (the peer stack RSTs), then a connect after a listener appears succeeds; and a /// connect after the previous link was dropped succeeds again. #[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 +368,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 +389,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 +430,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/framing.rs b/aimdb-tcp-connector/tests/framing.rs index 9f7ffa59..c1674824 100644 --- a/aimdb-tcp-connector/tests/framing.rs +++ b/aimdb-tcp-connector/tests/framing.rs @@ -82,3 +82,79 @@ 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(feature = "connector")] +mod framer { + use aimdb_core::session::{FrameFault, Framer, FramerFactory}; + use aimdb_tcp_connector::framing::{LengthFramer, LengthFramers}; + + #[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" + ); + } + + /// The cap is settable again: a `fn()` factory is stateless and could only + /// ever produce `DEFAULT_MAX_FRAME`, so `LengthFramers` carries it instead. + #[test] + fn the_factory_carries_its_bound_into_every_framer() { + let mut wire = Vec::new(); + + let bounded = LengthFramers::new(4).framer(); + assert_eq!( + bounded.encode(b"12345", &mut wire), + Err(FrameFault::Recoverable), + "a 5-byte frame exceeds the 4-byte cap this factory was built with" + ); + assert!(wire.is_empty()); + + // The same payload is fine under the default, so the bound really came + // from the factory rather than being hard-wired. + LengthFramers::default() + .framer() + .encode(b"12345", &mut wire) + .expect("well under the default cap"); + assert!(!wire.is_empty()); + } +} 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), diff --git a/aimdb-tokio-adapter/CHANGELOG.md b/aimdb-tokio-adapter/CHANGELOG.md index ce01dde3..c5c3aa55 100644 --- a/aimdb-tokio-adapter/CHANGELOG.md +++ b/aimdb-tokio-adapter/CHANGELOG.md @@ -9,6 +9,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Changed (breaking) +- **`TokioNet::listen` yields `std::io::Result` instead of `TransportResult`.** + `TransportError` cannot carry a cause — it is `Clone + PartialEq + Eq` in a + `no_std` crate — so every bind failure arrived as a bare `Io`: "port already + taken", "port privileged", and "no such interface" were indistinguishable, + though only the OS can tell them apart and an operator has to act on which. + `listen` is a constructor rather than a trait method, and this adapter is + `std`, so it is free to say. Callers using `?` or `.expect(..)` need no change; + they simply start reporting the reason. - **Issue #131 — `TokioRecordRegistrarExt` shrinks to `.buffer(cfg)` only.** `source`/`tap`/`transform` are inherent methods on the non-generic `aimdb_core::RecordRegistrar<'a, T>` (closures keep the `(ctx, producer)` arg order — typically only the import changes); `join_queue.rs` (`TokioJoinQueue`) is deleted with the `JoinFanInRuntime` family (join fan-in lives in core on `async-channel`). ### Removed (breaking) diff --git a/aimdb-tokio-adapter/src/net.rs b/aimdb-tokio-adapter/src/net.rs index 574d6f09..afe5454f 100644 --- a/aimdb-tokio-adapter/src/net.rs +++ b/aimdb-tokio-adapter/src/net.rs @@ -27,11 +27,15 @@ impl TokioNet { } /// Bind a TCP listener on `addr` (`"host:port"`). - pub async fn listen(addr: &str) -> TransportResult { - TcpListener::bind(addr) - .await - .map(TokioTcpListener) - .map_err(|_| TransportError::Io) + /// + /// Yields `std::io::Error` rather than [`TransportError`], which cannot + /// carry a cause: it is `Clone + PartialEq + Eq` in a `no_std` crate. A bind + /// failure is one an operator has to act on and cannot infer — the port is + /// taken, the port is privileged, the interface is absent — and only the OS + /// distinguishes them. This is a constructor, not a trait method, so it is + /// free to say which. + pub async fn listen(addr: &str) -> std::io::Result { + TcpListener::bind(addr).await.map(TokioTcpListener) } /// A UDP binder on `local_ip`, which sockets are bound to as @@ -78,6 +82,7 @@ where } /// Dials TCP connections. +#[derive(Clone, Copy, Default)] pub struct TokioTcpDialer; impl StreamDialer for TokioTcpDialer { @@ -180,7 +185,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`. @@ -190,14 +197,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; @@ -208,6 +216,26 @@ mod tests { } } + /// A bind failure names its cause. `TransportError` cannot carry one, so + /// `listen` yields `io::Error` — the difference between "port taken", + /// "port privileged" and "no such interface" is the whole content of the + /// error, and an operator cannot infer it from the address they supplied. + #[tokio::test] + async fn a_failed_bind_reports_why() { + let held = TokioNet::listen("127.0.0.1:0").await.expect("first bind"); + let addr = held.local_addr().expect("bound addr"); + + let Err(err) = TokioNet::listen(&addr.to_string()).await else { + panic!("binding a held port must fail"); + }; + + assert_eq!( + err.kind(), + std::io::ErrorKind::AddrInUse, + "the OS reason must survive, not collapse to a bare Io" + ); + } + #[tokio::test] async fn tcp_round_trips_between_dialer_and_listener() { let mut listener = TokioNet::listen("127.0.0.1:0").await.unwrap(); diff --git a/docs/design/052-runtime-neutral-connectors.md b/docs/design/052-runtime-neutral-connectors.md index 9524c7b4..c644242d 100644 --- a/docs/design/052-runtime-neutral-connectors.md +++ b/docs/design/052-runtime-neutral-connectors.md @@ -169,7 +169,7 @@ Three details the prototype settled: not `Rd`/`Wr` halves like today's `EmbassyConnection`. That is what lets it wrap an owned `embassy_net::tcp::TcpSocket`, whose `split()` yields only *borrowed* halves while `Connection` must own the socket — the exact reason - [`embassy_transport.rs`](../../aimdb-tcp-connector/src/embassy_transport.rs) + `embassy_transport.rs` (deleted by this design) gives for not reusing `connector-io` today. It costs nothing: `Connection`'s own `recv`/`send` already take `&mut self`, so reads and writes were already serialized. @@ -192,7 +192,7 @@ Three details the prototype settled: serial connector under `std`; that dependency does not belong in the adapter. - **`aimdb-embassy-adapter`**: `EmbassyNet::tcp(stack, rx, tx)`, `EmbassyNet::listen::(stack, endpoint, rx[N], tx[N])` (the socket-slot pool - moves here from [`aimdb-tcp-connector/src/embassy_transport.rs`](../../aimdb-tcp-connector/src/embassy_transport.rs)), + moves here from `aimdb-tcp-connector/src/embassy_transport.rs`, since deleted), `EmbassyNet::udp(stack, …)`, `EmbassyUart::split(rx, tx)`, and `Delay` returning `embassy_time::Timer`. Each stream/datagram newtype is `unsafe impl Send` and wraps the inner future in `SendFutureWrapper`. The diff --git a/examples/mqtt-connector-demo-common/src/lib.rs b/examples/mqtt-connector-demo-common/src/lib.rs index b080e105..397d24a3 100644 --- a/examples/mqtt-connector-demo-common/src/lib.rs +++ b/examples/mqtt-connector-demo-common/src/lib.rs @@ -16,7 +16,8 @@ //! # Compile-Time Safe Keys //! //! This crate also demonstrates the `RecordKey` derive macro for type-safe -//! record keys. See the [`keys`] module for examples. +//! record keys. See the `keys` module for examples — unlinked, as it exists +//! only behind the `derive` feature. #![cfg_attr(not(feature = "std"), no_std)]