diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 323ae9b..1a8b3b9 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -311,7 +311,7 @@ seam described above (section 3) usable. | `src/socket_manager.rs` | Owns the spawned socket task; bridges `FramedRead`/`FramedWrite` to two mpsc channels; enforces the general inactivity timeout | | `src/client_inner.rs` | The client state machine: a `ControlMessage` enum plus a select loop matching responses to pending requests | | `src/client.rs` | Public `Client` — connect, routing activation, send/receive diagnostic messages | -| `src/server.rs` | `Server`, `ServerConnectionHandler`, `ClientConnectionInfo` | +| `src/server.rs` | `Server`, `ServerConnectionHandler`, `ResponseWriter`, `ClientConnectionInfo` | The client is a channel sandwich, described in one comment at the top of `src/client_inner.rs`: @@ -366,6 +366,9 @@ future change can silently reintroduce a dropped-`Sender` bug. regenerate them**. Note they cover *bodies*, encoded via `Encode`; they do not independently pin header payload types (see section 7.2). - `tests/integration_test.rs` — real client-against-real-server over loopback TCP. +- `tests/udp_identification.rs` — drives `Server::run_udp_responder` on a loopback + `UdpSocket`: an answered broadcast probe, silence for the directed EID/VIN + forms, and a responder that keeps serving after datagrams it cannot answer. - `tests/nested_encode.rs` — a permanent regression test for the encode hot path. - `examples/bare_metal_codec.rs` — encode into a `[u8; N]`, frame, decode; builds and runs with `--no-default-features`. This is the executable proof that the @@ -377,42 +380,38 @@ future change can silently reintroduce a dropped-`Sender` bug. ## 7. Known issues and deferred work Everything in this section was found by review during the handoff cleanup. It is -recorded here so the analysis does not have to be re-derived. None of it is -scheduled; all of it is a decision for the crate's next owner. +recorded here so the analysis does not have to be re-derived. Except where an +entry is marked RESOLVED, none of it is scheduled; it is a decision for the +crate's next owner. Resolved entries are kept because the analysis that led to +the fix is still the fastest way to understand the shape the API ended up with. -### 7.1 `ServerConnectionHandler::diagnostic_message` cannot express correct DoIP behavior +### 7.1 `ServerConnectionHandler::diagnostic_message` — RESOLVED in 0.4.0 -This is the one place where the API shape prevents a correct implementation. +DoIP prescribes that a DoIP entity receiving a diagnostic message first sends a +`DiagnosticMessageAck`, and then — separately and later — sends any functional +(e.g. UDS) response as its own `DiagnosticMessage`. Two messages, in order. + +The trait used to return a **single** `OwnedMessage`, so an implementer had to +choose one of the two and no handler could drive a real UDS tester. + +It now takes a sink instead: ```rust async fn diagnostic_message( &self, message: &DiagnosticMessage<'_>, -) -> Result; + responses: &mut dyn ResponseWriter, +) -> Result<(), Error>; ``` (`ServerConnectionHandler::diagnostic_message` in `src/server.rs`) -DoIP prescribes that a DoIP entity receiving a diagnostic message first sends a -`DiagnosticMessageAck`, and then — separately and later — sends any functional -(e.g. UDS) response as its own `DiagnosticMessage`. Two messages, in order. - -The trait returns a **single** `OwnedMessage`, and the dispatch site -(the `OwnedPayload::DiagnosticMessage` arm of `Server::handle_client_message`) -maps that one value through `Some(..)` into `Server::handle_client_connection`, -whose read loop writes at most one message per received message. There is no -path by which a handler can emit both. - -So an implementer must choose: send the required acknowledgement, or send the -functional response. `examples/echo_server.rs` picks the acknowledgement and -smuggles the request bytes back inside the ack's `previous_message_data` field — -which is an echo demo, not protocol-correct behavior, and the example says so in -a comment (in its `diagnostic_message` implementation, `examples/echo_server.rs`). - -**Recommendation:** revisit the trait signature. Plausible shapes are returning a -collection, taking a sink/writer the handler can push to, or splitting the ack -decision (which the server could synthesize itself) from the response. +Each `ResponseWriter::send` writes straight to the connection's framed write +half, so a handler emits as many messages as the exchange needs — ack, any +number of NRC `0x78` "response pending" messages, then the final answer — and +may await arbitrary work between them. `examples/echo_server.rs` shows the +two-message shape. -Note that `routing_activation` has the same single-`OwnedMessage` return shape, +Note that `routing_activation` still has the single-`OwnedMessage` return shape, but that is fine — routing activation genuinely is one request, one response. ### 7.2 `diagnostic_message_ack` hardcodes the positive acknowledgement payload type @@ -568,11 +567,20 @@ returning `true` breaks it. Either delete the field or make the loop honor it. ### 7.6 Other rough edges These are documented in `README.md` under **Status** and are repeated here only -as a pointer: no TLS; no UDP vehicle announcement or discovery; the server's -accept loop serves one TCP connection at a time; entity status and vehicle -identification requests over TCP are silently dropped; -`ClientConnectionInfo::logical_address` is hard-coded to `0x0000` because the -server tracks no per-connection state; a failed `accept()` panics the server task. +as a pointer: no TLS; no unsolicited UDP vehicle announcement at power-on +(identification requests over UDP *are* answered, but only by +`Server::run_udp_responder` on a socket the caller binds and drives — `run_server` +binds TCP alone — and only the broadcast `0x0001` form, since `Payload::decode` +discards the EID/VIN the directed forms name); the server's accept loop serves +one TCP connection at a time; entity status and vehicle identification requests +over TCP are silently dropped; `ClientConnectionInfo::logical_address` is +hard-coded to `0x0000` because the server tracks no per-connection state; the +handler passed to `Server::new` is not validated. + +A failed `accept()` no longer panics the server task — as of 0.4.0 both the TCP +accept loop and the UDP responder log the error, sleep briefly, and continue, so +neither a transient peer reset nor a persistent condition such as `EMFILE` can +take the entity down or spin a core. --- diff --git a/Cargo.lock b/Cargo.lock index f545e8d..c06ef3a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -330,7 +330,7 @@ dependencies = [ [[package]] name = "simple_doip" -version = "0.3.1" +version = "0.4.0" dependencies = [ "anyhow", "async-trait", diff --git a/Cargo.toml b/Cargo.toml index 2a3dfc2..b56a213 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "simple_doip" -version = "0.3.1" +version = "0.4.0" edition = "2024" rust-version = "1.88" description = "An ISO 13400-2 (DoIP) implementation with a no_std, zero-copy protocol core and optional async client and server" @@ -49,6 +49,10 @@ name = "golden_vectors" name = "integration_test" required-features = ["client", "server"] +[[test]] +name = "udp_identification" +required-features = ["server"] + [[example]] name = "simple_client" required-features = ["client"] diff --git a/README.md b/README.md index 980d890..850bd79 100644 --- a/README.md +++ b/README.md @@ -18,22 +18,27 @@ gaps a new integrator should know about before relying on them: - **No TLS.** Connections are established in the clear on `TCP_PORT` (`13400`); `TCP_TLS_PORT` (`3496`) is defined per ISO 13400-2 but nothing in this crate uses it. -- **No UDP vehicle announcement / discovery.** The server does not send the - UDP vehicle-announcement broadcast on startup, nor answer vehicle - identification requests over UDP. -- **The server accepts one TCP connection at a time.** `Server::run_server`'s - accept loop awaits each client's connection handling to completion before - calling `accept()` again, so a second client cannot connect while the first - is still being served. +- **No unsolicited UDP vehicle announcement.** The server never sends the UDP + vehicle-announcement broadcast on startup, so a tester learns of an entity + only by asking. Vehicle identification requests over UDP *are* answered, but + only by `Server::run_udp_responder`, on a `UdpSocket` the caller bound and + drives: `run_server` binds TCP alone, so an entity that starts through it + and nothing else is invisible to a discovery probe. `run_udp_responder` + answers the broadcast request form (`0x0001`) only; the directed + with-EID (`0x0002`) and with-VIN (`0x0003`) forms are declined, because + `Payload::decode` discards the EID/VIN bytes and the responder cannot tell + whether it is the addressee. +- **The server accepts one TCP connection at a time.** The accept loop in + `Server::run_server_with_listener` (which `Server::run_server` delegates to) + awaits each client's connection handling to completion before calling + `accept()` again, so a second client cannot connect while the first is still + being served — and one tester that connects and then stalls wedges that + entity until it disconnects. - **Entity status requests and vehicle identification requests over TCP are silently dropped.** `Server::handle_client_message` logs a warning and sends no reply for either, so a tester that asks gets silence rather than an error - or a negative response. -- **A connection handler can only answer a diagnostic message with a single - message.** `ServerConnectionHandler::diagnostic_message` returns one - `OwnedMessage`, so a handler can send the required acknowledgement *or* a - functional response, not the acknowledgement followed by a separate - response as DoIP prescribes. + or a negative response. Identification requests are answered on the UDP path + only. - **A `DiagnosticMessage` arriving while the client is waiting for an ACK is silently discarded.** After `Client::send_diagnostic_message`, the inner client is in its `AwaitAck` state; a `DiagnosticMessage` that arrives before @@ -46,11 +51,10 @@ gaps a new integrator should know about before relying on them: - **`ClientConnectionInfo::logical_address` is always `0x0000`.** The server does not yet track per-connection logical addresses, so this field is a placeholder rather than the client's real address. -- The handler passed to `Server::new` is not validated, and a failed - `accept()` currently panics the server task rather than being handled. +- The handler passed to `Server::new` is not validated. None of this blocks bare-metal or single-client use; it matters if you need -concurrent clients, discovery, or TLS today. +concurrent clients, unsolicited announcement, or TLS today. ## Quickstart @@ -127,9 +131,14 @@ cargo test --features client,server `TCP_PORT` (`13400`), returning `Error::InvalidPort` — pointing a client at a non-standard port requires your own `Connector` implementation. - `echo_server` answers a diagnostic message with a positive acknowledgement - that carries the received bytes back in its previous-message-data field; see - the single-response handler limitation under [Status](#status). + `echo_server` answers a diagnostic message the way DoIP prescribes: first a + positive acknowledgement carrying the received bytes back in its + previous-message-data field, then the echo itself as a separate + `DiagnosticMessage`. Both are written into the `ResponseWriter` the handler + is given, which is the shape a real UDS response takes. + + `echo_server` calls `run_server`, so it serves TCP only and does not answer + UDP discovery probes; see `Server::run_udp_responder` for that half. ## Relationship to `automotive-wire-codec` diff --git a/examples/echo_server.rs b/examples/echo_server.rs index 0c725ab..ab9da94 100644 --- a/examples/echo_server.rs +++ b/examples/echo_server.rs @@ -6,7 +6,7 @@ use simple_doip::{ DiagnosticAckCode, DiagnosticMessage, OwnedMessage, RoutingActivationRequest, RoutingActivationResponseCode, }, - server::{Server, ServerConnectionHandler}, + server::{ResponseWriter, Server, ServerConnectionHandler}, }; use tracing::{debug, info}; @@ -53,25 +53,35 @@ impl ServerConnectionHandler for ServerHandler { async fn diagnostic_message( &self, message: &DiagnosticMessage<'_>, - ) -> Result { + responses: &mut dyn ResponseWriter, + ) -> Result<(), Error> { debug!( "Received diagnostic message from {:?} to {:?}", message.source_address, message.target_address ); // A DoIP entity must acknowledge a diagnostic message first; any - // functional (UDS) response is a separate, later DiagnosticMessage. - // `ServerConnectionHandler::diagnostic_message` can only return a single - // message, so this example echoes the received bytes back inside the - // positive acknowledgement's previous-message-data field. A real entity - // that must also send a UDS response needs its own write path; the - // handler trait as it stands cannot emit two messages for one request. - Ok(OwnedMessage::diagnostic_message_ack( - self.protocol_version(), - message.target_address, // We are the target, so we answer as source - message.source_address, // ...back to the tester that asked - DiagnosticAckCode::RoutingConfirmationAck, - message.user_data.to_vec(), - )) + // functional (UDS) response is a separate, later DiagnosticMessage. Both + // go through `responses`, in that order, which is the sequence a UDS + // tester waits for. + responses + .send(OwnedMessage::diagnostic_message_ack( + self.protocol_version(), + message.target_address, // We are the target, so we answer as source + message.source_address, // ...back to the tester that asked + DiagnosticAckCode::RoutingConfirmationAck, + message.user_data.to_vec(), + )) + .await?; + // The echo itself: a real entity would put its UDS response here. + responses + .send(OwnedMessage::diagnostic_message( + self.protocol_version(), + message.target_address, + message.source_address, + message.user_data.to_vec(), + )) + .await?; + Ok(()) } } diff --git a/src/messages/mod.rs b/src/messages/mod.rs index bd770e0..c53404b 100644 --- a/src/messages/mod.rs +++ b/src/messages/mod.rs @@ -464,6 +464,37 @@ impl OwnedMessage { payload: OwnedPayload::DiagnosticMessageAck(ack), } } + + /// Construct a directed reply to a vehicle identification request. + /// + /// The header is stamped with [`PayloadType::VehicleAnnouncement`] (0x0004) + /// because ISO 13400-2 defines exactly one wire payload type for both the + /// unsolicited announcement and the directed reply. The payload nonetheless + /// uses [`OwnedPayload::VehicleIdentificationResponse`], which encodes + /// identically but records at construction time that this is an answer to a + /// request rather than a spontaneous announcement. A peer decoding these + /// bytes gets [`Payload::VehicleAnnouncement`] either way. + /// + /// # Panics + /// Panics if the payload's `encoded_size` errors, or if the resulting size + /// does not fit in a `u32`. Neither is reachable here: the payload is fixed + /// size (33 bytes) and `encoded_size` is pure arithmetic over the struct's + /// own fields, with no I/O to fail. + #[must_use] + pub fn vehicle_identification_response( + protocol_version: ProtocolVersion, + response: VehicleIdentificationResponse, + ) -> OwnedMessage { + let payload_size = payload_len(&Payload::VehicleIdentificationResponse(response)); + OwnedMessage { + header: Header::new( + protocol_version, + PayloadType::VehicleAnnouncement, + payload_size, + ), + payload: OwnedPayload::VehicleIdentificationResponse(response), + } + } } /// Encode delegates through the borrowed view so there is exactly one wire diff --git a/src/server.rs b/src/server.rs index d5fe6de..7d654ef 100644 --- a/src/server.rs +++ b/src/server.rs @@ -9,9 +9,9 @@ use crate::{ logical_address::LogicalAddress, message_codec::MessageCodec, messages::{ - DiagnosticMessage, DiagnosticPowerModeCode, FurtherActionRequired, OwnedMessage, - OwnedPayload, ProtocolVersion, RoutingActivationRequest, VehicleIdentificationResponse, - VinGidSyncStatus, + Decode, DiagnosticMessage, DiagnosticPowerModeCode, Encode, FurtherActionRequired, Message, + OwnedMessage, OwnedPayload, Payload, PayloadType, ProtocolVersion, + RoutingActivationRequest, VehicleIdentificationResponse, VinGidSyncStatus, }, }; use async_trait::async_trait; @@ -23,10 +23,23 @@ use std::{ Arc, atomic::{AtomicUsize, Ordering}, }, + time::Duration, +}; +use tokio::{ + net::{TcpListener, TcpStream, UdpSocket}, + time::sleep, }; -use tokio::net::{TcpListener, TcpStream}; use tokio_util::codec::{FramedRead, FramedWrite}; -use tracing::{error, warn}; +use tracing::{debug, error, warn}; + +/// How long a socket loop waits before retrying after a non-fatal error. +/// +/// Neither the TCP accept loop nor the UDP responder gives up on an error, so +/// each needs a floor on its retry rate: a condition that does not clear on its +/// own — descriptor exhaustion is the usual one — makes the failing call return +/// immediately, and retrying it without a delay pegs a core and floods the log. +/// Short enough that a genuinely transient error costs one interval and no more. +const ACCEPT_ERROR_BACKOFF: Duration = Duration::from_millis(100); /// Identifies the tester on the other end of a `DoIP` TCP connection, passed to /// [`ServerConnectionHandler`] methods so an implementation can tell which peer @@ -65,6 +78,43 @@ impl Drop for ActiveConnectionGuard<'_> { self.active_connections.fetch_sub(1, Ordering::Relaxed); } } + +/// Sink a [`ServerConnectionHandler`] writes its responses into. +/// +/// Exists so one request can produce several messages - the +/// `DiagnosticMessageAck` that ISO 13400 requires before a UDS response, and any +/// number of NRC `0x78` "response pending" messages before the final answer. +/// Each `send` goes straight to the socket, so a handler may await arbitrary work +/// between calls and the tester observes a genuinely *held* pending wait rather +/// than a burst delivered all at once. +#[async_trait] +pub trait ResponseWriter: Send { + /// Write one message to the tester, in call order. + /// + /// # Errors + /// Returns an [`Error`] if the message cannot be encoded or the socket write + /// fails. + async fn send(&mut self, message: OwnedMessage) -> Result<(), Error>; +} + +/// [`ResponseWriter`] over the connection's framed write half. +struct FramedResponseWriter<'a, W> { + sink: &'a mut FramedWrite, +} + +#[async_trait] +impl ResponseWriter for FramedResponseWriter<'_, W> +where + W: tokio::io::AsyncWrite + Unpin + Send, +{ + async fn send(&mut self, message: OwnedMessage) -> Result<(), Error> { + // The codec reports encode failures as `MessageError`; `?` widens it to the + // crate's `Error` so handlers only ever deal with one error type. + self.sink.send(&message).await?; + Ok(()) + } +} + /// Trait for handling `DoIP` connections as a server. /// Implement this trait to create a custom `DoIP` server. /// Most protocol functions have a simple, default implementation @@ -98,16 +148,20 @@ pub trait ServerConnectionHandler { request: &RoutingActivationRequest, ) -> Result; - /// Handle a diagnostic message addressed to this entity - /// and build the acknowledgement/response message to send back. + /// Handle a diagnostic message addressed to this entity, writing zero or more + /// responses into `responses`. + /// + /// A UDS tester expects a `DiagnosticMessageAck` before any response, so a + /// typical implementation sends the ack first and the UDS payload second. /// /// # Errors - /// Returns an [`Error`] if the message cannot be processed or the response - /// cannot be constructed. + /// Returns an [`Error`] if the message cannot be processed or a response + /// cannot be written. async fn diagnostic_message( &self, message: &DiagnosticMessage<'_>, - ) -> Result; + responses: &mut dyn ResponseWriter, + ) -> Result<(), Error>; // Optional Functions // These functions *may* be overridden to provide custom behavior @@ -210,7 +264,12 @@ pub trait ServerConnectionHandler { } /// A running `DoIP` entity: accepts TCP connections and dispatches incoming -/// messages to a [`ServerConnectionHandler`] implementation. +/// messages to a [`ServerConnectionHandler`] implementation, and — on a socket +/// the caller supplies to [`run_udp_responder`](Self::run_udp_responder) — +/// answers UDP vehicle-identification probes from the same handler. +/// +/// The two halves are independent futures, so an entity that wants both must +/// drive both; see [`run_server`](Self::run_server). #[derive(Debug)] pub struct Server { connection_handler: Arc, @@ -233,17 +292,71 @@ where }) } - /// Start listening for incoming `DoIP` TCP connections + /// Start listening for incoming `DoIP` TCP connections on the standard + /// [`TCP_PORT`] across all interfaces. /// - /// # Errors - /// Returns an [`Error`] if the TCP listener cannot be bound + /// Connections are served one at a time, as described on + /// [`run_server_with_listener`](Self::run_server_with_listener), which this + /// delegates to after binding. + /// + /// # Discovery + /// This binds **TCP only**. An entity started through this method and + /// nothing else answers no UDP vehicle-identification probe, so a tester + /// that does not already know its IP address will never find it — and + /// nothing logs that fact, because no datagram is ever received. /// - /// # Panics - /// Panics if accepting a new TCP client connection fails + /// Discovery lives in [`run_udp_responder`](Self::run_udp_responder), on a + /// socket the caller binds. Both methods take `&self`, and neither + /// completes normally — this method can still return early if the bind + /// fails, per the `# Errors` section below — so compose them on one + /// `Server`: + /// + /// ``` + /// # use simple_doip::{Error, UDP_DISCOVERY_PORT, server::{Server, ServerConnectionHandler}}; + /// # use tokio::net::UdpSocket; + /// # async fn serve(server: &Server) -> Result<(), Error> { + /// let socket = UdpSocket::bind(("0.0.0.0", UDP_DISCOVERY_PORT)).await?; + /// tokio::try_join!(server.run_server(), server.run_udp_responder(socket))?; + /// # Ok(()) + /// # } + /// ``` + /// + /// Unsolicited vehicle announcement at power-on is a separate thing again, + /// and this crate does not implement it in any form. + /// + /// # Errors + /// Returns an [`Error`] if the TCP listener cannot be bound. pub async fn run_server(&self) -> Result<(), Error> { - // TODO: Vehicle Announcement over UDP + // TODO: unsolicited Vehicle Announcement over UDP at power-on. Answering + // vehicle identification requests already exists as + // `run_udp_responder`, which the caller drives with its own socket. let tcp_listener = TcpListener::bind(("0.0.0.0", TCP_PORT)).await?; + self.run_server_with_listener(tcp_listener).await + } + + /// Serve connections from a listener the caller already bound. + /// + /// Lets the caller choose the interface and port — a loopback alias such as + /// `127.0.0.2:13400` so several entities coexist on one host, or port `0` + /// for an OS-assigned port the caller reads back with + /// [`TcpListener::local_addr`] before calling this. + /// + /// **One connection at a time.** The loop awaits each accepted connection's + /// handling to completion before calling `accept()` again, so a tester that + /// connects and then stalls wedges this entity until it disconnects; a + /// second tester is not even accepted meanwhile. That matters most for the + /// multi-entity topology above, where each entity gets its own listener but + /// each also gets its own single-connection bottleneck. See the **Status** + /// section of `README.md` for the full account. + /// + /// # Errors + /// This method does not currently return. Both accept errors and handler + /// errors are logged and the loop continues, so the future never resolves. + /// The `Result` is retained so [`run_server`](Self::run_server) can + /// propagate its bind failure through a matching return type, and so a + /// future shutdown path has somewhere to report one. + pub async fn run_server_with_listener(&self, tcp_listener: TcpListener) -> Result<(), Error> { loop { match tcp_listener.accept().await { Ok((tcp_stream, client_socket_addr)) => { @@ -255,15 +368,196 @@ where } } Err(accept_error) => { - // TODO: Don't panic here, this might happen - panic!("Failed to accept new TCP client: {accept_error}"); + // An accept error must not take the entity down — a + // simulator that aborts on a peer resetting between the SYN + // and our accept turns a client bug into an opaque + // transport failure. + error!("Failed to accept TCP client, continuing: {accept_error}"); + // Not every accept error is transient. Descriptor + // exhaustion (EMFILE/ENFILE) persists until something else + // in the process releases an fd, and until then `accept` + // fails immediately on every iteration — an unbounded retry + // would spin a core and flood the log, which is harder to + // diagnose in an unattended simulator than the panic this + // replaced. The delay bounds that to ten retries a second + // and costs a genuinely transient error only one interval. + sleep(ACCEPT_ERROR_BACKOFF).await; + } + } + } + } + + /// Answer UDP vehicle-identification probes on a caller-bound socket. + /// + /// ISO 13400-2 puts vehicle identification on UDP + /// [`crate::UDP_DISCOVERY_PORT`]: a tester broadcasts a + /// `VehicleIdentificationRequest` and every entity that matches answers with + /// its identity, which is how a tester discovers entities it has no address + /// for. Without this an entity is reachable only by a tester that already + /// knows its IP. + /// + /// The socket is bound by the caller, not here, so an entity can sit on one + /// specific interface (several simulated entities coexisting on loopback + /// aliases, say) or on an ephemeral port under test. The flip side is that + /// the caller owns whether real discovery works at all: a tester broadcasts + /// its request, and a socket bound to a specific unicast address does not + /// receive broadcast datagrams, so answering real probes means binding + /// `0.0.0.0` on [`crate::UDP_DISCOVERY_PORT`] — a narrower bind serves only + /// testers that already know the address, which is the case discovery + /// exists to solve. The response content + /// comes from + /// [`ServerConnectionHandler::received_vehicle_identification_request`], so an + /// implementation customizes what it announces without reimplementing the + /// datagram loop. + /// + /// Every failure inside the loop - a socket error, an undecodable datagram, a + /// payload this responder does not answer, or a handler that declines to + /// produce an identity - is logged and skipped rather than returned. A UDP + /// socket is reachable by every host on the network, so any fatal path here + /// would hand an arbitrary host a way to end discovery for the life of the + /// process. This matches the reasoning behind the TCP accept loop in + /// [`run_server_with_listener`](Self::run_server_with_listener). + /// + /// # Known limitation + /// Only the plain request form (0x0001) is answered. The with-EID (0x0002) + /// and with-VIN (0x0003) forms name a specific entity, but [`Payload::decode`] + /// collapses all three into [`Payload::VehicleIdentificationRequest`] and + /// discards the EID or VIN bytes, so this responder cannot tell whether it is + /// the addressee. It stays silent rather than answering a probe that may have + /// been meant for a different entity: a wrong answer actively misleads a + /// tester, whereas silence degrades to a discovery timeout that testers + /// already handle. Consequently the + /// [`ServerConnectionHandler::vehicle_identification_with_eid`] and + /// [`ServerConnectionHandler::vehicle_identification_with_vin`] hooks are + /// never consulted. Answering the directed forms requires [`Payload`] to + /// preserve the EID/VIN through decoding. + /// + /// # Errors + /// This method does not currently return. Socket, decode, and handler errors + /// are all logged and the loop continues, so the future never resolves. The + /// `Result` is retained so a caller can compose this with the equally + /// non-returning [`run_server_with_listener`](Self::run_server_with_listener), + /// and so a future shutdown path has somewhere to report one. + pub async fn run_udp_responder(&self, socket: UdpSocket) -> Result<(), Error> { + // A vehicle identification request is 8 bytes and its response 41, so + // this is generous. Anything longer is not a message this loop answers. + let mut buf = [0u8; 1024]; + loop { + // A receive error must not be fatal. On Windows an oversized datagram + // fails `recvfrom` with `WSAEMSGSIZE` instead of truncating the way + // Linux does, so a single 2 KB packet from anyone on the network + // would otherwise kill discovery permanently. + let (len, peer) = match socket.recv_from(&mut buf).await { + Ok(received) => received, + Err(recv_error) => { + warn!("UDP receive failed, continuing: {recv_error}"); + // As in the accept loop: a socket error that persists (the + // interface going away under a bound socket, say) would + // otherwise return immediately on every iteration and spin + // this loop at full speed. Only the socket-error path needs + // the delay — the decode and handler paths below consumed a + // datagram, so they are already paced by the peer. + sleep(ACCEPT_ERROR_BACKOFF).await; + continue; + } + }; + + // `Decode` is implemented for the borrowed `Message<'a>` and yields + // (message, remaining_bytes). The borrow of `buf` ends with this + // iteration, before the next `recv_from` overwrites it. + let (message, _rest) = match Message::decode(&buf[..len]) { + Ok(decoded) => decoded, + Err(decode_error) => { + warn!("Undecodable UDP datagram from {peer}, ignoring: {decode_error}"); + continue; + } + }; + + if !matches!(message.payload, Payload::VehicleIdentificationRequest) { + // Routine, not a fault: on `0.0.0.0:13400` this socket sees + // every DoIP datagram on the network, most of which this + // responder is not the addressee for. Warning about them would + // make a healthy entity look broken. + debug!( + "Unsupported UDP payload type {:?} from {peer}, ignoring", + message.header.payload_type + ); + continue; + } + + // 0x0002/0x0003 name a specific entity by EID/VIN, but `Payload::decode` + // drops those bytes, so we cannot tell whether we are the addressee. + // Answering regardless would actively mislead a tester; staying quiet + // degrades to a timeout, which testers already handle. See the method's + // known-limitation note. + if !matches!( + message.header.payload_type, + PayloadType::VehicleIdentificationRequest + ) { + // Also routine: a tester doing directed discovery on a live + // network sends these as a matter of course, and declining is + // the designed behavior rather than a problem to report. + debug!( + "Ignoring directed identification request {:?} from {peer}: this crate \ + cannot match the EID/VIN it names", + message.header.payload_type + ); + continue; + } + + // UDP carries no connection, so there is no routing activation to + // have learned a tester logical address from; `0x0000` matches what + // the TCP path currently supplies. + let client_info = ClientConnectionInfo { + ip_address: peer.ip(), + logical_address: LogicalAddress(0x0000), + }; + // An implementation is entitled to fail transiently - identity not yet + // read out of NVM at power-on, say - so one refusal must cost this + // probe only, not every future one. + let response = match self + .connection_handler + .received_vehicle_identification_request(&client_info) + { + Ok(response) => response, + Err(handler_error) => { + warn!("Identification handler failed for {peer}, skipping: {handler_error}"); + continue; + } + }; + let reply = OwnedMessage::vehicle_identification_response( + self.connection_handler.protocol_version(), + response, + ); + + // Mirrors `MessageCodec`'s `Encoder` impl: size the message, encode + // into a `Vec`, then write it. There is no framing to do - a + // datagram is already one message. + let mut encoded = match reply.encoded_size() { + Ok(size) => std::vec::Vec::with_capacity(size), + Err(size_error) => { + warn!("Failed to size identification response for {peer}: {size_error}"); + continue; } + }; + if let Err(encode_error) = reply.encode(&mut encoded) { + warn!("Failed to encode identification response for {peer}: {encode_error}"); + continue; + } + + if let Err(send_error) = socket.send_to(&encoded, peer).await { + // ENETUNREACH, a firewall EPERM - transient and peer-specific. + warn!("Failed to answer identification probe from {peer}: {send_error}"); } } } /// Handle an individual client TCP connection, reading and responding to messages /// + /// Sets `TCP_NODELAY` on `tcp_stream`, overriding the caller's setting if it + /// configured one, because diagnostics write consecutive small frames whose + /// latency Nagle would otherwise inflate (see the comment on the call). + /// /// # Errors /// Returns an [`Error`] if message handling or response encoding fails pub async fn handle_client_connection( @@ -272,6 +566,19 @@ where tcp_stream: TcpStream, ) -> Result<(), Error> { let _active_connection_guard = ActiveConnectionGuard::new(&self.active_connections); + + // Diagnostics are a request/response conversation of small frames: an + // ack, then a response, then often several NRC 0x78 pendings. With + // Nagle enabled the second small write waits on the peer's delayed + // ACK of the first — up to ~40ms per exchange, straight out of the P2 + // budget. `ConnectorSocket` already disables it on the client side + // (`connection.rs`); an accepted socket needs the same treatment. + // A failure here is not fatal: the connection still works, just with + // worse latency, so log and carry on rather than dropping the tester. + if let Err(nodelay_error) = tcp_stream.set_nodelay(true) { + warn!("Failed to set TCP_NODELAY for {client_socket_addr}: {nodelay_error}"); + } + let (rx, tx) = tcp_stream.into_split(); let mut read_stream = FramedRead::new(rx, MessageCodec::new()); let mut write_sink = FramedWrite::new(tx, MessageCodec::new()); @@ -280,7 +587,7 @@ where match read_stream.next().await { Some(Ok(message)) => { if let Some(response) = self - .handle_client_message(client_socket_addr, message) + .handle_client_message(client_socket_addr, message, &mut write_sink) .await? { write_sink.send(&response).await?; @@ -302,11 +609,21 @@ where } } - async fn handle_client_message( + /// Dispatch one decoded request to the handler. + /// + /// Returns the single response the caller must write, or `None` when there is + /// nothing left to send - either because the message needs no answer, or + /// because the handler already wrote its responses into `write_sink` itself + /// (the diagnostic-message path, which may emit several messages). + async fn handle_client_message( &self, client_socket_addr: SocketAddr, request_message: OwnedMessage, - ) -> Result, Error> { + write_sink: &mut FramedWrite, + ) -> Result, Error> + where + W: tokio::io::AsyncWrite + Unpin + Send, + { // TODO: Need to handle active sockets by adding clients to a map // client count should come from that map, as well as the logical address missing below let connection_info = ClientConnectionInfo { @@ -320,11 +637,13 @@ where .alive_check(&connection_info) .await .map(Some), - OwnedPayload::DiagnosticMessage(diagnostic_message) => self - .connection_handler - .diagnostic_message(&diagnostic_message.as_ref()) - .await - .map(Some), + OwnedPayload::DiagnosticMessage(diagnostic_message) => { + let mut responses = FramedResponseWriter { sink: write_sink }; + self.connection_handler + .diagnostic_message(&diagnostic_message.as_ref(), &mut responses) + .await?; + Ok(None) + } OwnedPayload::EntityStatusRequest => { warn!( "Entity Status Request is not yet supported, ignoring. source: {client_socket_addr}" diff --git a/tests/integration_test.rs b/tests/integration_test.rs index 7ae2f91..e05c7c3 100644 --- a/tests/integration_test.rs +++ b/tests/integration_test.rs @@ -1,11 +1,13 @@ //! End-to-end integration tests that exercise a real client and server over an //! actual TCP socket on localhost. //! -//! The server side is driven through [`Server::handle_client_connection`], which is -//! the same per-connection entry point [`Server::run_server`] uses internally. This -//! lets each test bind to `127.0.0.1:0` (an OS-assigned ephemeral port) instead of the -//! fixed [`simple_doip::TCP_PORT`] that `run_server` hardcodes, keeping the tests free -//! of port collisions. +//! Most tests drive the server through [`Server::handle_client_connection`], the same +//! per-connection entry point the accept loop uses internally, so they can bind +//! `127.0.0.1:0` (an OS-assigned ephemeral port) instead of the fixed +//! [`simple_doip::TCP_PORT`] that [`Server::run_server`] hardcodes, keeping the tests +//! free of port collisions. The tests covering the accept loop itself instead pass a +//! listener they bound on an ephemeral port to [`Server::run_server_with_listener`], +//! which exercises the shipped loop at no cost in port collisions. //! //! The client side uses the real [`Client`] API, but with a small test-only //! [`Connector`] implementation instead of [`simple_doip::connection::ConnectorSocket`]. @@ -15,15 +17,17 @@ //! Implementation" example in `src/connection.rs`) and requires no changes to `src/`. use async_trait::async_trait; +use futures::{SinkExt, StreamExt}; use simple_doip::{ Error, LogicalAddress, client::{AddressType, Client, ClientOptions, RoutingActivationOptions}, connection::Connector, + message_codec::MessageCodec, messages::{ ActivationTypeCode, DiagnosticAckCode, DiagnosticMessage, Encode, OwnedMessage, - ProtocolVersion, RoutingActivationRequest, RoutingActivationResponseCode, + OwnedPayload, ProtocolVersion, RoutingActivationRequest, RoutingActivationResponseCode, }, - server::{Server, ServerConnectionHandler}, + server::{ResponseWriter, Server, ServerConnectionHandler}, }; use std::{ net::{IpAddr, SocketAddr}, @@ -34,13 +38,14 @@ use std::{ time::Duration, }; use tokio::{ - io::{AsyncReadExt, AsyncWriteExt}, + io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt}, net::{ TcpListener, TcpStream, tcp::{OwnedReadHalf, OwnedWriteHalf}, }, task::JoinHandle, }; +use tokio_util::codec::{FramedRead, FramedWrite}; /// Generous but finite bound for every await in these tests, so a regression that hangs /// the client or server fails the test quickly instead of hanging CI. @@ -59,6 +64,28 @@ async fn with_timeout(context: &str, fut: F) -> F::Outpu .unwrap_or_else(|_| panic!("timed out waiting for: {context}")) } +/// Send the positive `DiagnosticMessageAck` that ISO 13400 requires before any UDS +/// response, addressed from this entity back to the requesting tester and echoing the +/// request's bytes. +/// +/// Every handler below needs this exact ack and differs only in what it does *after* +/// it, so it lives here once instead of being re-derived (and re-mistyped) per handler. +async fn send_positive_ack( + handler: &impl ServerConnectionHandler, + message: &DiagnosticMessage<'_>, + responses: &mut dyn ResponseWriter, +) -> Result<(), Error> { + responses + .send(OwnedMessage::diagnostic_message_ack( + handler.protocol_version(), + handler.get_logical_address(), + message.source_address, + DiagnosticAckCode::RoutingConfirmationAck, + message.user_data.to_vec(), + )) + .await +} + /// Test [`ServerConnectionHandler`]. Always accepts routing activation and positively /// acknowledges diagnostic messages, recording the last diagnostic payload it received /// so tests can assert on it (the [`Client`] facade only surfaces ack success/failure, @@ -105,15 +132,10 @@ impl ServerConnectionHandler for TestHandler { async fn diagnostic_message( &self, message: &DiagnosticMessage<'_>, - ) -> Result { + responses: &mut dyn ResponseWriter, + ) -> Result<(), Error> { *self.last_diagnostic_payload.lock().unwrap() = Some(message.user_data.to_vec()); - Ok(OwnedMessage::diagnostic_message_ack( - self.protocol_version(), - message.source_address, - message.target_address, - DiagnosticAckCode::RoutingConfirmationAck, - message.user_data.to_vec(), - )) + send_positive_ack(self, message, responses).await } } @@ -160,8 +182,29 @@ async fn start_server() -> TestServer { last_diagnostic_payload: Arc::clone(&last_diagnostic_payload), routing_activation_requests: Arc::clone(&routing_activation_requests), }; - let server = Server::new(handler).expect("server should construct"); + let (addr, accept_loop) = start_server_with(handler).await; + + TestServer { + addr, + last_diagnostic_payload, + routing_activation_requests, + accept_loop, + } +} + +/// Start a [`Server`] with a caller-supplied handler on an OS-assigned localhost port. +/// [`start_server`] delegates here; tests needing a handler other than [`TestHandler`] +/// call this directly. +/// +/// The accept loop awaits each connection INLINE, mirroring `run_server`, so a panic +/// escaping a handler kills the loop here exactly as it would in production - which is +/// what [`TestServer::shutdown`] asserts against. +async fn start_server_with(handler: H) -> (SocketAddr, JoinHandle<()>) +where + H: ServerConnectionHandler + Send + Sync + 'static, +{ + let server = Server::new(handler).expect("server should construct"); let listener = TcpListener::bind(("127.0.0.1", 0)) .await .expect("failed to bind test server to an ephemeral port"); @@ -181,12 +224,64 @@ async fn start_server() -> TestServer { } }); - TestServer { - addr, - last_diagnostic_payload, - routing_activation_requests, - accept_loop, - } + (addr, accept_loop) +} + +/// Read one framed message off a raw socket, as an [`OwnedMessage`]. +/// +/// Raw rather than through [`Client`], because the `Client` facade consumes the +/// `DiagnosticMessageAck` internally and never surfaces it - and the ack is exactly what +/// the multi-response tests assert on. +/// +/// This and the two send helpers below take an already-split [`FramedRead`]/ +/// [`FramedWrite`] half rather than a bare [`TcpStream`], because a codec must not be +/// reconstructed per call: a fresh `FramedRead` drops whatever the previous one +/// buffered, which silently loses a message when two arrive in one TCP segment. +async fn read_message(framed: &mut FramedRead) -> OwnedMessage +where + R: AsyncRead + Unpin, +{ + with_timeout("read message", framed.next()) + .await + .expect("stream closed before a message arrived") + .expect("decode message") +} + +/// Send a routing activation request over a raw socket. +async fn send_routing_activation( + framed: &mut FramedWrite, + source_address: LogicalAddress, +) where + W: AsyncWrite + Unpin, +{ + let request = OwnedMessage::routing_activation_request( + ProtocolVersion::V2012, + source_address, + ActivationTypeCode::Default, + None, + ); + with_timeout("send routing activation", framed.send(&request)) + .await + .expect("send routing activation"); +} + +/// Send a diagnostic message carrying `user_data` over a raw socket. +async fn send_diagnostic_message( + framed: &mut FramedWrite, + source_address: LogicalAddress, + user_data: &[u8], +) where + W: AsyncWrite + Unpin, +{ + let request = OwnedMessage::diagnostic_message( + ProtocolVersion::V2012, + source_address, + SERVER_LOGICAL_ADDRESS, + user_data.to_vec(), + ); + with_timeout("send diagnostic message", framed.send(&request)) + .await + .expect("send diagnostic message"); } /// Test-only [`Connector`] that dials whatever address it's given, unlike @@ -464,14 +559,9 @@ impl ServerConnectionHandler for MisbehavingHandler { async fn diagnostic_message( &self, message: &DiagnosticMessage<'_>, - ) -> Result { - Ok(OwnedMessage::diagnostic_message_ack( - self.protocol_version(), - message.source_address, - message.target_address, - DiagnosticAckCode::RoutingConfirmationAck, - message.user_data.to_vec(), - )) + responses: &mut dyn ResponseWriter, + ) -> Result<(), Error> { + send_positive_ack(self, message, responses).await } } @@ -565,14 +655,9 @@ impl ServerConnectionHandler for DenyingRoutingHandler { async fn diagnostic_message( &self, message: &DiagnosticMessage<'_>, - ) -> Result { - Ok(OwnedMessage::diagnostic_message_ack( - self.protocol_version(), - message.source_address, - message.target_address, - DiagnosticAckCode::RoutingConfirmationAck, - message.user_data.to_vec(), - )) + responses: &mut dyn ResponseWriter, + ) -> Result<(), Error> { + send_positive_ack(self, message, responses).await } } @@ -666,14 +751,9 @@ impl ServerConnectionHandler for NackingRoutingHandler { async fn diagnostic_message( &self, message: &DiagnosticMessage<'_>, - ) -> Result { - Ok(OwnedMessage::diagnostic_message_ack( - self.protocol_version(), - message.source_address, - message.target_address, - DiagnosticAckCode::RoutingConfirmationAck, - message.user_data.to_vec(), - )) + responses: &mut dyn ResponseWriter, + ) -> Result<(), Error> { + send_positive_ack(self, message, responses).await } } @@ -819,7 +899,8 @@ impl ServerConnectionHandler for SilentOnDiagnosticHandler { async fn diagnostic_message( &self, _message: &DiagnosticMessage<'_>, - ) -> Result { + _responses: &mut dyn ResponseWriter, + ) -> Result<(), Error> { // Never resolves: the server received the message but never acknowledges it, // so the client must hit its own internal deadline rather than any // server-driven signal. @@ -887,3 +968,370 @@ async fn timed_out_diagnostic_message_ack_surfaces_timeout_not_connection_closed accept_loop.abort(); let _ = accept_loop.await; } + +/// A handler that answers every diagnostic message with a positive +/// `DiagnosticMessageAck` followed by a separate diagnostic-message response. +/// This is the shape `uds_on_ip` requires - it waits for the ack before it reads a +/// response, so a single-message server deadlocks it. +struct AckThenRespondHandler; + +#[async_trait] +impl ServerConnectionHandler for AckThenRespondHandler { + fn get_vin(&self) -> [u8; 17] { + [0x00; 17] + } + + fn get_logical_address(&self) -> LogicalAddress { + SERVER_LOGICAL_ADDRESS + } + + fn get_entity_id(&self) -> [u8; 6] { + [0x00; 6] + } + + fn get_group_id(&self) -> Option<[u8; 6]> { + None + } + + async fn routing_activation( + &self, + request: &RoutingActivationRequest, + ) -> Result { + Ok(OwnedMessage::routing_activation_response( + self.protocol_version(), + request.source_address, + self.get_logical_address(), + RoutingActivationResponseCode::RoutingSuccessfullyActivated, + [0; 4], + None, + )) + } + + async fn diagnostic_message( + &self, + message: &DiagnosticMessage<'_>, + responses: &mut dyn ResponseWriter, + ) -> Result<(), Error> { + send_positive_ack(self, message, responses).await?; + responses + .send(OwnedMessage::diagnostic_message( + self.protocol_version(), + self.get_logical_address(), + message.source_address, + vec![0x62, 0xFD, 0x69, 0xAA], + )) + .await?; + Ok(()) + } +} + +/// A single diagnostic request must be answerable with two messages on the wire: the +/// `DiagnosticMessageAck` ISO 13400 requires, then the UDS response itself. +#[tokio::test] +async fn handler_can_emit_ack_then_response() { + let (server_addr, accept_loop) = start_server_with(AckThenRespondHandler).await; + let mut stream = with_timeout("connect", TcpStream::connect(server_addr)) + .await + .expect("connect to test server"); + let (rx, tx) = stream.split(); + let mut reader = FramedRead::new(rx, MessageCodec::new()); + let mut writer = FramedWrite::new(tx, MessageCodec::new()); + + // Routing activation first, so the server accepts diagnostic messages. + send_routing_activation(&mut writer, CLIENT_LOGICAL_ADDRESS).await; + let _activation = read_message(&mut reader).await; + + send_diagnostic_message(&mut writer, CLIENT_LOGICAL_ADDRESS, &[0x22, 0xFD, 0x69]).await; + + let first = read_message(&mut reader).await; + match first.payload { + OwnedPayload::DiagnosticMessageAck(ref ack) => { + // Assert the code, not just the variant: the ack payload type is + // hardcoded positive regardless of the code (ARCHITECTURE §7.2), so + // a variant-only check would pass on a negative ack too and would + // depend on that bug staying exactly as it is. + assert_eq!(ack.ack_code, DiagnosticAckCode::RoutingConfirmationAck); + } + other => panic!("expected DiagnosticMessageAck first, got {other:?}"), + } + + let second = read_message(&mut reader).await; + match second.payload { + OwnedPayload::DiagnosticMessage(ref diag) => { + assert_eq!(diag.user_data, vec![0x62, 0xFD, 0x69, 0xAA]); + } + other => panic!("expected DiagnosticMessage second, got {other:?}"), + } + + accept_loop.abort(); + let _ = accept_loop.await; +} + +/// A caller must be able to hand the server a listener it bound itself, instead of being +/// forced onto `0.0.0.0:13400`. +#[tokio::test] +async fn run_server_with_listener_serves_a_caller_bound_socket() { + // A caller-bound listener is how the sim reaches port 13400 on a loopback + // alias, and how tests get an ephemeral port they can run in parallel on. + let listener = TcpListener::bind("127.0.0.1:0") + .await + .expect("bind ephemeral port"); + let addr = listener.local_addr().expect("read local addr"); + + let server = Server::new(AckThenRespondHandler).expect("construct server"); + let accept_loop = tokio::spawn(async move { + let _ = server.run_server_with_listener(listener).await; + }); + + let mut stream = with_timeout("connect", TcpStream::connect(addr)) + .await + .expect("connect to caller-bound server"); + let (rx, tx) = stream.split(); + let mut reader = FramedRead::new(rx, MessageCodec::new()); + let mut writer = FramedWrite::new(tx, MessageCodec::new()); + send_routing_activation(&mut writer, CLIENT_LOGICAL_ADDRESS).await; + let activation = read_message(&mut reader).await; + assert!(matches!( + activation.payload, + OwnedPayload::RoutingActivationResponse(_) + )); + + accept_loop.abort(); + let _ = accept_loop.await; +} + +/// The accept loop must survive clients that come and go without saying anything. +/// +/// Note the limit of this test: it exercises the loop's resilience across many +/// connections, NOT the `Err` branch of `accept()` itself. Provoking a real accept +/// failure means exhausting the process's file descriptors, which is not something a +/// test in this suite can do without destabilizing every other test in the binary. The +/// no-panic-on-accept-error change therefore remains unverified by automated test; this +/// covers only that the loop keeps serving after connections churn. +#[tokio::test] +async fn server_keeps_accepting_after_clients_disconnect_abruptly() { + let listener = TcpListener::bind("127.0.0.1:0") + .await + .expect("bind ephemeral port"); + let addr = listener.local_addr().expect("read local addr"); + let server = Server::new(AckThenRespondHandler).expect("construct server"); + let accept_loop = tokio::spawn(async move { + let _ = server.run_server_with_listener(listener).await; + }); + + // Connect and drop without saying anything, several times over. + for _ in 0..5 { + let stream = with_timeout("connect", TcpStream::connect(addr)) + .await + .expect("connect"); + drop(stream); + } + + // The entity must still serve a well-behaved client afterwards. + let mut stream = with_timeout("connect", TcpStream::connect(addr)) + .await + .expect("connect after churn"); + let (rx, tx) = stream.split(); + let mut reader = FramedRead::new(rx, MessageCodec::new()); + let mut writer = FramedWrite::new(tx, MessageCodec::new()); + send_routing_activation(&mut writer, CLIENT_LOGICAL_ADDRESS).await; + let activation = read_message(&mut reader).await; + assert!(matches!( + activation.payload, + OwnedPayload::RoutingActivationResponse(_) + )); + + accept_loop.abort(); + let _ = accept_loop.await; +} + +/// A handler that emits two NRC `0x78` "response pending" messages with a real delay +/// between them, then the final positive response. The delay is what distinguishes a +/// held pending wait from a burst: a client that mishandles P2* timing sees the gap, +/// whereas back-to-back writes hide it. +struct HeldPendingHandler; + +#[async_trait] +impl ServerConnectionHandler for HeldPendingHandler { + fn get_vin(&self) -> [u8; 17] { + [0x00; 17] + } + + fn get_logical_address(&self) -> LogicalAddress { + SERVER_LOGICAL_ADDRESS + } + + fn get_entity_id(&self) -> [u8; 6] { + [0x00; 6] + } + + fn get_group_id(&self) -> Option<[u8; 6]> { + None + } + + async fn routing_activation( + &self, + request: &RoutingActivationRequest, + ) -> Result { + Ok(OwnedMessage::routing_activation_response( + self.protocol_version(), + request.source_address, + self.get_logical_address(), + RoutingActivationResponseCode::RoutingSuccessfullyActivated, + [0; 4], + None, + )) + } + + async fn diagnostic_message( + &self, + message: &DiagnosticMessage<'_>, + responses: &mut dyn ResponseWriter, + ) -> Result<(), Error> { + send_positive_ack(self, message, responses).await?; + + for _ in 0..2 { + responses + .send(OwnedMessage::diagnostic_message( + self.protocol_version(), + self.get_logical_address(), + message.source_address, + // 0x7F 0x78 = requestCorrectlyReceived-ResponsePending + vec![0x7F, 0x22, 0x78], + )) + .await?; + tokio::time::sleep(Duration::from_millis(50)).await; + } + + responses + .send(OwnedMessage::diagnostic_message( + self.protocol_version(), + self.get_logical_address(), + message.source_address, + vec![0x62, 0xFD, 0x69, 0xAA], + )) + .await?; + Ok(()) + } +} + +/// Margin for the interleaving bounds below. The handler sleeps 50ms between sends; +/// 40ms leaves 10ms of slack for scheduling and socket jitter on a loaded CI box while +/// staying far away from the ~100ms a batched implementation would produce. +/// +/// If [`INTERLEAVING_MARGIN`] ever proves too tight under load, raise it - never delete +/// the assertions that use it, since they are the only thing distinguishing a streamed +/// response sequence from a batched one. +const INTERLEAVING_MARGIN: Duration = Duration::from_millis(40); + +/// A handler must be able to hold a pending wait open: emit an NRC `0x78`, await real +/// work, then emit more, with the tester observing each message as it is produced rather +/// than all of them at the end. +/// +/// This is the requirement that ruled out returning a `Vec` from +/// `diagnostic_message`, so the test is an executable guard on the [`ResponseWriter`] +/// sink design, not a red-green cycle - it is expected to pass as written, and to fail +/// loudly if the sink is ever replaced by a batched return. +/// +/// The property that discriminates the two designs is INTERLEAVING, not total duration. +/// A batched rewrite would keep this fixture's sleeps (they are handler logic, not sink +/// logic), push four messages into a `Vec` over the same ~100ms, and only then let the +/// server write them - so the *last* message still arrives at ~100ms either way. What +/// changes is when the *earlier* messages arrive: streamed, the ack is on the wire before +/// the handler's first sleep and the two pendings are 50ms apart; batched, all four land +/// together once the handler returns. Hence the two bounds below. +#[tokio::test] +async fn handler_holds_pending_wait_open_between_sends() { + let (server_addr, accept_loop) = start_server_with(HeldPendingHandler).await; + let mut stream = with_timeout("connect", TcpStream::connect(server_addr)) + .await + .expect("connect to test server"); + let (rx, tx) = stream.split(); + let mut reader = FramedRead::new(rx, MessageCodec::new()); + let mut writer = FramedWrite::new(tx, MessageCodec::new()); + + send_routing_activation(&mut writer, CLIENT_LOGICAL_ADDRESS).await; + let activation = read_message(&mut reader).await; + assert!( + matches!( + activation.payload, + OwnedPayload::RoutingActivationResponse(_) + ), + "routing activation must succeed before any diagnostic message is sent, otherwise \ + the failures below describe the wrong cause; got {:?}", + activation.payload + ); + + let started = std::time::Instant::now(); + send_diagnostic_message(&mut writer, CLIENT_LOGICAL_ADDRESS, &[0x22, 0xFD, 0x69]).await; + + let ack = read_message(&mut reader).await; + let ack_at = started.elapsed(); + match ack.payload { + // As above: the code, not just the variant. + OwnedPayload::DiagnosticMessageAck(ref ack) => { + assert_eq!(ack.ack_code, DiagnosticAckCode::RoutingConfirmationAck); + } + ref other => panic!("expected DiagnosticMessageAck first, got {other:?}"), + } + + let mut pending_at = Vec::new(); + for index in 0..2 { + let pending = read_message(&mut reader).await; + pending_at.push(started.elapsed()); + match pending.payload { + OwnedPayload::DiagnosticMessage(ref diag) => { + assert_eq!( + diag.user_data, + vec![0x7F, 0x22, 0x78], + "message {index} should be an NRC 0x78 pending" + ); + } + other => panic!("expected pending DiagnosticMessage, got {other:?}"), + } + } + + let final_response = read_message(&mut reader).await; + match final_response.payload { + OwnedPayload::DiagnosticMessage(ref diag) => { + assert_eq!(diag.user_data, vec![0x62, 0xFD, 0x69, 0xAA]); + } + other => panic!("expected final DiagnosticMessage, got {other:?}"), + } + + // Bound 1, and the one that actually catches the regression: the ack is written + // before the handler's first sleep, so it must arrive almost immediately. A batched + // return puts nothing on the socket until the handler returns ~100ms later, and this + // assertion goes red. + assert!( + ack_at < INTERLEAVING_MARGIN, + "the ack arrived {ack_at:?} after the request; a streamed sink delivers it before \ + the handler's first sleep, so anything near the handler's total runtime means \ + responses are being batched and flushed at the end" + ); + + // Bound 2: the two pendings are separated by the handler's 50ms sleep. Batched, they + // arrive in the same flush and the gap collapses to microseconds. + let pending_gap = pending_at[1] - pending_at[0]; + assert!( + pending_gap >= INTERLEAVING_MARGIN, + "the two pending responses arrived {pending_gap:?} apart (at {:?} and {:?}); the \ + handler sleeps 50ms between them, so a smaller gap means they were flushed \ + together rather than as the handler produced them", + pending_at[0], + pending_at[1] + ); + + // Secondary check: the handler's two 50ms waits really happened. This does NOT + // discriminate streaming from batching - a batched implementation takes just as long + // overall, because the sleeps are in the handler either way. It only guards against a + // fixture that quietly stops sleeping, which would make the two bounds above vacuous. + assert!( + started.elapsed() >= Duration::from_millis(100), + "responses arrived in {:?}; expected >=100ms of held pending waits", + started.elapsed() + ); + + accept_loop.abort(); + let _ = accept_loop.await; +} diff --git a/tests/udp_identification.rs b/tests/udp_identification.rs new file mode 100644 index 0000000..f5fdb45 --- /dev/null +++ b/tests/udp_identification.rs @@ -0,0 +1,215 @@ +//! The UDP half of the `DoIP` entity: answering vehicle-identification probes. +//! +//! Binds an ephemeral UDP port rather than the real +//! [`simple_doip::UDP_DISCOVERY_PORT`] so these tests run in parallel with +//! everything else, and so several of them can run at once without colliding. + +use async_trait::async_trait; +use simple_doip::{ + Error, LogicalAddress, + messages::{ + Decode, DiagnosticMessage, Message, OwnedMessage, Payload, RoutingActivationRequest, + RoutingActivationResponseCode, + }, + server::{ResponseWriter, Server, ServerConnectionHandler}, +}; +use std::time::Duration; +use tokio::net::UdpSocket; + +/// Generous but finite bound for every await in these tests, so a regression that +/// stops the responder answering fails the test quickly instead of hanging CI. +const TEST_TIMEOUT: Duration = Duration::from_secs(5); + +const SERVER_LOGICAL_ADDRESS: LogicalAddress = LogicalAddress(0x0001); +const TEST_VIN: [u8; 17] = *b"MVIS0000000000001"; +const TEST_ENTITY_ID: [u8; 6] = [0xAA, 0xBB, 0xCC, 0xDD, 0xEE, 0xFF]; + +/// Minimal handler: it supplies the identity fields the default +/// `received_vehicle_identification_request` reads, and stubs out the TCP-side +/// methods these tests never reach. +struct IdentityHandler; + +#[async_trait] +impl ServerConnectionHandler for IdentityHandler { + fn get_vin(&self) -> [u8; 17] { + TEST_VIN + } + fn get_logical_address(&self) -> LogicalAddress { + SERVER_LOGICAL_ADDRESS + } + fn get_entity_id(&self) -> [u8; 6] { + TEST_ENTITY_ID + } + fn get_group_id(&self) -> Option<[u8; 6]> { + None + } + + async fn routing_activation( + &self, + request: &RoutingActivationRequest, + ) -> Result { + Ok(OwnedMessage::routing_activation_response( + self.protocol_version(), + request.source_address, + self.get_logical_address(), + RoutingActivationResponseCode::RoutingSuccessfullyActivated, + [0; 4], + None, + )) + } + + async fn diagnostic_message( + &self, + _message: &DiagnosticMessage<'_>, + _responses: &mut dyn ResponseWriter, + ) -> Result<(), Error> { + Ok(()) + } +} + +/// A `VehicleIdentificationRequest` on the wire: 8-byte header, no payload. +/// Protocol version 0x02 with its inverse 0xFD, payload type 0x0001, length 0. +const VEHICLE_IDENTIFICATION_REQUEST: [u8; 8] = [0x02, 0xFD, 0x00, 0x01, 0x00, 0x00, 0x00, 0x00]; + +/// Too short to be a `DoIP` generic header, so `Message::decode` rejects it. Any +/// host on the network can send this, so the responder must log and carry on. +const TRUNCATED_DATAGRAM: [u8; 3] = [0x02, 0xFD, 0x00]; + +/// A well-formed `AliveCheckRequest` (payload type 0x0007, length 0). Decodes +/// cleanly but is not something this responder answers, exercising the +/// wrong-payload-type skip rather than the decode-failure one. +const ALIVE_CHECK_REQUEST: [u8; 8] = [0x02, 0xFD, 0x00, 0x07, 0x00, 0x00, 0x00, 0x00]; + +/// `VehicleIdentificationRequestWithEID` (0x0002), naming an entity that is not +/// this one: 8-byte header plus a 6-byte EID. +const IDENTIFICATION_REQUEST_WITH_EID: [u8; 14] = [ + 0x02, 0xFD, 0x00, 0x02, 0x00, 0x00, 0x00, 0x06, 0x11, 0x22, 0x33, 0x44, 0x55, 0x66, +]; + +/// `VehicleIdentificationRequestWithVIN` (0x0003), naming a VIN that is not this +/// entity's: 8-byte header plus a 17-byte VIN. +const IDENTIFICATION_REQUEST_WITH_VIN: [u8; 25] = [ + 0x02, 0xFD, 0x00, 0x03, 0x00, 0x00, 0x00, 0x11, b'O', b'T', b'H', b'E', b'R', b'0', b'0', b'0', + b'0', b'0', b'0', b'0', b'0', b'0', b'0', b'0', b'1', +]; + +/// How long to wait before concluding the responder deliberately sent nothing. +/// A reply travels loopback in microseconds, so this is generous while keeping a +/// negative assertion cheap. +const SILENCE_WINDOW: Duration = Duration::from_millis(250); + +/// Bind an ephemeral responder socket, start [`Server::run_udp_responder`] on it, +/// and return the address a probing client should send to. +async fn start_udp_responder() -> std::net::SocketAddr { + let server_socket = UdpSocket::bind("127.0.0.1:0").await.expect("bind server"); + let server_addr = server_socket.local_addr().expect("server addr"); + + let server = Server::new(IdentityHandler).expect("construct server"); + tokio::spawn(async move { + let _ = server.run_udp_responder(server_socket).await; + }); + + server_addr +} + +/// Await one datagram on `client` and assert it is this entity's identification +/// response. +async fn expect_identification_response(client: &UdpSocket) { + let mut buf = [0u8; 256]; + let (len, _from) = tokio::time::timeout(TEST_TIMEOUT, client.recv_from(&mut buf)) + .await + .expect("timed out waiting for identification response") + .expect("recv identification response"); + + // `Decode` is implemented for the BORROWED `Message<'a>`, not `OwnedMessage`, + // and returns (message, remaining_bytes) — not a bare message. + let (message, _rest) = Message::decode(&buf[..len]).expect("decode identification response"); + // ISO 13400-2 has a single wire payload type (0x0004) for both the + // unsolicited announcement and the directed reply, so `Payload::decode` + // always yields `VehicleAnnouncement` — never the `VehicleIdentificationResponse` + // variant the responder constructs. Both encode identically. + match message.payload { + Payload::VehicleAnnouncement(response) => { + assert_eq!(response.vin, TEST_VIN); + assert_eq!(response.entity_id, TEST_ENTITY_ID); + assert_eq!(response.logical_address, SERVER_LOGICAL_ADDRESS); + } + other => panic!("expected a vehicle identification response, got {other:?}"), + } +} + +#[tokio::test] +async fn udp_responder_answers_a_vehicle_identification_request() { + let server_addr = start_udp_responder().await; + + let client = UdpSocket::bind("127.0.0.1:0").await.expect("bind client"); + client + .send_to(&VEHICLE_IDENTIFICATION_REQUEST, server_addr) + .await + .expect("send identification request"); + + expect_identification_response(&client).await; +} + +/// Datagrams the responder cannot answer - undecodable ones and well-formed ones +/// of the wrong payload type - must not take it down. Anyone on the network can +/// reach a UDP responder, so a single bad probe killing the loop would let a +/// stray packet make the entity permanently undiscoverable. +#[tokio::test] +async fn udp_responder_keeps_serving_after_datagrams_it_cannot_answer() { + let server_addr = start_udp_responder().await; + + let client = UdpSocket::bind("127.0.0.1:0").await.expect("bind client"); + client + .send_to(&TRUNCATED_DATAGRAM, server_addr) + .await + .expect("send truncated datagram"); + client + .send_to(&ALIVE_CHECK_REQUEST, server_addr) + .await + .expect("send alive check request"); + client + .send_to(&VEHICLE_IDENTIFICATION_REQUEST, server_addr) + .await + .expect("send identification request"); + + // The good probe still gets answered, so the bad ones were skipped rather than + // fatal. This also proves the responder sent nothing for either: the read below + // takes the *next* datagram on the socket, so a spurious reply would arrive + // first and fail its assertions. + expect_identification_response(&client).await; +} + +/// The directed request forms name an entity by EID or VIN, and `Payload::decode` +/// discards those bytes, so the responder cannot tell whether it is the addressee. +/// It must stay silent rather than claim an identity that may have been meant for +/// someone else - a wrong answer misleads a tester, while silence degrades to a +/// discovery timeout testers already handle. +#[tokio::test] +async fn udp_responder_stays_silent_for_directed_eid_and_vin_requests() { + let server_addr = start_udp_responder().await; + + let client = UdpSocket::bind("127.0.0.1:0").await.expect("bind client"); + client + .send_to(&IDENTIFICATION_REQUEST_WITH_EID, server_addr) + .await + .expect("send identification request with EID"); + client + .send_to(&IDENTIFICATION_REQUEST_WITH_VIN, server_addr) + .await + .expect("send identification request with VIN"); + + let mut buf = [0u8; 256]; + let unexpected = tokio::time::timeout(SILENCE_WINDOW, client.recv_from(&mut buf)).await; + assert!( + unexpected.is_err(), + "responder answered a directed request it cannot match: {unexpected:?}" + ); + + // Silence must mean "declined", not "died", so a plain probe still works. + client + .send_to(&VEHICLE_IDENTIFICATION_REQUEST, server_addr) + .await + .expect("send identification request"); + expect_identification_response(&client).await; +}