Feat/platform agnostic tcp connectors

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

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`

aimdb-client/src/endpoint.rs

@@ -144,7 +144,14 @@ pub fn dial(endpoint: &str) -> ClientResult<Box<dyn Dialer>> {
         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::<u16>().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.

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

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<T>` 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

aimdb-core/src/session/client.rs

@@ -501,7 +501,7 @@ async fn client_loop<D, C>(
                 // `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;

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<u16>), 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"
+        );
+    }
+}