From 3b497b9e21dfc6a588644de1dcc2dc243544cd03 Mon Sep 17 00:00:00 2001 From: Justin Kovacich Date: Wed, 12 Aug 2026 16:09:53 -0400 Subject: [PATCH 01/19] feat(server)!: write diagnostic responses into a sink instead of returning one MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ServerConnectionHandler::diagnostic_message returned exactly one OwnedMessage, so a handler could never emit the DiagnosticMessageAck that ISO 13400 requires before a UDS response — uds_on_ip blocks waiting for that ack, which made this server unusable for driving a real UdsClient. Pass a &mut dyn ResponseWriter instead. Each send goes straight to the socket, so a handler can also await between writes and hold an NRC 0x78 pending wait open rather than emitting a burst. BREAKING CHANGE: diagnostic_message takes a responses writer and returns Result<(), Error>. Co-Authored-By: Claude Opus 5 --- examples/echo_server.rs | 40 +++--- src/server.rs | 79 +++++++++-- tests/integration_test.rs | 276 ++++++++++++++++++++++++++++++++------ 3 files changed, 324 insertions(+), 71 deletions(-) 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/server.rs b/src/server.rs index d5fe6de..f7bdbd7 100644 --- a/src/server.rs +++ b/src/server.rs @@ -65,6 +65,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 +135,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 @@ -280,7 +321,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 +343,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 +371,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..71995a4 100644 --- a/tests/integration_test.rs +++ b/tests/integration_test.rs @@ -15,15 +15,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 +36,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. @@ -105,15 +108,18 @@ 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(), - )) + responses + .send(OwnedMessage::diagnostic_message_ack( + self.protocol_version(), + message.source_address, + message.target_address, + DiagnosticAckCode::RoutingConfirmationAck, + message.user_data.to_vec(), + )) + .await } } @@ -160,8 +166,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 +208,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 +543,17 @@ 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> { + responses + .send(OwnedMessage::diagnostic_message_ack( + self.protocol_version(), + message.source_address, + message.target_address, + DiagnosticAckCode::RoutingConfirmationAck, + message.user_data.to_vec(), + )) + .await } } @@ -565,14 +647,17 @@ 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> { + responses + .send(OwnedMessage::diagnostic_message_ack( + self.protocol_version(), + message.source_address, + message.target_address, + DiagnosticAckCode::RoutingConfirmationAck, + message.user_data.to_vec(), + )) + .await } } @@ -666,14 +751,17 @@ 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> { + responses + .send(OwnedMessage::diagnostic_message_ack( + self.protocol_version(), + message.source_address, + message.target_address, + DiagnosticAckCode::RoutingConfirmationAck, + message.user_data.to_vec(), + )) + .await } } @@ -819,7 +907,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 +976,104 @@ 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> { + responses + .send(OwnedMessage::diagnostic_message_ack( + self.protocol_version(), + self.get_logical_address(), + message.source_address, + DiagnosticAckCode::RoutingConfirmationAck, + Vec::new(), + )) + .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; + assert!( + matches!(first.payload, OwnedPayload::DiagnosticMessageAck(_)), + "expected DiagnosticMessageAck first, got {:?}", + first.payload + ); + + 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; +} From 500c8d30f7d8f07475c0ec05bd400e84d4fe4b84 Mon Sep 17 00:00:00 2001 From: Justin Kovacich Date: Wed, 12 Aug 2026 16:10:24 -0400 Subject: [PATCH 02/19] docs: retire the single-response handler limitation README's known-limitations list and ARCHITECTURE section 7.1 both described ServerConnectionHandler::diagnostic_message as unable to emit an ack followed by a response. The preceding commit gave it a ResponseWriter, so both entries now describe behavior the crate no longer has. Co-Authored-By: Claude Opus 5 --- ARCHITECTURE.md | 40 +++++++++++++++++----------------------- README.md | 5 ----- 2 files changed, 17 insertions(+), 28 deletions(-) diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 323ae9b..1ef0884 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -380,39 +380,33 @@ 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. -### 7.1 `ServerConnectionHandler::diagnostic_message` cannot express correct DoIP behavior +### 7.1 `ServerConnectionHandler::diagnostic_message` — RESOLVED -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 diff --git a/README.md b/README.md index 980d890..b72d154 100644 --- a/README.md +++ b/README.md @@ -29,11 +29,6 @@ gaps a new integrator should know about before relying on them: 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. - **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 From 2b2bead5739211faa3cf1b588a5efd1141393776 Mon Sep 17 00:00:00 2001 From: Justin Kovacich Date: Wed, 12 Aug 2026 16:19:56 -0400 Subject: [PATCH 03/19] test(server): prove a handler can hold an NRC 0x78 pending wait open Guards the design decision behind ResponseWriter: a Vec return would drain back-to-back and make the elapsed-time assertion here impossible, so this test fails if the sink is ever replaced by a batch return. Also folds the six near-identical positive-ack bodies in this file into one `send_positive_ack` helper, so a new handler no longer re-derives the ack. The helper addresses the ack from the entity to the requesting tester, which is the direction ISO 13400 specifies and the one the multi-response handlers already used; the four handlers that had the addresses reversed now match. Nothing asserts on the ack's addresses and the client ignores them, so behavior is unchanged. Co-Authored-By: Claude Opus 5 --- tests/integration_test.rs | 202 +++++++++++++++++++++++++++++--------- 1 file changed, 157 insertions(+), 45 deletions(-) diff --git a/tests/integration_test.rs b/tests/integration_test.rs index 71995a4..db8b518 100644 --- a/tests/integration_test.rs +++ b/tests/integration_test.rs @@ -62,6 +62,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, @@ -111,15 +133,7 @@ impl ServerConnectionHandler for TestHandler { responses: &mut dyn ResponseWriter, ) -> Result<(), Error> { *self.last_diagnostic_payload.lock().unwrap() = Some(message.user_data.to_vec()); - responses - .send(OwnedMessage::diagnostic_message_ack( - self.protocol_version(), - message.source_address, - message.target_address, - DiagnosticAckCode::RoutingConfirmationAck, - message.user_data.to_vec(), - )) - .await + send_positive_ack(self, message, responses).await } } @@ -545,15 +559,7 @@ impl ServerConnectionHandler for MisbehavingHandler { message: &DiagnosticMessage<'_>, responses: &mut dyn ResponseWriter, ) -> Result<(), Error> { - responses - .send(OwnedMessage::diagnostic_message_ack( - self.protocol_version(), - message.source_address, - message.target_address, - DiagnosticAckCode::RoutingConfirmationAck, - message.user_data.to_vec(), - )) - .await + send_positive_ack(self, message, responses).await } } @@ -649,15 +655,7 @@ impl ServerConnectionHandler for DenyingRoutingHandler { message: &DiagnosticMessage<'_>, responses: &mut dyn ResponseWriter, ) -> Result<(), Error> { - responses - .send(OwnedMessage::diagnostic_message_ack( - self.protocol_version(), - message.source_address, - message.target_address, - DiagnosticAckCode::RoutingConfirmationAck, - message.user_data.to_vec(), - )) - .await + send_positive_ack(self, message, responses).await } } @@ -753,15 +751,7 @@ impl ServerConnectionHandler for NackingRoutingHandler { message: &DiagnosticMessage<'_>, responses: &mut dyn ResponseWriter, ) -> Result<(), Error> { - responses - .send(OwnedMessage::diagnostic_message_ack( - self.protocol_version(), - message.source_address, - message.target_address, - DiagnosticAckCode::RoutingConfirmationAck, - message.user_data.to_vec(), - )) - .await + send_positive_ack(self, message, responses).await } } @@ -1020,15 +1010,7 @@ impl ServerConnectionHandler for AckThenRespondHandler { message: &DiagnosticMessage<'_>, responses: &mut dyn ResponseWriter, ) -> Result<(), Error> { - responses - .send(OwnedMessage::diagnostic_message_ack( - self.protocol_version(), - self.get_logical_address(), - message.source_address, - DiagnosticAckCode::RoutingConfirmationAck, - Vec::new(), - )) - .await?; + send_positive_ack(self, message, responses).await?; responses .send(OwnedMessage::diagnostic_message( self.protocol_version(), @@ -1077,3 +1059,133 @@ async fn handler_can_emit_ack_then_response() { 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(()) + } +} + +/// 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 the delay on the wire. +/// +/// This is the requirement that ruled out returning a `Vec` from +/// `diagnostic_message`: a batch return drains back-to-back once the handler has already +/// finished, so the elapsed-time assertion below would be unsatisfiable. The test is +/// therefore 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. +#[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; + + 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; + assert!(matches!(ack.payload, OwnedPayload::DiagnosticMessageAck(_))); + + for index in 0..2 { + let pending = read_message(&mut reader).await; + 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:?}"), + } + + // Two 50ms sleeps must actually have elapsed on the wire. If the sink buffered + // everything and flushed at the end, this is near-zero and the held-pending property + // does not hold. + assert!( + started.elapsed() >= Duration::from_millis(100), + "responses arrived in {:?}; expected >=100ms of held pending waits", + started.elapsed() + ); +} From 900b99403e797d7b2b1eb440cb54e521348569f5 Mon Sep 17 00:00:00 2001 From: Justin Kovacich Date: Wed, 12 Aug 2026 16:36:38 -0400 Subject: [PATCH 04/19] test(server): assert response interleaving, not just total elapsed time The >=100ms bound could not fail for the regression it guarded. The two 50ms sleeps are handler logic, so a batched Vec rewrite keeps them: the handler still takes ~100ms, the four messages are just written together at the end, and the final one still lands at ~100ms. Measured under a simulated batch: total 102.77ms, i.e. green. What separates the designs is interleaving, so assert on that. The ack must arrive before the handler's first sleep (<40ms; 50us streamed, 102.75ms batched), and the two pending responses must be >=40ms apart (51.3ms streamed, 4.3us batched). The total-time check stays as a guard against a fixture that stops sleeping, re-commented to say it does not discriminate the two designs. Both bounds are stated against INTERLEAVING_MARGIN, documented as raise-never- delete if it proves tight under CI load. Disabling Nagle on the accepted socket is required to observe any of this: ConnectorSocket sets TCP_NODELAY for the client but nothing sets it for an accepted connection, so consecutive small responses stalled ~40ms on the tester's delayed ACK - the pending gap measured 7.3ms through a genuinely streamed sink. The fixture sets it; whether run_server should is left open. Also assert the routing activation response arrives before the diagnostic exchange, so a broken activation reports itself instead of surfacing as a missing ack, and abort the accept loop like the neighboring test does. Co-Authored-By: Claude Opus 5 --- tests/integration_test.rs | 87 ++++++++++++++++++++++++++++++++++----- 1 file changed, 76 insertions(+), 11 deletions(-) diff --git a/tests/integration_test.rs b/tests/integration_test.rs index db8b518..b04d10c 100644 --- a/tests/integration_test.rs +++ b/tests/integration_test.rs @@ -215,6 +215,16 @@ where let Ok((stream, peer_addr)) = listener.accept().await else { break; }; + // Disable Nagle on the server side. `ConnectorSocket` already does this for + // the client (`src/connection.rs`), but nothing does it for an accepted + // connection, so consecutive small responses stall on the tester's delayed + // ACK - measured at ~40ms between two 5-byte writes on loopback. That is a + // transport artifact with nothing to say about handler behavior, and it + // would otherwise swamp the sub-50ms timings + // `handler_holds_pending_wait_open_between_sends` asserts on. Setting it here + // rather than in `src/` keeps this a test-only change; whether `run_server` + // itself should set it is a separate question about live P2 timing. + let _ = stream.set_nodelay(true); // Await the connection inline, sequentially, matching `run_server`'s // control flow. `run_server` logs a handler error and keeps accepting; // mirror that by discarding the error here. @@ -1130,18 +1140,34 @@ impl ServerConnectionHandler for HeldPendingHandler { } } +/// 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 the delay on the wire. +/// 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`: a batch return drains back-to-back once the handler has already -/// finished, so the elapsed-time assertion below would be unsatisfiable. The test is -/// therefore 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. +/// `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 (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"); @@ -1150,16 +1176,28 @@ async fn handler_holds_pending_wait_open_between_sends() { let mut writer = FramedWrite::new(tx, MessageCodec::new()); send_routing_activation(&mut writer, CLIENT_LOGICAL_ADDRESS).await; - let _activation = read_message(&mut reader).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(); assert!(matches!(ack.payload, OwnedPayload::DiagnosticMessageAck(_))); + 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!( @@ -1180,12 +1218,39 @@ async fn handler_holds_pending_wait_open_between_sends() { other => panic!("expected final DiagnosticMessage, got {other:?}"), } - // Two 50ms sleeps must actually have elapsed on the wire. If the sink buffered - // everything and flushed at the end, this is near-zero and the held-pending property - // does not hold. + // 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; } From 13945e82057eb5264e475149f5d004b93dc41ce5 Mon Sep 17 00:00:00 2001 From: Justin Kovacich Date: Wed, 12 Aug 2026 16:47:07 -0400 Subject: [PATCH 05/19] feat(server): caller-bound listener, TCP_NODELAY, no panic on accept errors run_server hardcoded bind("0.0.0.0", 13400), so an entity could not be placed on a loopback alias and tests could not use an ephemeral port in parallel. run_server_with_listener takes a listener the caller already bound; run_server delegates to it and is unchanged for existing callers. The accept loop also panicked on any accept error. Transient conditions must not take the entity down, so log and continue. Accepted sockets never had TCP_NODELAY set, though ConnectorSocket sets it client-side. Consecutive small frames - an ack then a response, or successive NRC 0x78 pendings - waited on the peer's delayed ACK, costing up to ~40ms of P2 budget per exchange (measured 43.6ms in the response-pending test). Set it on every accepted connection and drop the test fixture's workaround, so the interleaving guard measures shipped behavior. Co-Authored-By: Claude Opus 5 --- src/server.rs | 42 ++++++++++++++++---- tests/integration_test.rs | 82 ++++++++++++++++++++++++++++++++++----- 2 files changed, 107 insertions(+), 17 deletions(-) diff --git a/src/server.rs b/src/server.rs index f7bdbd7..c9d355a 100644 --- a/src/server.rs +++ b/src/server.rs @@ -274,17 +274,29 @@ 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 - /// - /// # Panics - /// Panics if accepting a new TCP client connection fails + /// Returns an [`Error`] if the TCP listener cannot be bound. pub async fn run_server(&self) -> Result<(), Error> { // TODO: Vehicle Announcement over UDP 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. + /// + /// # Errors + /// Returns an [`Error`] only if a connection handler fails fatally; accept + /// errors are logged and the loop continues. + 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)) => { @@ -296,8 +308,11 @@ where } } Err(accept_error) => { - // TODO: Don't panic here, this might happen - panic!("Failed to accept new TCP client: {accept_error}"); + // Transient conditions (EMFILE, a peer resetting between + // the SYN and our accept) must not take the entity down — + // a simulator that aborts here turns a client bug into an + // opaque transport failure. + error!("Failed to accept TCP client, continuing: {accept_error}"); } } } @@ -313,6 +328,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()); diff --git a/tests/integration_test.rs b/tests/integration_test.rs index b04d10c..d16dde3 100644 --- a/tests/integration_test.rs +++ b/tests/integration_test.rs @@ -215,16 +215,6 @@ where let Ok((stream, peer_addr)) = listener.accept().await else { break; }; - // Disable Nagle on the server side. `ConnectorSocket` already does this for - // the client (`src/connection.rs`), but nothing does it for an accepted - // connection, so consecutive small responses stall on the tester's delayed - // ACK - measured at ~40ms between two 5-byte writes on loopback. That is a - // transport artifact with nothing to say about handler behavior, and it - // would otherwise swamp the sub-50ms timings - // `handler_holds_pending_wait_open_between_sends` asserts on. Setting it here - // rather than in `src/` keeps this a test-only change; whether `run_server` - // itself should set it is a separate question about live P2 timing. - let _ = stream.set_nodelay(true); // Await the connection inline, sequentially, matching `run_server`'s // control flow. `run_server` logs a handler error and keeps accepting; // mirror that by discarding the error here. @@ -1070,6 +1060,78 @@ async fn handler_can_emit_ack_then_response() { 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 _task = 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(_) + )); +} + +/// 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 _task = 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(_) + )); +} + /// 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, From 66bfa2652656260c7536d978bafbe4aef02b440b Mon Sep 17 00:00:00 2001 From: Justin Kovacich Date: Wed, 12 Aug 2026 16:53:19 -0400 Subject: [PATCH 06/19] docs(server): correct run_server_with_listener's # Errors section The section promised an Err on fatal handler failure, but the loop logs every error and never breaks, so the future never resolves and a caller would build recovery logic on a dead branch. Say so plainly, and record why the Result is kept: run_server needs a matching return type to propagate its bind failure, and a future shutdown path needs somewhere to report one. Also note on handle_client_connection that it sets TCP_NODELAY, since it is public and overrides a caller-configured stream, and align the two new accept loop tests with the file's convention of aborting the spawned server task. Co-Authored-By: Claude Opus 5 --- src/server.rs | 11 +++++++++-- tests/integration_test.rs | 22 +++++++++++++++------- 2 files changed, 24 insertions(+), 9 deletions(-) diff --git a/src/server.rs b/src/server.rs index c9d355a..082a055 100644 --- a/src/server.rs +++ b/src/server.rs @@ -294,8 +294,11 @@ where /// [`TcpListener::local_addr`] before calling this. /// /// # Errors - /// Returns an [`Error`] only if a connection handler fails fatally; accept - /// errors are logged and the loop continues. + /// 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 { @@ -320,6 +323,10 @@ where /// 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( diff --git a/tests/integration_test.rs b/tests/integration_test.rs index d16dde3..a3bda50 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`]. @@ -1072,7 +1074,7 @@ async fn run_server_with_listener_serves_a_caller_bound_socket() { let addr = listener.local_addr().expect("read local addr"); let server = Server::new(AckThenRespondHandler).expect("construct server"); - let _task = tokio::spawn(async move { + let accept_loop = tokio::spawn(async move { let _ = server.run_server_with_listener(listener).await; }); @@ -1088,6 +1090,9 @@ async fn run_server_with_listener_serves_a_caller_bound_socket() { activation.payload, OwnedPayload::RoutingActivationResponse(_) )); + + accept_loop.abort(); + let _ = accept_loop.await; } /// The accept loop must survive clients that come and go without saying anything. @@ -1105,7 +1110,7 @@ async fn server_keeps_accepting_after_clients_disconnect_abruptly() { .expect("bind ephemeral port"); let addr = listener.local_addr().expect("read local addr"); let server = Server::new(AckThenRespondHandler).expect("construct server"); - let _task = tokio::spawn(async move { + let accept_loop = tokio::spawn(async move { let _ = server.run_server_with_listener(listener).await; }); @@ -1130,6 +1135,9 @@ async fn server_keeps_accepting_after_clients_disconnect_abruptly() { 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 From b4b1645949597e379031ad63cc2c727bc3cd55a6 Mon Sep 17 00:00:00 2001 From: Justin Kovacich Date: Wed, 12 Aug 2026 17:00:54 -0400 Subject: [PATCH 07/19] feat(server): answer UDP vehicle-identification probes run_server carried a "TODO: Vehicle Announcement over UDP" and the TCP VehicleIdentificationRequest arm only warned, so a DoIP entity built on this crate was invisible to any UDP discovery probe. run_udp_responder serves a caller-bound UdpSocket, delegating the response content to the existing received_vehicle_identification_request hook. Malformed datagrams are logged and skipped. The reply is stamped with PayloadType::VehicleAnnouncement (0x0004): ISO 13400-2 defines a single wire payload type for both the unsolicited announcement and the directed reply, and there is no PayloadType::VehicleIdentificationResponse. Co-Authored-By: Claude Opus 5 --- Cargo.toml | 4 + src/messages/mod.rs | 31 ++++++++ src/server.rs | 89 +++++++++++++++++++-- tests/udp_identification.rs | 152 ++++++++++++++++++++++++++++++++++++ 4 files changed, 271 insertions(+), 5 deletions(-) create mode 100644 tests/udp_identification.rs diff --git a/Cargo.toml b/Cargo.toml index 2a3dfc2..410c274 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -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/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 082a055..f53e254 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, ProtocolVersion, RoutingActivationRequest, + VehicleIdentificationResponse, VinGidSyncStatus, }, }; use async_trait::async_trait; @@ -24,7 +24,7 @@ use std::{ atomic::{AtomicUsize, Ordering}, }, }; -use tokio::net::{TcpListener, TcpStream}; +use tokio::net::{TcpListener, TcpStream, UdpSocket}; use tokio_util::codec::{FramedRead, FramedWrite}; use tracing::{error, warn}; @@ -280,7 +280,9 @@ where /// # 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 @@ -321,6 +323,83 @@ where } } + /// 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 response content + /// comes from + /// [`ServerConnectionHandler::received_vehicle_identification_request`], so an + /// implementation customizes what it announces without reimplementing the + /// datagram loop. + /// + /// Runs until the socket read fails. Datagrams that cannot be decoded, and + /// decodable ones carrying anything other than a vehicle-identification + /// request, are logged and skipped: a UDP socket is reachable by every host + /// on the network, so one stray or hostile packet must not stop the entity + /// answering good probes. + /// + /// # Errors + /// Returns an [`Error`] if reading from or writing to the socket fails, or if + /// the handler fails to build an identification response. + 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, + // and arrives truncated - which the decode below then rejects. + let mut buf = [0u8; 1024]; + loop { + let (len, peer) = socket.recv_from(&mut buf).await?; + + // `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) { + warn!( + "Unsupported UDP payload type {:?} from {peer}, ignoring", + 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), + }; + let response = self + .connection_handler + .received_vehicle_identification_request(&client_info)?; + 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 = std::vec::Vec::with_capacity(reply.encoded_size()?); + reply.encode(&mut encoded)?; + socket.send_to(&encoded, peer).await?; + } + } + /// Handle an individual client TCP connection, reading and responding to messages /// /// Sets `TCP_NODELAY` on `tcp_stream`, overriding the caller's setting if it diff --git a/tests/udp_identification.rs b/tests/udp_identification.rs new file mode 100644 index 0000000..1e5fecf --- /dev/null +++ b/tests/udp_identification.rs @@ -0,0 +1,152 @@ +//! 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]; + +/// 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; +} + +/// A malformed datagram must not take the responder 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_survives_a_malformed_datagram() { + 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(&VEHICLE_IDENTIFICATION_REQUEST, server_addr) + .await + .expect("send identification request"); + + // The good probe still gets answered, so the bad one was skipped rather than + // fatal. It also proves the responder sent nothing for the bad datagram: this + // read would otherwise pick up that spurious reply and fail to decode it. + expect_identification_response(&client).await; +} From 9ae2170f2b7232a26374a000183e63cb71fdcb03 Mon Sep 17 00:00:00 2001 From: Justin Kovacich Date: Wed, 12 Aug 2026 17:01:33 -0400 Subject: [PATCH 08/19] docs(server): disclose that run_udp_responder cannot filter EID/VIN requests Payload::decode collapses the plain, with-EID, and with-VIN request forms into one variant and drops the EID/VIN bytes, so the responder answers a directed request that named a different entity and never consults the vehicle_identification_with_eid / _with_vin hooks. Reading run_udp_responder next to those hooks otherwise suggests they are wired up. Co-Authored-By: Claude Opus 5 --- src/server.rs | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/src/server.rs b/src/server.rs index f53e254..96e04b0 100644 --- a/src/server.rs +++ b/src/server.rs @@ -346,6 +346,17 @@ where /// on the network, so one stray or hostile packet must not stop the entity /// answering good probes. /// + /// # Known limitation + /// [`Payload::decode`] collapses all three request forms - plain (0x0001), + /// with-EID (0x0002), and with-VIN (0x0003) - into + /// [`Payload::VehicleIdentificationRequest`], discarding the EID or VIN the + /// latter two carry. This responder therefore answers a directed request even + /// when it named a different entity, and the + /// [`ServerConnectionHandler::vehicle_identification_with_eid`] and + /// [`ServerConnectionHandler::vehicle_identification_with_vin`] hooks are + /// never consulted. Filtering them correctly needs the payload preserved + /// through decoding, which is a change to [`Payload`]. + /// /// # Errors /// Returns an [`Error`] if reading from or writing to the socket fails, or if /// the handler fails to build an identification response. From f6678119c4a1dbabb4b0cd051157cd483ad07898 Mon Sep 17 00:00:00 2001 From: Justin Kovacich Date: Wed, 12 Aug 2026 17:10:27 -0400 Subject: [PATCH 09/19] fix(server): make every UDP responder failure non-fatal, and decline directed probes Only decode failures were skipped; socket and handler errors returned. That let any host on the network end discovery for the life of the process: on Windows an oversized datagram fails recvfrom with WSAEMSGSIZE rather than truncating, so a 2 KB packet killed the loop. A transient send failure (ENETUNREACH, firewall EPERM) or a handler that could not yet read its identity out of NVM did the same. Every failure in the loop is now logged and skipped, matching the TCP accept loop, so run_udp_responder no longer returns. The responder also answered VehicleIdentificationRequestWithEID/WithVIN probes unconditionally, because Payload::decode collapses all three request forms and drops the EID/VIN bytes. Answering a probe addressed to another entity actively misleads a tester, while staying silent degrades to a discovery timeout testers already handle, so the header payload type is now checked and the directed forms are declined. The known-limitation note states that trade-off instead of implying a Payload change was the only option. Co-Authored-By: Claude Opus 5 --- src/server.rs | 105 +++++++++++++++++++++++++++--------- tests/udp_identification.rs | 75 +++++++++++++++++++++++--- 2 files changed, 150 insertions(+), 30 deletions(-) diff --git a/src/server.rs b/src/server.rs index 96e04b0..bbe07a1 100644 --- a/src/server.rs +++ b/src/server.rs @@ -10,8 +10,8 @@ use crate::{ message_codec::MessageCodec, messages::{ Decode, DiagnosticMessage, DiagnosticPowerModeCode, Encode, FurtherActionRequired, Message, - OwnedMessage, OwnedPayload, Payload, ProtocolVersion, RoutingActivationRequest, - VehicleIdentificationResponse, VinGidSyncStatus, + OwnedMessage, OwnedPayload, Payload, PayloadType, ProtocolVersion, + RoutingActivationRequest, VehicleIdentificationResponse, VinGidSyncStatus, }, }; use async_trait::async_trait; @@ -340,33 +340,50 @@ where /// implementation customizes what it announces without reimplementing the /// datagram loop. /// - /// Runs until the socket read fails. Datagrams that cannot be decoded, and - /// decodable ones carrying anything other than a vehicle-identification - /// request, are logged and skipped: a UDP socket is reachable by every host - /// on the network, so one stray or hostile packet must not stop the entity - /// answering good probes. + /// 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 - /// [`Payload::decode`] collapses all three request forms - plain (0x0001), - /// with-EID (0x0002), and with-VIN (0x0003) - into - /// [`Payload::VehicleIdentificationRequest`], discarding the EID or VIN the - /// latter two carry. This responder therefore answers a directed request even - /// when it named a different entity, and the + /// 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. Filtering them correctly needs the payload preserved - /// through decoding, which is a change to [`Payload`]. + /// never consulted. Answering the directed forms requires [`Payload`] to + /// preserve the EID/VIN through decoding. /// /// # Errors - /// Returns an [`Error`] if reading from or writing to the socket fails, or if - /// the handler fails to build an identification response. + /// 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, - // and arrives truncated - which the decode below then rejects. + // this is generous. Anything longer is not a message this loop answers. let mut buf = [0u8; 1024]; loop { - let (len, peer) = socket.recv_from(&mut buf).await?; + // 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}"); + continue; + } + }; // `Decode` is implemented for the borrowed `Message<'a>` and yields // (message, remaining_bytes). The borrow of `buf` ends with this @@ -387,6 +404,23 @@ where 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 + ) { + warn!( + "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. @@ -394,9 +428,19 @@ where ip_address: peer.ip(), logical_address: LogicalAddress(0x0000), }; - let response = self + // 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)?; + .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, @@ -405,9 +449,22 @@ where // 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 = std::vec::Vec::with_capacity(reply.encoded_size()?); - reply.encode(&mut encoded)?; - socket.send_to(&encoded, peer).await?; + 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}"); + } } } diff --git a/tests/udp_identification.rs b/tests/udp_identification.rs index 1e5fecf..f5fdb45 100644 --- a/tests/udp_identification.rs +++ b/tests/udp_identification.rs @@ -75,6 +75,29 @@ const VEHICLE_IDENTIFICATION_REQUEST: [u8; 8] = [0x02, 0xFD, 0x00, 0x01, 0x00, 0 /// 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 { @@ -128,11 +151,12 @@ async fn udp_responder_answers_a_vehicle_identification_request() { expect_identification_response(&client).await; } -/// A malformed datagram must not take the responder down. Anyone on the network -/// can reach a UDP responder, so a single bad probe killing the loop would let a +/// 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_survives_a_malformed_datagram() { +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"); @@ -140,13 +164,52 @@ async fn udp_responder_survives_a_malformed_datagram() { .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 one was skipped rather than - // fatal. It also proves the responder sent nothing for the bad datagram: this - // read would otherwise pick up that spurious reply and fail to decode it. + // 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; } From e1c1c63bef9112ad6526efc7b8fdb608db0dad8c Mon Sep 17 00:00:00 2001 From: Justin Kovacich Date: Wed, 12 Aug 2026 17:19:00 -0400 Subject: [PATCH 10/19] chore(release): 0.4.0 Breaking: ServerConnectionHandler::diagnostic_message takes a ResponseWriter. Co-Authored-By: Claude Opus 5 --- CHANGELOG.md | 66 ++++++++++++++++++++++++++++++++++++++++++++++++++++ Cargo.toml | 2 +- 2 files changed, 67 insertions(+), 1 deletion(-) create mode 100644 CHANGELOG.md diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 index 0000000..d1790af --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,66 @@ +# Changelog + +All notable changes to this crate are documented here. + +The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and +this crate adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +While the crate is at `0.x`, the minor position carries breaking changes. + +## 0.4.0 + +The release that makes `simple_doip::server` able to drive a real UDS tester: a +handler can now emit the `DiagnosticMessageAck` and the UDS response as separate +messages, hold an NRC `0x78` pending wait open, serve on a caller-chosen socket, +and answer UDP discovery. + +### Breaking + +- `ServerConnectionHandler::diagnostic_message` now takes a + `responses: &mut dyn ResponseWriter` and returns `Result<(), Error>` instead of + returning a single `OwnedMessage`. Handlers write the `DiagnosticMessageAck` + and the UDS response separately. Migration: what was `Ok(message)` becomes + `responses.send(message).await?; Ok(())`, with an ack sent first — a tester + blocks waiting for that ack, which is why the previous single-message signature + could not drive one. See `examples/echo_server.rs` for the ack-then-response + shape. + +### Added + +- `ResponseWriter` — the sink handlers write responses into. Each send reaches + the socket immediately, so a handler can hold an NRC `0x78` pending wait open + across awaits. +- `Server::run_server_with_listener` — serve a `TcpListener` the caller bound, + for loopback aliases and ephemeral ports. `run_server` keeps its signature and + delegates to it. +- `Server::run_udp_responder` — answer UDP vehicle-identification probes on a + caller-bound `UdpSocket`. Note that `run_server` still binds TCP only: an + entity that wants to be discoverable must drive both, via `join!`/`select!` or + a second task. +- `OwnedMessage::vehicle_identification_response`. + +### Fixed + +- The TCP accept loop no longer panics on an accept error; it logs and + continues. +- `TCP_NODELAY` is now set on accepted connections. `ConnectorSocket` already + set it client-side, but an accepted socket did not, so consecutive small + frames — an ack then a response, or successive NRC `0x78` pendings — waited + on the peer's delayed ACK. Measured at 43 ms of added latency per exchange, + straight out of the P2 budget. +- Every failure inside `run_udp_responder` is logged and skipped rather than + ending the loop. A fatal path here would be reachable by any host on the + network: on Windows, where an oversized datagram fails `recvfrom` with + `WSAEMSGSIZE` instead of truncating, a single 2 KB packet could otherwise end + discovery permanently. A failing identification handler could do the same. + +### Changed + +- `run_udp_responder` answers only the broadcast vehicle-identification request + (`0x0001`). The directed forms — `0x0002` naming an EID and `0x0003` naming a + VIN — are declined, because `Payload::decode` discards the EID/VIN bytes and + the responder cannot tell whether it is the addressee; answering regardless + would mean every entity on a network replied to a tester's directed probe. + Declining degrades to a discovery timeout, which testers already handle. A + correct implementation requires `Payload` to preserve those bytes, and the + `ServerConnectionHandler::vehicle_identification_with_eid` / + `vehicle_identification_with_vin` hooks remain unconsulted until it does. diff --git a/Cargo.toml b/Cargo.toml index 410c274..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" From 28ed45386e5e8f202c3d5377cde59908aecd3219 Mon Sep 17 00:00:00 2001 From: Justin Kovacich Date: Wed, 12 Aug 2026 17:36:17 -0400 Subject: [PATCH 11/19] docs: correct status claims this branch made false MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit README's Status list and ARCHITECTURE §7.6 both still said the server answers no vehicle-identification requests over UDP and that a failed `accept()` panics the server task. `run_udp_responder` answers the broadcast form, and the accept loop has logged-and-continued since e9cc750, so both claims were wrong in the two files a new integrator reads first. - Split the UDP claim into what is still true (no unsolicited announcement at power-on) and what replaced the rest, including that the caller must drive `run_udp_responder` itself because `run_server` binds TCP alone, and that only the broadcast `0x0001` form is answered. - Drop the accept-panic clause; note the current behavior in §7.6. - Point the one-connection-at-a-time entry at `run_server_with_listener`, where the loop now lives, and name the consequence a sim author cares about: a stalled tester wedges the entity. - Retire the dangling "single-response handler limitation" cross- reference in the examples section — that limitation was removed in 964ab24 and the anchor it pointed at no longer exists. Describe what `echo_server` actually does instead. - Reword §7's preamble so "none of it is scheduled" no longer contradicts §7.1, which is marked RESOLVED. - List `tests/udp_identification.rs` and `ResponseWriter` in the module and test maps, which the branch left out. - Date the 0.4.0 changelog entry as Keep a Changelog wants, and say the file begins at 0.4.0 so a reader does not read the missing 0.1-0.3 history as "nothing changed". Co-Authored-By: Claude Opus 5 --- ARCHITECTURE.md | 32 +++++++++++++++++++++++--------- CHANGELOG.md | 6 +++++- README.md | 42 ++++++++++++++++++++++++++++-------------- 3 files changed, 56 insertions(+), 24 deletions(-) diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 1ef0884..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,10 +380,12 @@ 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` — RESOLVED +### 7.1 `ServerConnectionHandler::diagnostic_message` — RESOLVED in 0.4.0 DoIP prescribes that a DoIP entity receiving a diagnostic message first sends a `DiagnosticMessageAck`, and then — separately and later — sends any functional @@ -562,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/CHANGELOG.md b/CHANGELOG.md index d1790af..a5b6e0d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,7 +6,11 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and this crate adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). While the crate is at `0.x`, the minor position carries breaking changes. -## 0.4.0 +This changelog begins at 0.4.0. Releases 0.1.0 through 0.3.x predate it and have +no entries here; their absence records only that nothing was written down, not +that nothing changed. Consult the git history for those. + +## 0.4.0 - 2026-08-12 The release that makes `simple_doip::server` able to drive a real UDS tester: a handler can now emit the `DiagnosticMessageAck` and the UDS response as separate diff --git a/README.md b/README.md index b72d154..850bd79 100644 --- a/README.md +++ b/README.md @@ -18,17 +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. + 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 @@ -41,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 @@ -122,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` From cdd0a0f497c3f4685c612cdd2f9ab39196297c0e Mon Sep 17 00:00:00 2001 From: Justin Kovacich Date: Wed, 12 Aug 2026 17:37:09 -0400 Subject: [PATCH 12/19] fix(server): back off before retrying a failed socket call MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Both non-fatal error paths — the TCP accept loop and the UDP responder's `recv_from` — retried immediately. The accept loop's own comment named its counterexample: it cited EMFILE as transient, but descriptor exhaustion persists until something else in the process releases an fd, and until then `accept` returns `Err` immediately on every iteration, forever. The panic this replaced was bad but loud and terminal; an unbounded error-log flood pegging a core is harder to diagnose in an unattended simulator. Sleep 100 ms in each `Err` arm before continuing — the standard accept-loop mitigation. A genuinely transient error costs one interval; a persistent one is bounded to ten retries a second. Also reword the accept-loop comment so it no longer implies every error it handles is transient. Untested by design: simulating descriptor exhaustion would destabilize the suite, and this is a strict improvement to an already-untested branch. Co-Authored-By: Claude Opus 5 --- src/server.rs | 39 ++++++++++++++++++++++++++++++++++----- 1 file changed, 34 insertions(+), 5 deletions(-) diff --git a/src/server.rs b/src/server.rs index bbe07a1..af4a455 100644 --- a/src/server.rs +++ b/src/server.rs @@ -23,11 +23,24 @@ use std::{ Arc, atomic::{AtomicUsize, Ordering}, }, + time::Duration, +}; +use tokio::{ + net::{TcpListener, TcpStream, UdpSocket}, + time::sleep, }; -use tokio::net::{TcpListener, TcpStream, UdpSocket}; use tokio_util::codec::{FramedRead, FramedWrite}; use tracing::{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 /// is asking. @@ -313,11 +326,20 @@ where } } Err(accept_error) => { - // Transient conditions (EMFILE, a peer resetting between - // the SYN and our accept) must not take the entity down — - // a simulator that aborts here turns a client bug into an - // opaque transport failure. + // 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; } } } @@ -381,6 +403,13 @@ where 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; } }; From 59a8fa85c0ce563033b5a99f6d3d7515e6784f6b Mon Sep 17 00:00:00 2001 From: Justin Kovacich Date: Wed, 12 Aug 2026 17:38:22 -0400 Subject: [PATCH 13/19] docs(server): disclose the limits of the two new entry points MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three gaps a caller only finds by reading the implementation: - `run_server_with_listener` sells the multi-entity-on-loopback-aliases case in detail but never says the loop serves one connection to completion before accepting the next. A sim author reading only this method builds exactly the topology — several entities, several testers — where one hung tester silently wedges an entity. State it, and point at README's Status section rather than repeating the explanation. - `run_server`'s rendered doc said nothing about UDP; the only disclosure was a `//` comment in the body, which rustdoc never emits. An entity started through the default entry point is invisible to a discovery probe and nothing said so. Add a `# Discovery` section, and since both methods take `&self`, show the `try_join!` composition rather than leaving the reader to derive it. - `run_udp_responder` said the socket is bound by the caller without saying that receiving broadcast probes requires binding `0.0.0.0`, not a specific address — the non-obvious half of that sentence. Also widen `Server`'s struct doc, which still described the type as TCP-only. Co-Authored-By: Claude Opus 5 --- src/server.rs | 45 +++++++++++++++++++++++++++++++++++++++++++-- 1 file changed, 43 insertions(+), 2 deletions(-) diff --git a/src/server.rs b/src/server.rs index af4a455..df038db 100644 --- a/src/server.rs +++ b/src/server.rs @@ -264,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, @@ -290,6 +295,28 @@ where /// Start listening for incoming `DoIP` TCP connections on the standard /// [`TCP_PORT`] across all interfaces. /// + /// 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. + /// + /// Discovery lives in [`run_udp_responder`](Self::run_udp_responder), on a + /// socket the caller binds. Both methods take `&self` and neither returns, + /// so compose them on one `Server`: + /// + /// ```ignore + /// let socket = UdpSocket::bind(("0.0.0.0", UDP_DISCOVERY_PORT)).await?; + /// tokio::try_join!(server.run_server(), server.run_udp_responder(socket))?; + /// ``` + /// + /// 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> { @@ -308,6 +335,14 @@ where /// 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. @@ -356,7 +391,13 @@ where /// /// 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 response content + /// 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 From f1c977dabcc62c0cb3d9726ae4c49746dc4273fd Mon Sep 17 00:00:00 2001 From: Justin Kovacich Date: Wed, 12 Aug 2026 17:38:57 -0400 Subject: [PATCH 14/19] fix(server): log routine UDP traffic at debug, not warn MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit An entity bound to `0.0.0.0:13400` sees every DoIP datagram on the network. A payload type this responder does not answer, and a directed identification probe it deliberately declines, are both ordinary traffic on a live bus — a tester doing directed discovery produces the latter as a matter of course. Warning about them makes a healthy entity look broken and buries the two cases that are genuinely worth a warning. Decode failures and send failures keep `warn!`: those describe a peer sending malformed bytes or an answer that did not get out, neither of which is expected. Co-Authored-By: Claude Opus 5 --- src/server.rs | 13 ++++++++++--- 1 file changed, 10 insertions(+), 3 deletions(-) diff --git a/src/server.rs b/src/server.rs index df038db..d1dc733 100644 --- a/src/server.rs +++ b/src/server.rs @@ -30,7 +30,7 @@ use tokio::{ time::sleep, }; 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. /// @@ -467,7 +467,11 @@ where }; if !matches!(message.payload, Payload::VehicleIdentificationRequest) { - warn!( + // 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 ); @@ -483,7 +487,10 @@ where message.header.payload_type, PayloadType::VehicleIdentificationRequest ) { - warn!( + // 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 From 5dc803194fc27c3675685d2602cb0585aea9bd12 Mon Sep 17 00:00:00 2001 From: Justin Kovacich Date: Wed, 12 Aug 2026 17:40:54 -0400 Subject: [PATCH 15/19] test(server): assert the ack code, not just the payload variant MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `handler_can_emit_ack_then_response` and the held-pending test both matched only `OwnedPayload::DiagnosticMessageAck(_)`. `ack_code` is a public field and both handlers send `RoutingConfirmationAck` through `send_positive_ack`, so checking it is one line each. The variant-only check was load-bearing on a bug: the ack constructors hardcode the positive payload type regardless of the code (ARCHITECTURE §7.2), so a handler that regressed to a negative code would still produce a frame these assertions accept. Asserting the code decouples the tests from that hardcode surviving unchanged. Co-Authored-By: Claude Opus 5 --- tests/integration_test.rs | 23 +++++++++++++++++------ 1 file changed, 17 insertions(+), 6 deletions(-) diff --git a/tests/integration_test.rs b/tests/integration_test.rs index a3bda50..e05c7c3 100644 --- a/tests/integration_test.rs +++ b/tests/integration_test.rs @@ -1044,11 +1044,16 @@ async fn handler_can_emit_ack_then_response() { send_diagnostic_message(&mut writer, CLIENT_LOGICAL_ADDRESS, &[0x22, 0xFD, 0x69]).await; let first = read_message(&mut reader).await; - assert!( - matches!(first.payload, OwnedPayload::DiagnosticMessageAck(_)), - "expected DiagnosticMessageAck first, got {:?}", - first.payload - ); + 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 { @@ -1262,7 +1267,13 @@ async fn handler_holds_pending_wait_open_between_sends() { let ack = read_message(&mut reader).await; let ack_at = started.elapsed(); - assert!(matches!(ack.payload, OwnedPayload::DiagnosticMessageAck(_))); + 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 { From 3b2d8b447d525db08cfb0a89f36ef02d9324802f Mon Sep 17 00:00:00 2001 From: Justin Kovacich Date: Wed, 12 Aug 2026 17:40:54 -0400 Subject: [PATCH 16/19] docs(server): make the discovery composition example compile MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The `# Discovery` snippet on `run_server` was an `ignore` block, which rustdoc never compiles — exactly the kind of sample that rots into something that no longer builds. Hidden lines wrapping it in a function generic over the handler make it a real doctest without needing a concrete `ServerConnectionHandler` in the example. Co-Authored-By: Claude Opus 5 --- src/server.rs | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/src/server.rs b/src/server.rs index d1dc733..4583ab0 100644 --- a/src/server.rs +++ b/src/server.rs @@ -309,9 +309,14 @@ where /// socket the caller binds. Both methods take `&self` and neither returns, /// so compose them on one `Server`: /// - /// ```ignore + /// ``` + /// # 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, From 74b675a8f8f2e19409bafdb251cbf4bd57a1b81f Mon Sep 17 00:00:00 2001 From: Justin Kovacich Date: Wed, 12 Aug 2026 19:05:54 -0400 Subject: [PATCH 17/19] docs: correct the discovery doc's return claim and complete the changelog MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The # Discovery section said "neither returns", contradicting run_server's own # Errors block three lines below — a bind failure does propagate. Say what was meant: neither completes normally, and this one can still return early on a bind failure. The 0.4.0 Fixed entry also described the accept loop as "logs and continues" without the 100ms backoff that shipped with it, and omitted the UDP log-level demotions entirely. Co-Authored-By: Claude Opus 5 --- CHANGELOG.md | 11 +++++++++-- src/server.rs | 6 ++++-- 2 files changed, 13 insertions(+), 4 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index a5b6e0d..61758e3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -44,8 +44,15 @@ and answer UDP discovery. ### Fixed -- The TCP accept loop no longer panics on an accept error; it logs and - continues. +- The TCP accept loop no longer panics on an accept error; it logs, waits + 100 ms, and continues. The delay matters because not every accept error is + transient: descriptor exhaustion (EMFILE/ENFILE) persists until an unrelated + descriptor is released, and retrying it without a pause would peg a core and + emit an unbounded error log. The UDP receive loop backs off the same way. +- Routine UDP traffic — a datagram whose payload type this responder does not + serve, and a directed identification request it declines — now logs at + `debug` rather than `warn`. On an entity bound to `0.0.0.0` with a tester + doing directed discovery, both are expected traffic, not warnings. - `TCP_NODELAY` is now set on accepted connections. `ConnectorSocket` already set it client-side, but an accepted socket did not, so consecutive small frames — an ack then a response, or successive NRC `0x78` pendings — waited diff --git a/src/server.rs b/src/server.rs index 4583ab0..7d654ef 100644 --- a/src/server.rs +++ b/src/server.rs @@ -306,8 +306,10 @@ where /// nothing logs that fact, because no datagram is ever received. /// /// Discovery lives in [`run_udp_responder`](Self::run_udp_responder), on a - /// socket the caller binds. Both methods take `&self` and neither returns, - /// so compose them on one `Server`: + /// 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}}; From a28876cc091e9fb2058f7f8eaaaa58a048fc6535 Mon Sep 17 00:00:00 2001 From: Justin Kovacich Date: Wed, 12 Aug 2026 21:26:59 -0400 Subject: [PATCH 18/19] Revert "chore(release): add CHANGELOG.md" MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The changelog was not asked for and this repo did not have one. The 0.4.0 migration note lives in the PR body and the release notes instead. The version bump from the same commit stays — only the file is removed. Co-Authored-By: Claude Opus 5 --- CHANGELOG.md | 77 ---------------------------------------------------- 1 file changed, 77 deletions(-) delete mode 100644 CHANGELOG.md diff --git a/CHANGELOG.md b/CHANGELOG.md deleted file mode 100644 index 61758e3..0000000 --- a/CHANGELOG.md +++ /dev/null @@ -1,77 +0,0 @@ -# Changelog - -All notable changes to this crate are documented here. - -The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and -this crate adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). -While the crate is at `0.x`, the minor position carries breaking changes. - -This changelog begins at 0.4.0. Releases 0.1.0 through 0.3.x predate it and have -no entries here; their absence records only that nothing was written down, not -that nothing changed. Consult the git history for those. - -## 0.4.0 - 2026-08-12 - -The release that makes `simple_doip::server` able to drive a real UDS tester: a -handler can now emit the `DiagnosticMessageAck` and the UDS response as separate -messages, hold an NRC `0x78` pending wait open, serve on a caller-chosen socket, -and answer UDP discovery. - -### Breaking - -- `ServerConnectionHandler::diagnostic_message` now takes a - `responses: &mut dyn ResponseWriter` and returns `Result<(), Error>` instead of - returning a single `OwnedMessage`. Handlers write the `DiagnosticMessageAck` - and the UDS response separately. Migration: what was `Ok(message)` becomes - `responses.send(message).await?; Ok(())`, with an ack sent first — a tester - blocks waiting for that ack, which is why the previous single-message signature - could not drive one. See `examples/echo_server.rs` for the ack-then-response - shape. - -### Added - -- `ResponseWriter` — the sink handlers write responses into. Each send reaches - the socket immediately, so a handler can hold an NRC `0x78` pending wait open - across awaits. -- `Server::run_server_with_listener` — serve a `TcpListener` the caller bound, - for loopback aliases and ephemeral ports. `run_server` keeps its signature and - delegates to it. -- `Server::run_udp_responder` — answer UDP vehicle-identification probes on a - caller-bound `UdpSocket`. Note that `run_server` still binds TCP only: an - entity that wants to be discoverable must drive both, via `join!`/`select!` or - a second task. -- `OwnedMessage::vehicle_identification_response`. - -### Fixed - -- The TCP accept loop no longer panics on an accept error; it logs, waits - 100 ms, and continues. The delay matters because not every accept error is - transient: descriptor exhaustion (EMFILE/ENFILE) persists until an unrelated - descriptor is released, and retrying it without a pause would peg a core and - emit an unbounded error log. The UDP receive loop backs off the same way. -- Routine UDP traffic — a datagram whose payload type this responder does not - serve, and a directed identification request it declines — now logs at - `debug` rather than `warn`. On an entity bound to `0.0.0.0` with a tester - doing directed discovery, both are expected traffic, not warnings. -- `TCP_NODELAY` is now set on accepted connections. `ConnectorSocket` already - set it client-side, but an accepted socket did not, so consecutive small - frames — an ack then a response, or successive NRC `0x78` pendings — waited - on the peer's delayed ACK. Measured at 43 ms of added latency per exchange, - straight out of the P2 budget. -- Every failure inside `run_udp_responder` is logged and skipped rather than - ending the loop. A fatal path here would be reachable by any host on the - network: on Windows, where an oversized datagram fails `recvfrom` with - `WSAEMSGSIZE` instead of truncating, a single 2 KB packet could otherwise end - discovery permanently. A failing identification handler could do the same. - -### Changed - -- `run_udp_responder` answers only the broadcast vehicle-identification request - (`0x0001`). The directed forms — `0x0002` naming an EID and `0x0003` naming a - VIN — are declined, because `Payload::decode` discards the EID/VIN bytes and - the responder cannot tell whether it is the addressee; answering regardless - would mean every entity on a network replied to a tester's directed probe. - Declining degrades to a discovery timeout, which testers already handle. A - correct implementation requires `Payload` to preserve those bytes, and the - `ServerConnectionHandler::vehicle_identification_with_eid` / - `vehicle_identification_with_vin` hooks remain unconsulted until it does. From ff5085ad4fd6262b0ecd1133b8a28ea0fbb87430 Mon Sep 17 00:00:00 2001 From: Justin Kovacich Date: Thu, 13 Aug 2026 09:51:21 -0400 Subject: [PATCH 19/19] build: record the 0.4.0 version bump in Cargo.lock MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Cargo.toml went to 0.4.0 but the lockfile still pinned simple_doip 0.3.0, so `cargo publish --dry-run` re-resolved and rewrote Cargo.lock before checking the tree — then failed the CI `package` job with "1 files in the working directory contain changes that were not yet committed into git". Only this crate's own version entry changes; no dependency pins move. Co-Authored-By: Claude Opus 5 --- Cargo.lock | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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",