From ada2553fda5cb6b7a7a84677e9260127d6dd137a Mon Sep 17 00:00:00 2001 From: Justin Kovacich Date: Wed, 22 Apr 2026 14:44:32 -0400 Subject: [PATCH 01/12] Switch to existing collections to heapless, hide other non-embedded features behind gates --- src/client/error.rs | 10 ++ src/client/inner.rs | 177 ++++++++++++++++++++++++----- src/client/mod.rs | 26 +++++ src/client/session.rs | 83 ++++++++++++-- src/server/subscription_manager.rs | 127 ++++++++++++++++++--- 5 files changed, 374 insertions(+), 49 deletions(-) diff --git a/src/client/error.rs b/src/client/error.rs index 64af3814..f0f2268c 100644 --- a/src/client/error.rs +++ b/src/client/error.rs @@ -1,7 +1,12 @@ use thiserror::Error; /// Errors that can occur during SOME/IP client operations. +/// +/// Marked `#[non_exhaustive]` so that future variants (for example, new +/// transport-specific error conditions in upcoming releases) can be added +/// without a further breaking change. #[derive(Error, Debug)] +#[non_exhaustive] pub enum Error { /// A SOME/IP protocol-level error. #[error(transparent)] @@ -24,4 +29,9 @@ pub enum Error { /// An E2E protection or checking error occurred. #[error(transparent)] E2e(#[from] crate::e2e::Error), + /// A fixed-capacity internal structure is full. The argument names the + /// structure so bare-metal users can size the corresponding compile-time + /// constant up (e.g. `"unicast_sockets"`). + #[error("internal capacity exceeded: {0}")] + Capacity(&'static str), } diff --git a/src/client/inner.rs b/src/client/inner.rs index 412c6c6e..d10a0938 100644 --- a/src/client/inner.rs +++ b/src/client/inner.rs @@ -1,6 +1,6 @@ +use heapless::{Deque, index_map::FnvIndexMap}; use std::{ borrow::ToOwned, - collections::{HashMap, VecDeque}, future, net::{Ipv4Addr, SocketAddr, SocketAddrV4}, sync::{Arc, Mutex}, @@ -29,6 +29,19 @@ use crate::{ use super::error::Error; +/// Max depth of the internal control-message queue. Each entry is one +/// in-flight `ControlMessage`. Must be generous enough to absorb bursts +/// from `Client` callers between event-loop ticks. +const REQUEST_QUEUE_CAP: usize = 32; + +/// Max number of outstanding unicast request-response pairs. Each entry is +/// a `request_id` awaiting a reply. Must be a power of two. +const PENDING_RESPONSES_CAP: usize = 64; + +/// Max number of bound unicast sockets tracked by port. Must be a power of +/// two. +const UNICAST_SOCKETS_CAP: usize = 8; + pub(super) enum ControlMessage { SetInterface(Ipv4Addr, oneshot::Sender>), BindDiscovery(oneshot::Sender>), @@ -238,10 +251,11 @@ pub(super) struct Inner { /// MPSC Receiver used to receive control messages from outer client control_receiver: Receiver>, /// Queue of pending control messages to process - request_queue: VecDeque>, + request_queue: Deque, REQUEST_QUEUE_CAP>, /// Pending request-responses keyed by `request_id` (`client_id` << 16 | `session_counter`). /// Set by `SendToService`, cleared when a matching unicast arrives. - pending_responses: HashMap>>, + pending_responses: + FnvIndexMap>, PENDING_RESPONSES_CAP>, /// Unbounded sender used to send updates to outer client update_sender: mpsc::UnboundedSender>, /// Target interface for sockets @@ -249,7 +263,7 @@ pub(super) struct Inner { /// Socket manager for service discovery if bound discovery_socket: Option>, /// Socket managers for unicast messages, keyed by local port - unicast_sockets: HashMap>, + unicast_sockets: FnvIndexMap, UNICAST_SOCKETS_CAP>, /// Per-sender SD session state for reboot detection session_tracker: SessionTracker, /// Registry of known service endpoints (auto-populated from SD + manual) @@ -301,12 +315,12 @@ where let (update_sender, update_receiver) = mpsc::unbounded_channel(); let inner = Self { control_receiver, - request_queue: VecDeque::new(), - pending_responses: HashMap::new(), + request_queue: Deque::new(), + pending_responses: FnvIndexMap::new(), update_sender, interface, discovery_socket: None, - unicast_sockets: HashMap::new(), + unicast_sockets: FnvIndexMap::new(), session_tracker: SessionTracker::default(), service_registry: ServiceRegistry::default(), run: true, @@ -359,9 +373,30 @@ where { return Ok(socket.port()); } + // Check capacity before asking the OS for a port so we don't + // bind-then-drop a socket we can't track. + if self.unicast_sockets.len() >= UNICAST_SOCKETS_CAP { + warn!( + "unicast_sockets at capacity ({}); refusing new bind of port {}", + UNICAST_SOCKETS_CAP, port + ); + return Err(Error::Capacity("unicast_sockets")); + } let unicast_socket = SocketManager::bind(port, Arc::clone(&self.e2e_registry))?; let bound_port = unicast_socket.port(); - self.unicast_sockets.insert(bound_port, unicast_socket); + // Capacity was checked above, so insert cannot report "full" here. + // A defensive check guards against a future refactor that changes + // the ordering. + if self + .unicast_sockets + .insert(bound_port, unicast_socket) + .is_err() + { + error!( + "unicast_sockets insert failed after capacity check passed — invariant violation" + ); + return Err(Error::Capacity("unicast_sockets")); + } debug!("Bound unicast socket on port {}", bound_port); Ok(bound_port) } @@ -400,7 +435,11 @@ where /// Receive from any bound unicast socket. Returns the first message ready /// from any socket. If no sockets are bound, returns a future that never resolves. async fn receive_any_unicast( - unicast_sockets: &mut HashMap>, + unicast_sockets: &mut FnvIndexMap< + u16, + SocketManager, + UNICAST_SOCKETS_CAP, + >, ) -> Result, Error> { if unicast_sockets.is_empty() { return future::pending().await; @@ -432,14 +471,18 @@ where self.interface ); self.unbind_discovery().await; + // Re-enqueue after pop — queue has a free slot. self.request_queue - .push_front(ControlMessage::SetInterface(interface, response)); + .push_front(ControlMessage::SetInterface(interface, response)) + .ok(); return; } if self.interface != interface { self.set_interface(interface); + // Re-enqueue after pop — queue has a free slot. self.request_queue - .push_front(ControlMessage::SetInterface(interface, response)); + .push_front(ControlMessage::SetInterface(interface, response)) + .ok(); return; } info!("Binding to interface: {}", interface); @@ -474,10 +517,12 @@ where None => { match self.bind_discovery() { Ok(()) => { - // Discovery socket successfully bound, send the message on the next loop - self.request_queue.push_front(ControlMessage::SendSD( - target, header, response, - )); + // Re-enqueue after pop — queue has a free slot. + self.request_queue + .push_front(ControlMessage::SendSD( + target, header, response, + )) + .ok(); } Err(e) => { error!( @@ -609,7 +654,18 @@ where match send_result { Ok(()) => { let _ = send_complete.send(Ok(())); - self.pending_responses.insert(request_id, response); + if self.pending_responses.insert(request_id, response).is_err() { + // Map full: the response Sender was returned + // and is now dropped; caller's response oneshot + // will observe cancellation. Send already + // succeeded — the peer's reply, if any, will + // arrive as a ClientUpdate::Unicast instead. + warn!( + "pending_responses at capacity ({}); response tracking \ + dropped for request_id 0x{:08X}", + PENDING_RESPONSES_CAP, request_id + ); + } } Err(e) => { let _ = send_complete.send(Err(e)); @@ -674,15 +730,18 @@ where match &mut self.discovery_socket { None => match self.bind_discovery() { Ok(()) => { - self.request_queue.push_front(ControlMessage::Subscribe { - service_id, - instance_id, - major_version, - ttl, - event_group_id, - client_port, - response, - }); + // Re-enqueue after pop — queue has a free slot. + self.request_queue + .push_front(ControlMessage::Subscribe { + service_id, + instance_id, + major_version, + ttl, + event_group_id, + client_port, + response, + }) + .ok(); } Err(e) => { let _ = response.send(Err(e)); @@ -746,7 +805,16 @@ where ctrl = control_receiver.recv() => { if let Some(ctrl) = ctrl { debug!("Received control message: {:?}", ctrl); - request_queue.push_back(ctrl); + if request_queue.push_back(ctrl).is_err() { + // Queue full: the rejected ControlMessage is + // dropped, so any oneshot senders inside it + // cancel — callers awaiting those receivers + // will observe `RecvError`. + warn!( + "request_queue at capacity ({}); dropping control message", + REQUEST_QUEUE_CAP + ); + } } else { // The sender has been dropped, so we should exit *run = false; @@ -946,6 +1014,63 @@ mod tests { assert!(s.contains("event_group_id")); } + /// Build an [`Inner`] without spawning the run loop, for direct + /// unit-testing of state-mutating methods. + fn make_inner_for_test() -> Inner { + let (_control_sender, control_receiver) = mpsc::channel(4); + let (update_sender, _update_receiver) = mpsc::unbounded_channel(); + Inner { + control_receiver, + request_queue: Deque::new(), + pending_responses: FnvIndexMap::new(), + update_sender, + interface: Ipv4Addr::LOCALHOST, + discovery_socket: None, + unicast_sockets: FnvIndexMap::new(), + session_tracker: SessionTracker::default(), + service_registry: ServiceRegistry::default(), + run: true, + client_id: 0x1234, + session_counter: 1, + sd_session_id: 1, + sd_session_has_wrapped: false, + e2e_registry: Arc::new(Mutex::new(E2ERegistry::new())), + multicast_loopback: false, + phantom: std::marker::PhantomData, + } + } + + #[tokio::test] + async fn bind_unicast_returns_capacity_error_when_map_full() { + let mut inner = make_inner_for_test(); + + // Fill unicast_sockets to capacity using ephemeral binds (port 0). + // Each call with port=0 creates a fresh socket on a distinct OS-chosen + // port, so the cap is what gates — not duplicate-key collapse. + for _ in 0..UNICAST_SOCKETS_CAP { + let bound = inner + .bind_unicast(0) + .expect("ephemeral bind below cap should succeed"); + assert_ne!(bound, 0, "OS should assign a non-zero ephemeral port"); + } + assert_eq!(inner.unicast_sockets.len(), UNICAST_SOCKETS_CAP); + + // The next bind must fail with Error::Capacity and must NOT bind a + // socket (pre-bind capacity check). + let err = inner + .bind_unicast(0) + .expect_err("bind past cap should fail"); + match err { + Error::Capacity(name) => assert_eq!(name, "unicast_sockets"), + other => panic!("expected Error::Capacity, got {other:?}"), + } + assert_eq!( + inner.unicast_sockets.len(), + UNICAST_SOCKETS_CAP, + "map should remain at capacity, not bind-then-drop a new socket" + ); + } + #[tokio::test] async fn test_inner_spawn_and_shutdown() { let (control_sender, mut update_receiver) = Inner::::spawn( diff --git a/src/client/mod.rs b/src/client/mod.rs index 026f2b74..ca7080df 100644 --- a/src/client/mod.rs +++ b/src/client/mod.rs @@ -1,3 +1,17 @@ +//! SOME/IP client. +//! +//! # Memory footprint +//! +//! The client's internal `Inner` state is allocated inline rather than on +//! the heap. With the default capacity constants declared in `inner.rs` — +//! `REQUEST_QUEUE_CAP=32`, `PENDING_RESPONSES_CAP=64`, `UNICAST_SOCKETS_CAP=8`, +//! and `SESSION_CAP=64` — `Inner

` occupies on the order of **8–12 KiB**, +//! depending on `sizeof::

()` and `sizeof::>()`. On +//! `std + tokio`, this is allocated on the heap when the run-loop is +//! spawned, so the overhead is invisible to callers. On the bare-metal +//! port (future), whoever drives the future must arrange storage for it +//! (either a `static` or a heap allocator); the capacity constants are the +//! primary knob for trimming this footprint. mod error; mod inner; mod service_registry; @@ -550,6 +564,18 @@ where /// /// Call `.response()` on the returned handle to await the reply payload. /// + /// # Saturation behavior + /// + /// Response tracking uses a fixed-capacity internal map. If it is + /// saturated at the moment the reply-tracking slot would be installed, + /// this method still returns `Ok(PendingResponse)` — the UDP send has + /// already happened — but the returned `PendingResponse` will resolve to + /// `Err(RecvError)` because the tracking slot was dropped. Any reply that + /// later arrives for that `request_id` is delivered as + /// [`ClientUpdate::Unicast`] on the update stream instead of through the + /// `PendingResponse`. Treat `RecvError` as "reply lost to saturation", + /// not "send failed". A `warn!`-level log accompanies the drop. + /// /// # Errors /// /// Returns an error if the service is not found, unicast binding fails, diff --git a/src/client/session.rs b/src/client/session.rs index 9aa63663..19f3bdba 100644 --- a/src/client/session.rs +++ b/src/client/session.rs @@ -1,5 +1,12 @@ use crate::protocol::sd::RebootFlag; -use std::{collections::HashMap, net::SocketAddr}; +use heapless::index_map::FnvIndexMap; +use std::net::SocketAddr; + +/// Max number of distinct `(sender, transport, service, instance)` tuples tracked +/// for reboot detection. Must be a power of two (heapless `FnvIndexMap` +/// requirement). Sized for a small fleet of peers each offering several +/// services; bare-metal builds with more peers may need to edit this constant. +const SESSION_CAP: usize = 64; /// Distinguishes multicast vs unicast transport for per-sender session tracking. /// The AUTOSAR spec requires separate session ID tracking per transport. @@ -45,9 +52,30 @@ pub enum SessionVerdict { /// Tracking per service instance (rather than per sender) avoids false /// positives when a sensor interleaves SD offers for multiple services /// with independent session counters on the same source address. -#[derive(Debug, Default)] +/// +/// Capacity is bounded at compile time ([`SESSION_CAP`]); see module docs. +/// When the map is full, new sender entries are dropped with a `warn!` log +/// and reboot detection for those senders is disabled. +/// +/// # Security posture +/// +/// The backing map uses FNV hashing rather than the DoS-resistant hasher used +/// by `std::collections::HashMap`. For SOME/IP on isolated automotive or +/// sensor networks this is not a concern. Deployments where `SessionKey` +/// inputs (notably `SocketAddr`) are adversary-controlled should be aware +/// that an attacker can craft keys to force collisions and degrade lookup +/// cost; the blast radius is bounded by [`SESSION_CAP`]. +#[derive(Debug)] pub struct SessionTracker { - state: HashMap, + state: FnvIndexMap, +} + +impl Default for SessionTracker { + fn default() -> Self { + Self { + state: FnvIndexMap::new(), + } + } } impl SessionTracker { @@ -89,13 +117,22 @@ impl SessionTracker { } } }; - self.state.insert( - key, - SessionState { - last_session_id: session_id, - last_reboot_flag: reboot_flag, - }, - ); + let new_state = SessionState { + last_session_id: session_id, + last_reboot_flag: reboot_flag, + }; + if self.state.insert(key, new_state).is_err() { + // Map at capacity and key is new — silently dropping the update + // would lose reboot-detection state. Log once so bare-metal users + // can size `SESSION_CAP` up. + tracing::warn!( + "SessionTracker at capacity ({}); dropping new sender state for \ + svc=0x{:04X} inst=0x{:04X}. Reboot detection disabled for this entry.", + SESSION_CAP, + service_id, + instance_id + ); + } verdict } } @@ -310,4 +347,30 @@ mod tests { let verdict = tracker.check(addr(1000), TransportKind::Multicast, SVC, INST, 2, CONT); assert_eq!(verdict, SessionVerdict::Ok); } + + #[test] + fn capacity_overflow_drops_new_entries_but_keeps_existing_tracking() { + // Fill the tracker to capacity with unique (sender, service) tuples. + let mut tracker = SessionTracker::default(); + for i in 0..super::SESSION_CAP { + let port = 1000 + u16::try_from(i).unwrap(); + let v = tracker.check(addr(port), TransportKind::Multicast, SVC, INST, 1, RB); + assert_eq!(v, SessionVerdict::Initial); + } + + // One more insert — map is full, new entry dropped. The verdict is + // still Initial (no prior state for this key), but the state is + // never stored so a follow-up is also Initial. + let overflow_addr = addr(9999); + let v = tracker.check(overflow_addr, TransportKind::Multicast, SVC, INST, 1, RB); + assert_eq!(v, SessionVerdict::Initial); + // Because the insert failed, a second call with the same key still + // sees no stored state. + let v = tracker.check(overflow_addr, TransportKind::Multicast, SVC, INST, 2, RB); + assert_eq!(v, SessionVerdict::Initial); + + // Previously-tracked senders continue to work normally. + let v = tracker.check(addr(1000), TransportKind::Multicast, SVC, INST, 2, RB); + assert_eq!(v, SessionVerdict::Ok); + } } diff --git a/src/server/subscription_manager.rs b/src/server/subscription_manager.rs index 28667bc8..c0e21b93 100644 --- a/src/server/subscription_manager.rs +++ b/src/server/subscription_manager.rs @@ -1,13 +1,27 @@ //! Manages event group subscriptions use super::service_info::Subscriber; -use std::{collections::HashMap, net::SocketAddrV4, vec::Vec}; +use heapless::{Vec as HeaplessVec, index_map::FnvIndexMap}; +use std::{net::SocketAddrV4, vec::Vec}; -/// Manages subscriptions to event groups +/// Max number of distinct `(service_id, instance_id, event_group_id)` event +/// groups with active subscribers. Must be a power of two. +const EVENT_GROUPS_CAP: usize = 32; + +/// Max number of subscribers per event group. Excess subscribers are dropped +/// with a `warn!` log rather than silently. +const SUBSCRIBERS_PER_GROUP: usize = 16; + +type SubscribersList = HeaplessVec; + +/// Manages subscriptions to event groups. +/// +/// Capacity is bounded at compile time: up to [`EVENT_GROUPS_CAP`] distinct +/// event groups, each with up to [`SUBSCRIBERS_PER_GROUP`] subscribers. #[derive(Debug)] pub struct SubscriptionManager { /// Map of (`service_id`, `instance_id`, `event_group_id`) -> list of subscribers - subscriptions: HashMap<(u16, u16, u16), Vec>, + subscriptions: FnvIndexMap<(u16, u16, u16), SubscribersList, EVENT_GROUPS_CAP>, } impl SubscriptionManager { @@ -15,7 +29,7 @@ impl SubscriptionManager { #[must_use] pub fn new() -> Self { Self { - subscriptions: HashMap::new(), + subscriptions: FnvIndexMap::new(), } } @@ -28,12 +42,37 @@ impl SubscriptionManager { subscriber_addr: SocketAddrV4, ) { let key = (service_id, instance_id, event_group_id); - let subscribers = self.subscriptions.entry(key).or_default(); - // Deduplicate: if this address is already subscribed, just refresh (don't add again) - if subscribers.iter().any(|s| s.address == subscriber_addr) { - tracing::debug!( - "Refreshed existing subscriber {} for service 0x{:04X}, instance {}, event group 0x{:04X}", + if let Some(subscribers) = self.subscriptions.get_mut(&key) { + // Deduplicate: if this address is already subscribed, just refresh (don't add again) + if subscribers.iter().any(|s| s.address == subscriber_addr) { + tracing::debug!( + "Refreshed existing subscriber {} for service 0x{:04X}, instance {}, event group 0x{:04X}", + subscriber_addr, + service_id, + instance_id, + event_group_id + ); + return; + } + + let subscriber = + Subscriber::new(subscriber_addr, service_id, instance_id, event_group_id); + if subscribers.push(subscriber).is_err() { + tracing::warn!( + "Subscribers-per-group at capacity ({}); dropping new subscriber {} \ + for service 0x{:04X}, instance {}, event group 0x{:04X}", + SUBSCRIBERS_PER_GROUP, + subscriber_addr, + service_id, + instance_id, + event_group_id + ); + return; + } + + tracing::info!( + "Subscriber {} added for service 0x{:04X}, instance {}, event group 0x{:04X}", subscriber_addr, service_id, instance_id, @@ -42,8 +81,29 @@ impl SubscriptionManager { return; } - let subscriber = Subscriber::new(subscriber_addr, service_id, instance_id, event_group_id); - subscribers.push(subscriber); + // New event group — allocate the list and insert. + let mut list = SubscribersList::new(); + // Pushing into an empty heapless::Vec with cap >= 1 cannot fail. + list.push(Subscriber::new( + subscriber_addr, + service_id, + instance_id, + event_group_id, + )) + .ok(); + + if self.subscriptions.insert(key, list).is_err() { + tracing::warn!( + "Event-group map at capacity ({}); dropping subscriber {} for new group \ + service 0x{:04X}, instance {}, event group 0x{:04X}", + EVENT_GROUPS_CAP, + subscriber_addr, + service_id, + instance_id, + event_group_id + ); + return; + } tracing::info!( "Subscriber {} added for service 0x{:04X}, instance {}, event group 0x{:04X}", @@ -90,13 +150,16 @@ impl SubscriptionManager { event_group_id: u16, ) -> Vec { let key = (service_id, instance_id, event_group_id); - self.subscriptions.get(&key).cloned().unwrap_or_default() + self.subscriptions + .get(&key) + .map(|list| list.iter().cloned().collect()) + .unwrap_or_default() } /// Get total number of active subscriptions #[must_use] pub fn subscription_count(&self) -> usize { - self.subscriptions.values().map(std::vec::Vec::len).sum() + self.subscriptions.values().map(|v| v.len()).sum() } } @@ -164,4 +227,42 @@ mod tests { let manager = SubscriptionManager::default(); assert_eq!(manager.subscription_count(), 0); } + + #[test] + fn subscribers_per_group_capacity_overflow() { + let mut manager = SubscriptionManager::new(); + // Fill one event group to capacity. + for i in 0..SUBSCRIBERS_PER_GROUP { + let addr = + SocketAddrV4::new(Ipv4Addr::new(10, 0, 0, 1), 8000 + u16::try_from(i).unwrap()); + manager.subscribe(0x5B, 1, 0x01, addr); + } + assert_eq!(manager.subscription_count(), SUBSCRIBERS_PER_GROUP); + + // One more is dropped. + let extra = SocketAddrV4::new(Ipv4Addr::new(10, 0, 0, 1), 9999); + manager.subscribe(0x5B, 1, 0x01, extra); + assert_eq!(manager.subscription_count(), SUBSCRIBERS_PER_GROUP); + // Extra subscriber should not appear in the list. + let subs = manager.get_subscribers(0x5B, 1, 0x01); + assert!(subs.iter().all(|s| s.address != extra)); + } + + #[test] + fn event_groups_capacity_overflow() { + let mut manager = SubscriptionManager::new(); + let addr = SocketAddrV4::new(Ipv4Addr::new(10, 0, 0, 1), 8000); + // Fill the outer map to capacity with distinct event groups. + for i in 0..EVENT_GROUPS_CAP { + let eg = u16::try_from(i).unwrap(); + manager.subscribe(0x5B, 1, eg, addr); + } + assert_eq!(manager.subscription_count(), EVENT_GROUPS_CAP); + + // A new event group beyond capacity is dropped. + let overflow_eg = u16::try_from(EVENT_GROUPS_CAP).unwrap(); + manager.subscribe(0x5B, 1, overflow_eg, addr); + assert_eq!(manager.subscription_count(), EVENT_GROUPS_CAP); + assert!(manager.get_subscribers(0x5B, 1, overflow_eg).is_empty()); + } } From a36af2f59dd55aaa34a8d01785a281a7eaa4e0f9 Mon Sep 17 00:00:00 2001 From: Justin Kovacich Date: Wed, 22 Apr 2026 22:29:25 -0400 Subject: [PATCH 02/12] Address PR #76 review feedback MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - inner.rs: on request_queue overflow, reject the ControlMessage by sending Err(Error::Capacity("request_queue")) on every oneshot it carries (via new ControlMessage::reject_with_capacity helper) rather than silently dropping it — the drop previously cancelled the oneshot and panicked callers awaiting with .unwrap(). - inner.rs: on pending_responses.insert saturation, destructure the returned (request_id, response) and send Err(Error::Capacity("pending_responses")) on the response sender; previously the sender was dropped, panicking PendingResponse::response. - mod.rs: update the send_to_service saturation doc to match the new explicit-error behavior. - mod.rs: fix the module-header footprint doc — SESSION_CAP lives in session.rs, not inner.rs. - session.rs: gate the saturation warn! behind a one-shot saturation_warned latch; previously the warning fired on every check() against every new key once the map was full, which meant log-spam at the packet rate. - error.rs: drop #[non_exhaustive] on client::Error; downstream crates relying on exhaustive matches shouldn't silently break before a planned breaking release. New tests: reject_with_capacity_notifies_every_sender covers every ControlMessage variant (including SendToService's dual senders); capacity_overflow_warns_only_on_first_hit locks in the saturation latch. Co-Authored-By: Claude Opus 4.7 --- src/client/error.rs | 7 ++- src/client/inner.rs | 123 +++++++++++++++++++++++++++++++++++++----- src/client/mod.rs | 17 +++--- src/client/session.rs | 56 +++++++++++++++---- 4 files changed, 170 insertions(+), 33 deletions(-) diff --git a/src/client/error.rs b/src/client/error.rs index f0f2268c..45c984ac 100644 --- a/src/client/error.rs +++ b/src/client/error.rs @@ -2,11 +2,10 @@ use thiserror::Error; /// Errors that can occur during SOME/IP client operations. /// -/// Marked `#[non_exhaustive]` so that future variants (for example, new -/// transport-specific error conditions in upcoming releases) can be added -/// without a further breaking change. +/// Not marked `#[non_exhaustive]` today: downstream crates that match on +/// this enum rely on exhaustiveness, and adding the attribute now would be +/// a silent breaking change. Revisit when a breaking release is planned. #[derive(Error, Debug)] -#[non_exhaustive] pub enum Error { /// A SOME/IP protocol-level error. #[error(transparent)] diff --git a/src/client/inner.rs b/src/client/inner.rs index d10a0938..0b49bfcc 100644 --- a/src/client/inner.rs +++ b/src/client/inner.rs @@ -245,6 +245,36 @@ impl ControlMessage

{ Self::ForceSdSessionWrappedForTest(wrapped, sender), ) } + + /// Consume this message and notify its oneshot senders with + /// `Error::Capacity(structure_name)` instead of silently dropping them. + /// + /// Dropping the senders would let the awaiting `oneshot::Receiver`s + /// resolve to `RecvError`, which the public APIs currently `.unwrap()` + /// — that would panic callers under load. Delivering an explicit + /// `Err(Error::Capacity(..))` turns a would-be panic into a normal + /// `Result` with a stable, descriptive error. + fn reject_with_capacity(self, structure_name: &'static str) { + match self { + Self::SetInterface(_, response) + | Self::BindDiscovery(response) + | Self::UnbindDiscovery(response) + | Self::SendSD(_, _, response) + | Self::AddEndpoint(_, _, _, _, response) + | Self::RemoveEndpoint(_, _, response) + | Self::Subscribe { response, .. } => { + let _ = response.send(Err(Error::Capacity(structure_name))); + } + Self::SendToService { + send_complete, + response, + .. + } => { + let _ = send_complete.send(Err(Error::Capacity(structure_name))); + let _ = response.send(Err(Error::Capacity(structure_name))); + } + } + } } pub(super) struct Inner { @@ -654,17 +684,24 @@ where match send_result { Ok(()) => { let _ = send_complete.send(Ok(())); - if self.pending_responses.insert(request_id, response).is_err() { - // Map full: the response Sender was returned - // and is now dropped; caller's response oneshot - // will observe cancellation. Send already - // succeeded — the peer's reply, if any, will - // arrive as a ClientUpdate::Unicast instead. + if let Err((_req_id, response)) = + self.pending_responses.insert(request_id, response) + { + // Map full: the send already succeeded, but + // we cannot track the reply. Deliver an + // explicit capacity error through the + // returned response sender rather than + // dropping it — otherwise `.expect(...)` on + // the receiver side would panic. Any reply + // that later arrives for `request_id` is + // delivered as `ClientUpdate::Unicast` on + // the update stream instead. warn!( "pending_responses at capacity ({}); response tracking \ dropped for request_id 0x{:08X}", PENDING_RESPONSES_CAP, request_id ); + let _ = response.send(Err(Error::Capacity("pending_responses"))); } } Err(e) => { @@ -805,15 +842,18 @@ where ctrl = control_receiver.recv() => { if let Some(ctrl) = ctrl { debug!("Received control message: {:?}", ctrl); - if request_queue.push_back(ctrl).is_err() { - // Queue full: the rejected ControlMessage is - // dropped, so any oneshot senders inside it - // cancel — callers awaiting those receivers - // will observe `RecvError`. + if let Err(rejected) = request_queue.push_back(ctrl) { + // Queue full: rather than silently drop the + // rejected ControlMessage (which would + // cancel its oneshot senders and panic any + // caller awaiting with `.unwrap()`), reply + // on each sender with + // `Err(Error::Capacity("request_queue"))`. warn!( - "request_queue at capacity ({}); dropping control message", + "request_queue at capacity ({}); rejecting control message", REQUEST_QUEUE_CAP ); + rejected.reject_with_capacity("request_queue"); } } else { // The sender has been dropped, so we should exit @@ -974,6 +1014,65 @@ mod tests { assert!(matches!(msg, ControlMessage::Subscribe { .. })); } + /// `reject_with_capacity` must notify every oneshot sender inside a + /// rejected `ControlMessage` with `Err(Error::Capacity(..))` — for + /// `SendToService`, _both_ the `send_complete` and `response` + /// channels. Dropping either channel would let a caller's `.unwrap()` + /// (or `.expect(...)` inside `PendingResponse::response()`) panic on + /// the resulting `RecvError`, which is exactly what Copilot flagged. + #[test] + fn reject_with_capacity_notifies_every_sender() { + fn expect_capacity( + rx: &mut oneshot::Receiver>, + label: &str, + ) { + match rx.try_recv() { + Ok(Err(Error::Capacity(s))) => assert_eq!(s, "request_queue", "{label}"), + other => panic!("{label}: expected Err(Capacity), got {other:?}"), + } + } + + // Variants carrying a single Result<(), Error> response sender. + let (mut rx, msg) = TestControl::set_interface(Ipv4Addr::LOCALHOST); + msg.reject_with_capacity("request_queue"); + expect_capacity(&mut rx, "SetInterface"); + + let (mut rx, msg) = TestControl::bind_discovery(); + msg.reject_with_capacity("request_queue"); + expect_capacity(&mut rx, "BindDiscovery"); + + let (mut rx, msg) = TestControl::unbind_discovery(); + msg.reject_with_capacity("request_queue"); + expect_capacity(&mut rx, "UnbindDiscovery"); + + let target = SocketAddrV4::new(Ipv4Addr::LOCALHOST, 1234); + let (mut rx, msg) = TestControl::send_sd(target, empty_sd_header()); + msg.reject_with_capacity("request_queue"); + expect_capacity(&mut rx, "SendSD"); + + let addr = SocketAddrV4::new(Ipv4Addr::LOCALHOST, 5000); + let (mut rx, msg) = TestControl::add_endpoint(0x1234, 0x0001, addr, 0); + msg.reject_with_capacity("request_queue"); + expect_capacity(&mut rx, "AddEndpoint"); + + let (mut rx, msg) = TestControl::remove_endpoint(0x1234, 0x0001); + msg.reject_with_capacity("request_queue"); + expect_capacity(&mut rx, "RemoveEndpoint"); + + let (mut rx, msg) = TestControl::subscribe(0x1234, 0x0001, 1, 3, 0x01, 0); + msg.reject_with_capacity("request_queue"); + expect_capacity(&mut rx, "Subscribe"); + + // SendToService carries two senders — both must be notified so that + // neither `send_rx.await.unwrap()?` nor `PendingResponse::response()` + // panics. + let message = Message::::new_sd(1, &empty_sd_header()); + let (mut send_rx, mut resp_rx, msg) = TestControl::send_to_service(0x1234, 0x0001, message); + msg.reject_with_capacity("request_queue"); + expect_capacity(&mut send_rx, "SendToService.send_complete"); + expect_capacity(&mut resp_rx, "SendToService.response"); + } + #[test] fn test_control_message_debug() { let (_rx, msg) = TestControl::set_interface(Ipv4Addr::LOCALHOST); diff --git a/src/client/mod.rs b/src/client/mod.rs index ca7080df..8cb2cb89 100644 --- a/src/client/mod.rs +++ b/src/client/mod.rs @@ -3,15 +3,16 @@ //! # Memory footprint //! //! The client's internal `Inner` state is allocated inline rather than on -//! the heap. With the default capacity constants declared in `inner.rs` — -//! `REQUEST_QUEUE_CAP=32`, `PENDING_RESPONSES_CAP=64`, `UNICAST_SOCKETS_CAP=8`, -//! and `SESSION_CAP=64` — `Inner

` occupies on the order of **8–12 KiB**, +//! the heap. With the default capacity constants used by the client +//! internals — `REQUEST_QUEUE_CAP=32`, `PENDING_RESPONSES_CAP=64`, and +//! `UNICAST_SOCKETS_CAP=8` in `inner.rs`, plus `SESSION_CAP=64` in +//! `session.rs` — `Inner

` occupies on the order of **8–12 KiB**, //! depending on `sizeof::

()` and `sizeof::>()`. On //! `std + tokio`, this is allocated on the heap when the run-loop is //! spawned, so the overhead is invisible to callers. On the bare-metal //! port (future), whoever drives the future must arrange storage for it -//! (either a `static` or a heap allocator); the capacity constants are the -//! primary knob for trimming this footprint. +//! (either a `static` or a heap allocator); these capacity constants are +//! the primary knobs for trimming this footprint. mod error; mod inner; mod service_registry; @@ -570,10 +571,10 @@ where /// saturated at the moment the reply-tracking slot would be installed, /// this method still returns `Ok(PendingResponse)` — the UDP send has /// already happened — but the returned `PendingResponse` will resolve to - /// `Err(RecvError)` because the tracking slot was dropped. Any reply that - /// later arrives for that `request_id` is delivered as + /// `Err(Error::Capacity("pending_responses"))`. Any reply that later + /// arrives for that `request_id` is delivered as /// [`ClientUpdate::Unicast`] on the update stream instead of through the - /// `PendingResponse`. Treat `RecvError` as "reply lost to saturation", + /// `PendingResponse`. Treat this error as "reply lost to saturation", /// not "send failed". A `warn!`-level log accompanies the drop. /// /// # Errors diff --git a/src/client/session.rs b/src/client/session.rs index 19f3bdba..6dc9b72f 100644 --- a/src/client/session.rs +++ b/src/client/session.rs @@ -68,12 +68,17 @@ pub enum SessionVerdict { #[derive(Debug)] pub struct SessionTracker { state: FnvIndexMap, + /// Set after the first saturation warning. Prevents the saturated-map + /// log from firing on every `check()` for every new key once capacity + /// is reached — which would spam the log at the packet rate. + saturation_warned: bool, } impl Default for SessionTracker { fn default() -> Self { Self { state: FnvIndexMap::new(), + saturation_warned: false, } } } @@ -123,15 +128,21 @@ impl SessionTracker { }; if self.state.insert(key, new_state).is_err() { // Map at capacity and key is new — silently dropping the update - // would lose reboot-detection state. Log once so bare-metal users - // can size `SESSION_CAP` up. - tracing::warn!( - "SessionTracker at capacity ({}); dropping new sender state for \ - svc=0x{:04X} inst=0x{:04X}. Reboot detection disabled for this entry.", - SESSION_CAP, - service_id, - instance_id - ); + // would lose reboot-detection state. Log the first time we hit + // the wall so bare-metal users can size `SESSION_CAP` up, then + // suppress further warnings so a saturated tracker does not + // spam the log at the incoming-packet rate. + if !self.saturation_warned { + tracing::warn!( + "SessionTracker at capacity ({}); dropping new sender state for \ + svc=0x{:04X} inst=0x{:04X}. Reboot detection disabled for this \ + entry and any further new entries (subsequent drops not logged).", + SESSION_CAP, + service_id, + instance_id + ); + self.saturation_warned = true; + } } verdict } @@ -373,4 +384,31 @@ mod tests { let v = tracker.check(addr(1000), TransportKind::Multicast, SVC, INST, 2, RB); assert_eq!(v, SessionVerdict::Ok); } + + #[test] + fn capacity_overflow_warns_only_on_first_hit() { + // `saturation_warned` is the latch that guards the tracing::warn! + // call in `check()`. It must flip false → true on the first + // rejected insert and stay true for subsequent hits — otherwise + // a saturated tracker spams the log at the packet rate. + let mut tracker = SessionTracker::default(); + for i in 0..super::SESSION_CAP { + let port = 1000 + u16::try_from(i).unwrap(); + tracker.check(addr(port), TransportKind::Multicast, SVC, INST, 1, RB); + } + assert!( + !tracker.saturation_warned, + "filling to exactly capacity must not trip the warn flag", + ); + + // First overflowing key: flag flips to true. + tracker.check(addr(9001), TransportKind::Multicast, SVC, INST, 1, RB); + assert!(tracker.saturation_warned); + + // Subsequent overflows leave the flag true; the flag is what the + // implementation checks before emitting a fresh warn!. + tracker.check(addr(9002), TransportKind::Multicast, SVC, INST, 1, RB); + tracker.check(addr(9003), TransportKind::Multicast, SVC, INST, 1, RB); + assert!(tracker.saturation_warned); + } } From 7d3da6e35b8b9236e012c2063bae0d8f0782877d Mon Sep 17 00:00:00 2001 From: Justin Kovacich Date: Thu, 23 Apr 2026 12:01:12 -0400 Subject: [PATCH 03/12] phase 2: cover pending_responses saturation branch MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Commit cb1d0d1 added explicit capacity-error delivery when pending_responses.insert overflows — without the explicit Err send, the dropped Sender would cause PendingResponse::response().await to panic on RecvError instead of surfacing a clean Result. The recovery logic was inline in the SendToService run-loop arm, which made it hard to exercise without driving 64 live sockets. Lift it to a small `track_or_reject_pending_response` helper on Inner so the SendToService arm delegates, and the branch is testable against the exposed `pending_responses` map directly. Added two tests: - `track_or_reject_pending_response_inserts_when_room_available`: happy path — entry lands in the map, the sender is still live (receiver stays pending). - `track_or_reject_pending_response_rejects_on_saturation`: fill map to PENDING_RESPONSES_CAP, invoke the helper with one more request, assert the map is unchanged and the caller's receiver resolves to Err(Error::Capacity("pending_responses")) — which is the invariant cb1d0d1 guarantees. Co-Authored-By: Claude Opus 4.7 --- src/client/inner.rs | 124 +++++++++++++++++++++++++++++++++++++------- 1 file changed, 105 insertions(+), 19 deletions(-) diff --git a/src/client/inner.rs b/src/client/inner.rs index 0b49bfcc..4be8b46d 100644 --- a/src/client/inner.rs +++ b/src/client/inner.rs @@ -431,6 +431,31 @@ where Ok(bound_port) } + /// Tracks the caller's response channel against `request_id` so a + /// future unicast reply can be routed back. If the + /// `pending_responses` map is already at `PENDING_RESPONSES_CAP`, the + /// `response` sender is recovered from the failed `insert` and used + /// to deliver `Err(Error::Capacity("pending_responses"))` — the + /// caller's `PendingResponse::response().await` resolves cleanly + /// instead of panicking on the `RecvError` that dropping the Sender + /// would have produced. Any reply that later arrives for a dropped + /// `request_id` is surfaced on the update stream via + /// `ClientUpdate::Unicast` instead of matching a pending entry. + fn track_or_reject_pending_response( + &mut self, + request_id: u32, + response: oneshot::Sender>, + ) { + if let Err((_req_id, response)) = self.pending_responses.insert(request_id, response) { + warn!( + "pending_responses at capacity ({}); response tracking \ + dropped for request_id 0x{:08X}", + PENDING_RESPONSES_CAP, request_id + ); + let _ = response.send(Err(Error::Capacity("pending_responses"))); + } + } + async fn receive_discovery( socket_manager: &mut Option>, ) -> Result< @@ -684,25 +709,7 @@ where match send_result { Ok(()) => { let _ = send_complete.send(Ok(())); - if let Err((_req_id, response)) = - self.pending_responses.insert(request_id, response) - { - // Map full: the send already succeeded, but - // we cannot track the reply. Deliver an - // explicit capacity error through the - // returned response sender rather than - // dropping it — otherwise `.expect(...)` on - // the receiver side would panic. Any reply - // that later arrives for `request_id` is - // delivered as `ClientUpdate::Unicast` on - // the update stream instead. - warn!( - "pending_responses at capacity ({}); response tracking \ - dropped for request_id 0x{:08X}", - PENDING_RESPONSES_CAP, request_id - ); - let _ = response.send(Err(Error::Capacity("pending_responses"))); - } + self.track_or_reject_pending_response(request_id, response); } Err(e) => { let _ = send_complete.send(Err(e)); @@ -1170,6 +1177,85 @@ mod tests { ); } + /// Happy path: with room in `pending_responses`, the helper tracks + /// the entry and does NOT signal the caller — the sender stays + /// alive so a future unicast reply can resolve it. + #[tokio::test] + async fn track_or_reject_pending_response_inserts_when_room_available() { + let mut inner = make_inner_for_test(); + let (tx, mut rx) = oneshot::channel::>(); + + inner.track_or_reject_pending_response(0xDEAD_BEEF, tx); + + assert_eq!(inner.pending_responses.len(), 1); + assert!( + inner.pending_responses.contains_key(&0xDEAD_BEEF), + "entry should be keyed by the provided request_id", + ); + // Receiver is still waiting — helper did NOT pre-emptively + // resolve it with a capacity error on the happy path. + assert!( + matches!(rx.try_recv(), Err(oneshot::error::TryRecvError::Empty)), + "receiver must still be pending when the insert succeeds", + ); + } + + /// Regression guard against cb1d0d1: without explicit rejection, + /// the dropped Sender would cause `PendingResponse::response()` to + /// panic on `RecvError` rather than returning a clean + /// `Err(Error::Capacity("pending_responses"))`. Exercises the + /// overflow branch in `track_or_reject_pending_response`, which is + /// the same branch the `SendToService` run-loop arm now delegates + /// to. + #[tokio::test] + async fn track_or_reject_pending_response_rejects_on_saturation() { + let mut inner = make_inner_for_test(); + + // Fill the map to capacity with dummy oneshot senders. The + // receivers are stashed so the senders stay live (dropping the + // receiver would drop the sender via the channel disconnect). + let mut stashed: std::vec::Vec>> = + std::vec::Vec::with_capacity(PENDING_RESPONSES_CAP); + for i in 0..PENDING_RESPONSES_CAP { + let (tx, rx) = oneshot::channel::>(); + inner + .pending_responses + .insert(i as u32, tx) + .expect("filling under cap must succeed"); + stashed.push(rx); + } + assert_eq!(inner.pending_responses.len(), PENDING_RESPONSES_CAP); + + // One more entry — map is full, the helper must recover the + // sender from the failed insert and deliver an explicit + // capacity error on it. + let (overflow_tx, overflow_rx) = oneshot::channel::>(); + let overflow_key: u32 = 0xFFFF_FFFE; + inner.track_or_reject_pending_response(overflow_key, overflow_tx); + + // Map size unchanged — the overflow attempt was rejected, not + // silently dropping an existing entry. + assert_eq!( + inner.pending_responses.len(), + PENDING_RESPONSES_CAP, + "overflow must not evict existing entries", + ); + assert!( + !inner.pending_responses.contains_key(&overflow_key), + "overflowed key must not be in the map", + ); + + // The caller's receiver resolves to Err(Capacity), not a + // panicking RecvError — this is the invariant cb1d0d1 fixes. + let result = overflow_rx + .await + .expect("receiver should get the explicit Err, not RecvError from dropped Sender"); + match result { + Err(Error::Capacity(tag)) => assert_eq!(tag, "pending_responses"), + other => panic!("expected Err(Error::Capacity(\"pending_responses\")), got {other:?}"), + } + } + #[tokio::test] async fn test_inner_spawn_and_shutdown() { let (control_sender, mut update_receiver) = Inner::::spawn( From 540899d3bf06350d8f7a7c8b91a045590109be4b Mon Sep 17 00:00:00 2001 From: Justin Kovacich Date: Thu, 23 Apr 2026 14:15:10 -0400 Subject: [PATCH 04/12] client+server: harden re-enqueue + misc Copilot round-2 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Round-2 Copilot feedback on PR #76 caught 5 resolvable issues (6th rejected — see below): - src/client/inner.rs: the three `push_front(...).ok()` re-enqueue sites (SetInterface x2, SendSD, Subscribe) silently dropped the returned Err. Swapped each to `if let Err(rejected) = ... { ... rejected.reject_with_capacity("request_queue"); }`, matching the primary `push_back` overflow arm. These branches are defensive — by construction the slot we just popped is free — but a future refactor that changes queue usage would otherwise reintroduce the silent-drop-of-oneshot-senders regression cb1d0d1 was specifically written to prevent. - src/server/subscription_manager.rs: `list.push(...).ok()` on a freshly-allocated `SubscribersList` (first subscriber in a new event group) swapped to `.expect(...)` with a message naming the `SUBSCRIBERS_PER_GROUP >= 1` invariant. Tripwires a future cap-0 regression at test time instead of silently losing the only subscriber. - src/client/session.rs: reword the `SessionTracker` doc comment's reference to non-existent "module docs" — point directly at the `SESSION_CAP` constant instead. Rejected: src/client/error.rs `Capacity` variant as a breaking exhaustive-match break. This is consistent with the earlier decision recorded on #75/#80 to accept the breaking change rather than carry deprecated shims or wrap everything in `Error::Io` — the current release is a breaking version bump by design, and hiding the variant in `Io` would lose the lowercase-snake_case tag semantics the new error was specifically designed to carry. No production behavior changes — just defensive tripwires, docs. Co-Authored-By: Claude Opus 4.7 --- src/client/inner.rs | 53 +++++++++++++++++++++--------- src/client/session.rs | 2 +- src/server/subscription_manager.rs | 11 +++++-- 3 files changed, 47 insertions(+), 19 deletions(-) diff --git a/src/client/inner.rs b/src/client/inner.rs index 4be8b46d..d411bfa9 100644 --- a/src/client/inner.rs +++ b/src/client/inner.rs @@ -526,18 +526,31 @@ where self.interface ); self.unbind_discovery().await; - // Re-enqueue after pop — queue has a free slot. - self.request_queue + // Re-enqueue after pop. The slot we popped is free, + // so `push_front` should never fail here — but if a + // future refactor breaks that invariant, reject via + // the capacity path instead of silently dropping the + // response oneshot (matches the primary `push_back` + // overflow arm in the control-channel receiver). + if let Err(rejected) = self + .request_queue .push_front(ControlMessage::SetInterface(interface, response)) - .ok(); + { + error!("request_queue push_front failed after pop — invariant broken"); + rejected.reject_with_capacity("request_queue"); + } return; } if self.interface != interface { self.set_interface(interface); - // Re-enqueue after pop — queue has a free slot. - self.request_queue + // See re-enqueue note above. + if let Err(rejected) = self + .request_queue .push_front(ControlMessage::SetInterface(interface, response)) - .ok(); + { + error!("request_queue push_front failed after pop — invariant broken"); + rejected.reject_with_capacity("request_queue"); + } return; } info!("Binding to interface: {}", interface); @@ -572,12 +585,15 @@ where None => { match self.bind_discovery() { Ok(()) => { - // Re-enqueue after pop — queue has a free slot. - self.request_queue - .push_front(ControlMessage::SendSD( - target, header, response, - )) - .ok(); + // See re-enqueue note on SetInterface above. + if let Err(rejected) = self.request_queue.push_front( + ControlMessage::SendSD(target, header, response), + ) { + error!( + "request_queue push_front failed after pop — invariant broken" + ); + rejected.reject_with_capacity("request_queue"); + } } Err(e) => { error!( @@ -774,9 +790,9 @@ where match &mut self.discovery_socket { None => match self.bind_discovery() { Ok(()) => { - // Re-enqueue after pop — queue has a free slot. - self.request_queue - .push_front(ControlMessage::Subscribe { + // See re-enqueue note on SetInterface above. + if let Err(rejected) = + self.request_queue.push_front(ControlMessage::Subscribe { service_id, instance_id, major_version, @@ -785,7 +801,12 @@ where client_port, response, }) - .ok(); + { + error!( + "request_queue push_front failed after pop — invariant broken" + ); + rejected.reject_with_capacity("request_queue"); + } } Err(e) => { let _ = response.send(Err(e)); diff --git a/src/client/session.rs b/src/client/session.rs index 6dc9b72f..989779b3 100644 --- a/src/client/session.rs +++ b/src/client/session.rs @@ -53,7 +53,7 @@ pub enum SessionVerdict { /// positives when a sensor interleaves SD offers for multiple services /// with independent session counters on the same source address. /// -/// Capacity is bounded at compile time ([`SESSION_CAP`]); see module docs. +/// Capacity is bounded at compile time by [`SESSION_CAP`]. /// When the map is full, new sender entries are dropped with a `warn!` log /// and reboot detection for those senders is disabled. /// diff --git a/src/server/subscription_manager.rs b/src/server/subscription_manager.rs index c0e21b93..56b4c2fc 100644 --- a/src/server/subscription_manager.rs +++ b/src/server/subscription_manager.rs @@ -83,14 +83,21 @@ impl SubscriptionManager { // New event group — allocate the list and insert. let mut list = SubscribersList::new(); - // Pushing into an empty heapless::Vec with cap >= 1 cannot fail. + // The first push into an empty heapless::Vec cannot fail as long + // as SUBSCRIBERS_PER_GROUP >= 1 (enforced by the constant's + // definition). Use `expect` here — a future refactor setting the + // cap to 0 would trip this at test time instead of silently + // dropping the only subscriber for a new event group. list.push(Subscriber::new( subscriber_addr, service_id, instance_id, event_group_id, )) - .ok(); + .expect( + "new SubscribersList must accept the first subscriber; \ + SUBSCRIBERS_PER_GROUP must be >= 1", + ); if self.subscriptions.insert(key, list).is_err() { tracing::warn!( From d6cca3ae269f7934297cd454097381f972e04052 Mon Sep 17 00:00:00 2001 From: Justin Kovacich Date: Thu, 23 Apr 2026 14:46:44 -0400 Subject: [PATCH 05/12] chore(clippy): add Panics doc + safe cast in move_vec PR MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Address two clippy::pedantic warnings introduced by this branch's own commits: - missing_panics_doc on SubscriptionManager::subscribe — the heapless::Vec::push.expect on the first-insert path can only trip if SUBSCRIBERS_PER_GROUP is set to zero; document that. - cast_possible_truncation on 'i as u32' in the saturation test for pending_responses — use u32::try_from with an expect that documents why the 64-cap fits. --- src/client/inner.rs | 5 ++++- src/server/subscription_manager.rs | 6 ++++++ 2 files changed, 10 insertions(+), 1 deletion(-) diff --git a/src/client/inner.rs b/src/client/inner.rs index d411bfa9..284f04a1 100644 --- a/src/client/inner.rs +++ b/src/client/inner.rs @@ -1241,7 +1241,10 @@ mod tests { let (tx, rx) = oneshot::channel::>(); inner .pending_responses - .insert(i as u32, tx) + .insert( + u32::try_from(i).expect("PENDING_RESPONSES_CAP fits in u32"), + tx, + ) .expect("filling under cap must succeed"); stashed.push(rx); } diff --git a/src/server/subscription_manager.rs b/src/server/subscription_manager.rs index 56b4c2fc..16b076a5 100644 --- a/src/server/subscription_manager.rs +++ b/src/server/subscription_manager.rs @@ -34,6 +34,12 @@ impl SubscriptionManager { } /// Add a subscriber to an event group + /// + /// # Panics + /// + /// Panics if `SUBSCRIBERS_PER_GROUP == 0`, a compile-time constant that + /// must be at least one for a newly-allocated subscriber list to accept + /// its first entry. pub fn subscribe( &mut self, service_id: u16, From 951b02dc98f3b1670b06f6c59ce16b18d60037e3 Mon Sep 17 00:00:00 2001 From: Justin Kovacich Date: Fri, 24 Apr 2026 13:01:43 -0400 Subject: [PATCH 06/12] round-3: handle displaced-sender branch in pending_responses insert MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Copilot round-3 on PR #76 (comment 3138669839) caught a real gap in `track_or_reject_pending_response`: the helper only handled `Err((key, value))` (map-full + new-key), but `heapless::IndexMap:: insert` on an existing key returns `Ok(Some(old_sender))` and the old sender was silently dropped. If `request_id` is ever reused while an older pending entry is still live — the documented example is `session_counter` wrap-around — the caller awaiting the original request would see a `RecvError` on channel cancellation, which `PendingResponse::response()` turns into a panic. Reworked the `if let Err(...)` into a full `match` with all three arms: - `Ok(None)` — normal insert, nothing to do. - `Ok(Some(displaced))` — `warn!` that the slot was replaced, complete the displaced sender with `Err(Error::Capacity("pending_responses"))` so the original caller gets a clean `Result`, not a `RecvError` panic. - `Err((_req_id, r))` — existing saturation path (unchanged). Also updated the doc comment on `track_or_reject_pending_response` to describe the displacement contract. Coverage: new test `track_or_reject_pending_response_completes_displaced_sender` explicitly inserts the same key twice and asserts the first receiver resolves to `Err(Error::Capacity("pending_responses"))` before the second sender still sits pending in the map. Addresses Copilot comment 3138669839. Co-Authored-By: Claude Opus 4.7 (1M context) --- src/client/inner.rs | 85 +++++++++++++++++++++++++++++++++++++++------ 1 file changed, 75 insertions(+), 10 deletions(-) diff --git a/src/client/inner.rs b/src/client/inner.rs index 284f04a1..a1fe583d 100644 --- a/src/client/inner.rs +++ b/src/client/inner.rs @@ -438,21 +438,38 @@ where /// to deliver `Err(Error::Capacity("pending_responses"))` — the /// caller's `PendingResponse::response().await` resolves cleanly /// instead of panicking on the `RecvError` that dropping the Sender - /// would have produced. Any reply that later arrives for a dropped - /// `request_id` is surfaced on the update stream via - /// `ClientUpdate::Unicast` instead of matching a pending entry. + /// would have produced. If `request_id` is reused while an older + /// pending entry still exists (e.g. after a `session_counter` + /// wrap-around), the displaced sender is likewise completed with + /// `Err(Error::Capacity("pending_responses"))` rather than being + /// silently dropped — the caller awaiting the previous request + /// sees a clean error instead of a `RecvError` panic. Any reply + /// that later arrives for a dropped `request_id` is surfaced on + /// the update stream via `ClientUpdate::Unicast` instead of + /// matching a pending entry. fn track_or_reject_pending_response( &mut self, request_id: u32, response: oneshot::Sender>, ) { - if let Err((_req_id, response)) = self.pending_responses.insert(request_id, response) { - warn!( - "pending_responses at capacity ({}); response tracking \ - dropped for request_id 0x{:08X}", - PENDING_RESPONSES_CAP, request_id - ); - let _ = response.send(Err(Error::Capacity("pending_responses"))); + match self.pending_responses.insert(request_id, response) { + Ok(None) => {} + Ok(Some(displaced_response)) => { + warn!( + "pending_responses already contained request_id \ + 0x{:08X}; replacing existing pending response", + request_id + ); + let _ = displaced_response.send(Err(Error::Capacity("pending_responses"))); + } + Err((_req_id, response)) => { + warn!( + "pending_responses at capacity ({}); response tracking \ + dropped for request_id 0x{:08X}", + PENDING_RESPONSES_CAP, request_id + ); + let _ = response.send(Err(Error::Capacity("pending_responses"))); + } } } @@ -1280,6 +1297,54 @@ mod tests { } } + /// If a `request_id` is reused while an older pending entry is still + /// live (e.g. `session_counter` wrap-around), `insert` returns + /// `Ok(Some(old_sender))`. Without handling that case, the displaced + /// sender is dropped and the caller awaiting the original request + /// hits `RecvError` (which `PendingResponse::response()` treats as a + /// fatal panic). This test guards against that: the displaced + /// sender must be completed with + /// `Err(Error::Capacity("pending_responses"))` so the original + /// caller gets a clean `Result` instead of a panicking `RecvError`. + #[tokio::test] + async fn track_or_reject_pending_response_completes_displaced_sender() { + let mut inner = make_inner_for_test(); + let key: u32 = 0xCAFE_F00D; + + // First tracking: the sender lives in the map. + let (first_tx, first_rx) = oneshot::channel::>(); + inner.track_or_reject_pending_response(key, first_tx); + assert_eq!(inner.pending_responses.len(), 1); + + // Second tracking with the same key: displaces the first sender. + let (second_tx, mut second_rx) = oneshot::channel::>(); + inner.track_or_reject_pending_response(key, second_tx); + + // Map still has one entry — the second one replaced the first. + assert_eq!(inner.pending_responses.len(), 1); + assert!(inner.pending_responses.contains_key(&key)); + + // The original caller's receiver resolves to Err(Capacity) — not + // a dropped-sender RecvError. + let displaced_result = first_rx.await.expect( + "displaced sender must be completed with a real Err, \ + not dropped (which would produce RecvError)", + ); + match displaced_result { + Err(Error::Capacity(tag)) => assert_eq!(tag, "pending_responses"), + other => panic!("expected Err(Error::Capacity(\"pending_responses\")), got {other:?}"), + } + + // The new sender is still live and pending. + assert!( + matches!( + second_rx.try_recv(), + Err(oneshot::error::TryRecvError::Empty) + ), + "replacement sender must still be pending in the map", + ); + } + #[tokio::test] async fn test_inner_spawn_and_shutdown() { let (control_sender, mut update_receiver) = Inner::::spawn( From 073ee80a13f3aea175666b42dc176114002b1d49 Mon Sep 17 00:00:00 2001 From: Justin Kovacich Date: Fri, 24 Apr 2026 16:56:55 -0400 Subject: [PATCH 07/12] docs(error): flag variant additions on client::Error as breaking The old docstring explained why client::Error is not #[non_exhaustive] but didn't acknowledge that adding variants to a non-#[non_exhaustive] enum is itself a breaking change for downstream exhaustive matches. Rewrite the note to call that out explicitly and flag future #[non_exhaustive] as a planned breaking-release change, and record the new Capacity variant under the Unreleased Added section. Co-Authored-By: Claude Opus 4.7 (1M context) --- CHANGELOG.md | 4 ++++ src/client/error.rs | 17 ++++++++++++++--- 2 files changed, 18 insertions(+), 3 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 128fece8..02863532 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,10 @@ ## [Unreleased] +### Added + +- **`client::Error::Capacity(&'static str)`** — new variant returned when a fixed-capacity internal structure is full (e.g. `"unicast_sockets"`, `"udp_buffer"`). Because `client::Error` is not `#[non_exhaustive]`, this is a breaking change for downstream crates that match the enum exhaustively. + ### Changed - **`std` is now the default feature** — the crate enables `std` (with `thiserror` and `tracing`) by default. Users targeting `no_std` environments must set `default-features = false` in their `Cargo.toml`. diff --git a/src/client/error.rs b/src/client/error.rs index 45c984ac..1e5c3045 100644 --- a/src/client/error.rs +++ b/src/client/error.rs @@ -2,9 +2,20 @@ use thiserror::Error; /// Errors that can occur during SOME/IP client operations. /// -/// Not marked `#[non_exhaustive]` today: downstream crates that match on -/// this enum rely on exhaustiveness, and adding the attribute now would be -/// a silent breaking change. Revisit when a breaking release is planned. +/// # Stability +/// +/// This enum is **not** marked `#[non_exhaustive]`, so downstream crates +/// may currently match it exhaustively. That convenience comes with a +/// real cost: **any new variant added here is a breaking change** and +/// must be flagged in the changelog and reflected in the next SemVer +/// bump (pre-1.0, a minor bump is sufficient, but it still requires a +/// release-notes entry). The same is true of renaming or restructuring +/// existing variants. +/// +/// Marking this `#[non_exhaustive]` — so future additions become +/// non-breaking — is planned as part of an explicit breaking release; +/// until then, treat variant additions as breaking and plan the release +/// accordingly. #[derive(Error, Debug)] pub enum Error { /// A SOME/IP protocol-level error. From 34e1a05e7a9cd9c1fc14a56fcec8d1eb83998932 Mon Sep 17 00:00:00 2001 From: Justin Kovacich Date: Fri, 24 Apr 2026 17:53:34 -0400 Subject: [PATCH 08/12] server: NACK on Subscribe capacity overflow instead of false ACK MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit SubscriptionManager::subscribe previously returned () and logged a warn! on capacity failure, which let handle_sd_message send SubscribeAck to a client whose subscription had silently been dropped — the client would believe it was subscribed but never receive any events. - subscribe now returns Result<(), SubscribeError> with SubscribersPerGroupFull / EventGroupsFull variants; existing capacity/refresh semantics are unchanged. - handle_sd_message inspects the result and emits SubscribeNack with a reason derived from the SubscribeError variant on rejection. - EventPublisher::register_subscriber surfaces the same Result so external SD dispatchers can take the same corrective action. - SubscribeError is re-exported from server::mod. - Tests updated to consume the Result; subscription_manager overflow tests now assert the specific error variant. Co-Authored-By: Claude Opus 4.7 (1M context) --- src/server/event_publisher.rs | 45 +++++++++--------- src/server/mod.rs | 35 +++++++++++--- src/server/subscription_manager.rs | 74 ++++++++++++++++++++++++------ 3 files changed, 110 insertions(+), 44 deletions(-) diff --git a/src/server/event_publisher.rs b/src/server/event_publisher.rs index c32470f1..71131e9d 100644 --- a/src/server/event_publisher.rs +++ b/src/server/event_publisher.rs @@ -248,9 +248,9 @@ impl EventPublisher { instance_id: u16, event_group_id: u16, subscriber_addr: std::net::SocketAddrV4, - ) { + ) -> Result<(), crate::server::SubscribeError> { let mut mgr = self.subscriptions.write().await; - mgr.subscribe(service_id, instance_id, event_group_id, subscriber_addr); + mgr.subscribe(service_id, instance_id, event_group_id, subscriber_addr) } /// Remove a previously-registered subscriber from an event group. @@ -345,7 +345,7 @@ mod tests { // Add subscriber { let mut mgr = subscriptions.write().await; - mgr.subscribe(0x5B, 1, 0x01, recv_addr); + mgr.subscribe(0x5B, 1, 0x01, recv_addr).unwrap(); } let (publisher, _) = make_publisher(subscriptions).await; @@ -388,7 +388,7 @@ mod tests { { let mut mgr = subscriptions.write().await; - mgr.subscribe(0x5B, 1, 0x01, recv_addr); + mgr.subscribe(0x5B, 1, 0x01, recv_addr).unwrap(); } let (publisher, _) = make_publisher(subscriptions).await; @@ -422,8 +422,8 @@ mod tests { { let mut mgr = subscriptions.write().await; - mgr.subscribe(0x5B, 1, 0x01, addr1); - mgr.subscribe(0x5B, 1, 0x01, addr2); + mgr.subscribe(0x5B, 1, 0x01, addr1).unwrap(); + mgr.subscribe(0x5B, 1, 0x01, addr2).unwrap(); } let (publisher, _) = make_publisher(subscriptions).await; @@ -444,7 +444,8 @@ mod tests { 1, 0x01, SocketAddrV4::new(Ipv4Addr::new(127, 0, 0, 1), 9001), - ); + ) + .unwrap(); } assert!(publisher.has_subscribers(0x5B, 1, 0x01).await); @@ -466,7 +467,7 @@ mod tests { let (publisher, _) = make_publisher(Arc::clone(&subscriptions)).await; assert!(!publisher.has_subscribers(0x5B, 1, 0x01).await); - publisher.register_subscriber(0x5B, 1, 0x01, ADDR_A).await; + publisher.register_subscriber(0x5B, 1, 0x01, ADDR_A).await.unwrap(); assert!(publisher.has_subscribers(0x5B, 1, 0x01).await); assert_eq!(publisher.subscriber_count(0x5B, 1, 0x01).await, 1); } @@ -478,9 +479,9 @@ mod tests { // Simulate TTL refreshes — the same (tuple, addr) called repeatedly // must not grow the subscriber list. - publisher.register_subscriber(0x5B, 1, 0x01, ADDR_A).await; - publisher.register_subscriber(0x5B, 1, 0x01, ADDR_A).await; - publisher.register_subscriber(0x5B, 1, 0x01, ADDR_A).await; + publisher.register_subscriber(0x5B, 1, 0x01, ADDR_A).await.unwrap(); + publisher.register_subscriber(0x5B, 1, 0x01, ADDR_A).await.unwrap(); + publisher.register_subscriber(0x5B, 1, 0x01, ADDR_A).await.unwrap(); assert_eq!(publisher.subscriber_count(0x5B, 1, 0x01).await, 1); } @@ -490,8 +491,8 @@ mod tests { let subscriptions = Arc::new(RwLock::new(SubscriptionManager::new())); let (publisher, _) = make_publisher(Arc::clone(&subscriptions)).await; - publisher.register_subscriber(0x5B, 1, 0x01, ADDR_A).await; - publisher.register_subscriber(0x5B, 1, 0x02, ADDR_A).await; + publisher.register_subscriber(0x5B, 1, 0x01, ADDR_A).await.unwrap(); + publisher.register_subscriber(0x5B, 1, 0x02, ADDR_A).await.unwrap(); assert_eq!(publisher.subscriber_count(0x5B, 1, 0x01).await, 1); assert_eq!(publisher.subscriber_count(0x5B, 1, 0x02).await, 1); @@ -504,7 +505,7 @@ mod tests { let subscriptions = Arc::new(RwLock::new(SubscriptionManager::new())); let (publisher, _) = make_publisher(Arc::clone(&subscriptions)).await; - publisher.register_subscriber(0x5B, 1, 0x01, ADDR_A).await; + publisher.register_subscriber(0x5B, 1, 0x01, ADDR_A).await.unwrap(); assert!(publisher.has_subscribers(0x5B, 1, 0x01).await); publisher.remove_subscriber(0x5B, 1, 0x01, ADDR_A).await; @@ -517,9 +518,9 @@ mod tests { let subscriptions = Arc::new(RwLock::new(SubscriptionManager::new())); let (publisher, _) = make_publisher(Arc::clone(&subscriptions)).await; - publisher.register_subscriber(0x5B, 1, 0x01, ADDR_A).await; - publisher.register_subscriber(0x5B, 1, 0x01, ADDR_B).await; - publisher.register_subscriber(0x5B, 1, 0x01, ADDR_C).await; + publisher.register_subscriber(0x5B, 1, 0x01, ADDR_A).await.unwrap(); + publisher.register_subscriber(0x5B, 1, 0x01, ADDR_B).await.unwrap(); + publisher.register_subscriber(0x5B, 1, 0x01, ADDR_C).await.unwrap(); assert_eq!(publisher.subscriber_count(0x5B, 1, 0x01).await, 3); publisher.remove_subscriber(0x5B, 1, 0x01, ADDR_B).await; @@ -544,7 +545,7 @@ mod tests { assert_eq!(publisher.subscriber_count(0x5B, 1, 0x01).await, 0); // Register one subscriber, then remove a different address. - publisher.register_subscriber(0x5B, 1, 0x01, ADDR_A).await; + publisher.register_subscriber(0x5B, 1, 0x01, ADDR_A).await.unwrap(); publisher.remove_subscriber(0x5B, 1, 0x01, ADDR_B).await; assert_eq!(publisher.subscriber_count(0x5B, 1, 0x01).await, 1); @@ -558,8 +559,8 @@ mod tests { let subscriptions = Arc::new(RwLock::new(SubscriptionManager::new())); let (publisher, _) = make_publisher(Arc::clone(&subscriptions)).await; - publisher.register_subscriber(0x5B, 1, 0x01, ADDR_A).await; - publisher.register_subscriber(0x5B, 1, 0x01, ADDR_B).await; + publisher.register_subscriber(0x5B, 1, 0x01, ADDR_A).await.unwrap(); + publisher.register_subscriber(0x5B, 1, 0x01, ADDR_B).await.unwrap(); assert!(publisher.has_subscribers(0x5B, 1, 0x01).await); publisher.remove_subscriber(0x5B, 1, 0x01, ADDR_A).await; @@ -576,9 +577,9 @@ mod tests { let subscriptions = Arc::new(RwLock::new(SubscriptionManager::new())); let (publisher, _) = make_publisher(Arc::clone(&subscriptions)).await; - publisher.register_subscriber(0x5B, 1, 0x01, ADDR_A).await; + publisher.register_subscriber(0x5B, 1, 0x01, ADDR_A).await.unwrap(); publisher.remove_subscriber(0x5B, 1, 0x01, ADDR_A).await; - publisher.register_subscriber(0x5B, 1, 0x01, ADDR_A).await; + publisher.register_subscriber(0x5B, 1, 0x01, ADDR_A).await.unwrap(); assert_eq!(publisher.subscriber_count(0x5B, 1, 0x01).await, 1); } diff --git a/src/server/mod.rs b/src/server/mod.rs index 2bcc4b15..3a25e4ed 100644 --- a/src/server/mod.rs +++ b/src/server/mod.rs @@ -15,7 +15,7 @@ mod subscription_manager; pub use error::Error; pub use event_publisher::EventPublisher; pub use service_info::{EventGroupInfo, ServiceInfo}; -pub use subscription_manager::SubscriptionManager; +pub use subscription_manager::{SubscribeError, SubscriptionManager}; use sd_state::SdStateManager; @@ -583,16 +583,36 @@ impl Server { second_count, ) { let mut subs = self.subscriptions.write().await; - subs.subscribe( + let subscribe_result = subs.subscribe( entry_view.service_id(), entry_view.instance_id(), entry_view.event_group_id(), endpoint_addr, ); - - // Send SubscribeAck - self.send_subscribe_ack_from_view(&entry_view, sender) - .await?; + // Release the write lock before any await on the + // SD socket (keeps this arm off the lock while we + // emit the response). + drop(subs); + + match subscribe_result { + Ok(()) => { + self.send_subscribe_ack_from_view(&entry_view, sender) + .await?; + } + Err(e) => { + // Capacity-rejected subscription: NACK so + // the client doesn't believe it's + // subscribed. The warn! inside + // `subscribe` already logged the + // structure that was full. + self.send_subscribe_nack_from_view( + &entry_view, + sender, + &format!("subscription rejected: {e}"), + ) + .await?; + } + } } else { tracing::warn!("No endpoint found in Subscribe message options"); self.send_subscribe_nack_from_view( @@ -1878,7 +1898,8 @@ mod tests { let subscriber = SocketAddrV4::new(Ipv4Addr::new(127, 0, 0, 2), 40_000); publisher .register_subscriber(0x005C, 0x0001, 0x0001, subscriber) - .await; + .await + .unwrap(); assert!(publisher.has_subscribers(0x005C, 0x0001, 0x0001).await); assert_eq!(publisher.subscriber_count(0x005C, 0x0001, 0x0001).await, 1); diff --git a/src/server/subscription_manager.rs b/src/server/subscription_manager.rs index 16b076a5..4065df09 100644 --- a/src/server/subscription_manager.rs +++ b/src/server/subscription_manager.rs @@ -12,6 +12,35 @@ const EVENT_GROUPS_CAP: usize = 32; /// with a `warn!` log rather than silently. const SUBSCRIBERS_PER_GROUP: usize = 16; +/// Why a call to [`SubscriptionManager::subscribe`] failed to record a new +/// subscriber. Callers (typically the server's `Subscribe` handler) should +/// use this to emit a `SubscribeNack` instead of a misleading `SubscribeAck`. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum SubscribeError { + /// The per-event-group subscriber list is already full + /// ([`SUBSCRIBERS_PER_GROUP`] entries). The caller's request was not + /// recorded. + SubscribersPerGroupFull, + /// The outer event-group map is already full ([`EVENT_GROUPS_CAP`] + /// distinct `(service_id, instance_id, event_group_id)` keys). The + /// caller's request was not recorded. + EventGroupsFull, +} + +impl core::fmt::Display for SubscribeError { + fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result { + match self { + Self::SubscribersPerGroupFull => write!( + f, + "subscribers-per-group at capacity ({SUBSCRIBERS_PER_GROUP})" + ), + Self::EventGroupsFull => { + write!(f, "event-group map at capacity ({EVENT_GROUPS_CAP})") + } + } + } +} + type SubscribersList = HeaplessVec; /// Manages subscriptions to event groups. @@ -33,7 +62,13 @@ impl SubscriptionManager { } } - /// Add a subscriber to an event group + /// Add a subscriber to an event group. + /// + /// Returns `Ok(())` for a new or refreshed (deduplicated) subscription. + /// Returns `Err(SubscribeError)` when the request could not be recorded + /// because a bounded capacity was hit — the caller (typically the + /// server's `Subscribe` handler) should send a `SubscribeNack` on + /// `Err`, not a `SubscribeAck`. /// /// # Panics /// @@ -46,7 +81,7 @@ impl SubscriptionManager { instance_id: u16, event_group_id: u16, subscriber_addr: SocketAddrV4, - ) { + ) -> Result<(), SubscribeError> { let key = (service_id, instance_id, event_group_id); if let Some(subscribers) = self.subscriptions.get_mut(&key) { @@ -59,7 +94,7 @@ impl SubscriptionManager { instance_id, event_group_id ); - return; + return Ok(()); } let subscriber = @@ -74,7 +109,7 @@ impl SubscriptionManager { instance_id, event_group_id ); - return; + return Err(SubscribeError::SubscribersPerGroupFull); } tracing::info!( @@ -84,7 +119,7 @@ impl SubscriptionManager { instance_id, event_group_id ); - return; + return Ok(()); } // New event group — allocate the list and insert. @@ -115,7 +150,7 @@ impl SubscriptionManager { instance_id, event_group_id ); - return; + return Err(SubscribeError::EventGroupsFull); } tracing::info!( @@ -125,6 +160,7 @@ impl SubscriptionManager { instance_id, event_group_id ); + Ok(()) } /// Remove a subscriber from an event group @@ -193,7 +229,7 @@ mod tests { let addr = SocketAddrV4::new(Ipv4Addr::new(192, 168, 1, 1), 8080); // Subscribe - manager.subscribe(0x5B, 1, 0x01, addr); + manager.subscribe(0x5B, 1, 0x01, addr).unwrap(); assert_eq!(manager.subscription_count(), 1); // Get subscribers @@ -211,11 +247,11 @@ mod tests { let mut manager = SubscriptionManager::new(); let addr = SocketAddrV4::new(Ipv4Addr::new(192, 168, 1, 1), 8080); - manager.subscribe(0x5B, 1, 0x01, addr); + manager.subscribe(0x5B, 1, 0x01, addr).unwrap(); assert_eq!(manager.subscription_count(), 1); // Subscribe same address again — should deduplicate - manager.subscribe(0x5B, 1, 0x01, addr); + manager.subscribe(0x5B, 1, 0x01, addr).unwrap(); assert_eq!(manager.subscription_count(), 1); } @@ -248,13 +284,17 @@ mod tests { for i in 0..SUBSCRIBERS_PER_GROUP { let addr = SocketAddrV4::new(Ipv4Addr::new(10, 0, 0, 1), 8000 + u16::try_from(i).unwrap()); - manager.subscribe(0x5B, 1, 0x01, addr); + manager.subscribe(0x5B, 1, 0x01, addr).unwrap(); } assert_eq!(manager.subscription_count(), SUBSCRIBERS_PER_GROUP); - // One more is dropped. + // One more is dropped, and the call reports SubscribersPerGroupFull + // so the server can NACK. let extra = SocketAddrV4::new(Ipv4Addr::new(10, 0, 0, 1), 9999); - manager.subscribe(0x5B, 1, 0x01, extra); + assert_eq!( + manager.subscribe(0x5B, 1, 0x01, extra), + Err(SubscribeError::SubscribersPerGroupFull), + ); assert_eq!(manager.subscription_count(), SUBSCRIBERS_PER_GROUP); // Extra subscriber should not appear in the list. let subs = manager.get_subscribers(0x5B, 1, 0x01); @@ -268,13 +308,17 @@ mod tests { // Fill the outer map to capacity with distinct event groups. for i in 0..EVENT_GROUPS_CAP { let eg = u16::try_from(i).unwrap(); - manager.subscribe(0x5B, 1, eg, addr); + manager.subscribe(0x5B, 1, eg, addr).unwrap(); } assert_eq!(manager.subscription_count(), EVENT_GROUPS_CAP); - // A new event group beyond capacity is dropped. + // A new event group beyond capacity is dropped, and the call reports + // EventGroupsFull so the server can NACK. let overflow_eg = u16::try_from(EVENT_GROUPS_CAP).unwrap(); - manager.subscribe(0x5B, 1, overflow_eg, addr); + assert_eq!( + manager.subscribe(0x5B, 1, overflow_eg, addr), + Err(SubscribeError::EventGroupsFull), + ); assert_eq!(manager.subscription_count(), EVENT_GROUPS_CAP); assert!(manager.get_subscribers(0x5B, 1, overflow_eg).is_empty()); } From 97eca582da7a52892cd072f2fc7b4be06c33652a Mon Sep 17 00:00:00 2001 From: Justin Kovacich Date: Fri, 24 Apr 2026 17:59:29 -0400 Subject: [PATCH 09/12] round-4: address Copilot feedback on session/sub_manager docs + log wording MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - session::check: expand the doc comment with an explicit "Capacity behavior" section — the saturation path does not update stored state for new keys, which contradicts the old "Always updates the stored state" wording. Describe the actual behavior (normal path updates, new-key inserts under saturation are silently dropped and return Initial repeatedly) so callers don't rely on a strict guarantee the implementation doesn't hold. - session::check saturation warn!: include sender + transport in the log line so diagnosing which peer lost reboot-detection state no longer requires out-of-band correlation. - SubscriptionManager::subscribe dedup branch: rename the log from "Refreshed existing subscriber" to "already subscribed; skipping" and add a comment making it explicit that no per-subscriber state is modified. The old wording implied a refresh (TTL bump, etc) the code never performed. - inner.rs test comment: tokio::sync::oneshot doesn't drop the sender when the receiver is dropped — it flips send() to a failure mode. Rewrite the comment to describe the actual stash purpose (keep channels open for later observation). Co-Authored-By: Claude Opus 4.7 (1M context) --- src/client/inner.rs | 9 +++++++-- src/client/session.rs | 26 +++++++++++++++++++++++--- src/server/subscription_manager.rs | 9 +++++++-- 3 files changed, 37 insertions(+), 7 deletions(-) diff --git a/src/client/inner.rs b/src/client/inner.rs index a1fe583d..9c41526a 100644 --- a/src/client/inner.rs +++ b/src/client/inner.rs @@ -1250,8 +1250,13 @@ mod tests { let mut inner = make_inner_for_test(); // Fill the map to capacity with dummy oneshot senders. The - // receivers are stashed so the senders stay live (dropping the - // receiver would drop the sender via the channel disconnect). + // receivers are stashed to keep each channel open for the + // remainder of the test — on `tokio::sync::oneshot`, dropping + // the receiver does not drop the sender; it flips the sender + // into a state where `send()` fails with the value returned. + // The stash is what lets us later observe `sender.send(...)` + // succeeding against a still-open channel when the overflow + // case completes the displaced sender with a capacity error. let mut stashed: std::vec::Vec>> = std::vec::Vec::with_capacity(PENDING_RESPONSES_CAP); for i in 0..PENDING_RESPONSES_CAP { diff --git a/src/client/session.rs b/src/client/session.rs index 989779b3..46399890 100644 --- a/src/client/session.rs +++ b/src/client/session.rs @@ -85,7 +85,24 @@ impl Default for SessionTracker { impl SessionTracker { /// Check the session ID and reboot flag for a specific service instance - /// and return a verdict. Always updates the stored state after the check. + /// and return a verdict. + /// + /// On the normal (non-saturated) path, the stored state is updated + /// after the check so subsequent calls see the latest session id and + /// reboot flag for the key. + /// + /// # Capacity behavior + /// + /// The tracker is backed by a `heapless::FnvIndexMap` bounded by + /// [`SESSION_CAP`]. If the map is already full and the incoming key + /// is new, the insert fails and stored state is **not** updated for + /// that key — subsequent `check()` calls with the same new key will + /// continue to return [`SessionVerdict::Initial`] until an existing + /// key is evicted or capacity is raised. A single `warn!` fires the + /// first time saturation is hit (further saturation drops are + /// suppressed to avoid log spam at the packet rate). For existing + /// keys under saturation the update still succeeds, because + /// `FnvIndexMap::insert` replaces in place. /// /// Call this once per service entry in an SD message (not once per message), /// so each service instance gets its own session counter. @@ -135,9 +152,12 @@ impl SessionTracker { if !self.saturation_warned { tracing::warn!( "SessionTracker at capacity ({}); dropping new sender state for \ - svc=0x{:04X} inst=0x{:04X}. Reboot detection disabled for this \ - entry and any further new entries (subsequent drops not logged).", + sender={} transport={:?} svc=0x{:04X} inst=0x{:04X}. Reboot \ + detection disabled for this entry and any further new entries \ + (subsequent drops not logged).", SESSION_CAP, + sender, + transport, service_id, instance_id ); diff --git a/src/server/subscription_manager.rs b/src/server/subscription_manager.rs index 4065df09..4114b95b 100644 --- a/src/server/subscription_manager.rs +++ b/src/server/subscription_manager.rs @@ -85,10 +85,15 @@ impl SubscriptionManager { let key = (service_id, instance_id, event_group_id); if let Some(subscribers) = self.subscriptions.get_mut(&key) { - // Deduplicate: if this address is already subscribed, just refresh (don't add again) + // Deduplicate: if this address is already subscribed, skip adding + // it again. No stored subscriber state is modified — the log + // message reflects that. If real refresh semantics (e.g. TTL + // bump on re-subscribe) are wanted later, update the per- + // subscriber record here and rename the log accordingly. if subscribers.iter().any(|s| s.address == subscriber_addr) { tracing::debug!( - "Refreshed existing subscriber {} for service 0x{:04X}, instance {}, event group 0x{:04X}", + "Subscriber {} already subscribed for service 0x{:04X}, instance {}, \ + event group 0x{:04X}; skipping duplicate", subscriber_addr, service_id, instance_id, From cea2204f47ee4af417ad255209149687c70eff8e Mon Sep 17 00:00:00 2001 From: Justin Kovacich Date: Fri, 24 Apr 2026 18:10:20 -0400 Subject: [PATCH 10/12] round-5: clean up Subscribe-NACK reason path + CHANGELOG + doc - server::run NACK arm: stop allocating a String with format!() held across the SubscribeNack await; log the SubscribeError at warn! separately and pass a static "subscription rejected" reason string to send_subscribe_nack_from_view. Cleaner and avoids the borrow- across-await style issue Copilot flagged. - EventPublisher::register_subscriber: add # Errors section describing the two SubscribeError variants, that the subscriber is NOT registered on Err, and that external dispatchers should NACK on Err just like the server's own run() loop does. - CHANGELOG: add server::SubscribeError to Added, and list the breaking signature changes on SubscriptionManager::subscribe and EventPublisher::register_subscriber under Changed. Co-Authored-By: Claude Opus 4.7 (1M context) --- CHANGELOG.md | 3 +++ src/server/event_publisher.rs | 17 +++++++++++++++++ src/server/mod.rs | 12 +++++++++--- 3 files changed, 29 insertions(+), 3 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 02863532..0bb15fce 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,11 +5,14 @@ ### Added - **`client::Error::Capacity(&'static str)`** — new variant returned when a fixed-capacity internal structure is full (e.g. `"unicast_sockets"`, `"udp_buffer"`). Because `client::Error` is not `#[non_exhaustive]`, this is a breaking change for downstream crates that match the enum exhaustively. +- **`server::SubscribeError`** — new public enum (`SubscribersPerGroupFull`, `EventGroupsFull`) returned by `SubscriptionManager::subscribe` and `EventPublisher::register_subscriber` when a bounded capacity rejects a subscription. Re-exported from `server::mod`. ### Changed - **`std` is now the default feature** — the crate enables `std` (with `thiserror` and `tracing`) by default. Users targeting `no_std` environments must set `default-features = false` in their `Cargo.toml`. - **`thiserror` and `tracing` use `default-features = false`** — both dependencies are always included but their `std` features are only enabled when the crate's `std` feature is active. This removes the need for `#[cfg(feature = "std")]` gating on error types and logging macros. +- **Breaking: `server::SubscriptionManager::subscribe` signature change** — now returns `Result<(), server::SubscribeError>` instead of `()`. Previously, capacity rejections were silently dropped with only a `warn!` log, which let the server emit a `SubscribeAck` for a subscription that had not been recorded. Callers must now handle the `Err` path (the server's own SD loop emits `SubscribeNack` on `Err`). +- **Breaking: `server::EventPublisher::register_subscriber` signature change** — now returns `Result<(), server::SubscribeError>` instead of `()`, surfacing the same capacity-rejection signal to externally managed subscription dispatchers. ## [0.6.0](https://github.com/luminartech/simple_someip/compare/v0.5.3...v0.6.0) - 2026-04-20 diff --git a/src/server/event_publisher.rs b/src/server/event_publisher.rs index 71131e9d..caecdb65 100644 --- a/src/server/event_publisher.rs +++ b/src/server/event_publisher.rs @@ -242,6 +242,23 @@ impl EventPublisher { /// `remove_subscriber` when no refresh has arrived within the /// advertised TTL — otherwise subscribers accumulate for the /// lifetime of the process. + /// + /// # Errors + /// + /// Returns [`crate::server::SubscribeError`] when the underlying + /// [`SubscriptionManager`] cannot record the subscription because a + /// bounded capacity was hit: + /// - `SubscribersPerGroupFull` — the per-event-group subscriber list + /// is full. + /// - `EventGroupsFull` — the outer event-group map is full. + /// + /// On `Err`, the subscriber was **not** registered and no events + /// will be delivered to `subscriber_addr` for this event group. + /// External dispatchers should treat this the same way the server's + /// own `run()` loop does: emit a `SubscribeNack` (or equivalent + /// upstream notification) so the peer does not assume it is + /// subscribed. A duplicate registration for an already-subscribed + /// address returns `Ok(())` (deduplicated). pub async fn register_subscriber( &self, service_id: u16, diff --git a/src/server/mod.rs b/src/server/mod.rs index 3a25e4ed..97f5f15d 100644 --- a/src/server/mod.rs +++ b/src/server/mod.rs @@ -603,12 +603,18 @@ impl Server { // Capacity-rejected subscription: NACK so // the client doesn't believe it's // subscribed. The warn! inside - // `subscribe` already logged the - // structure that was full. + // `subscribe` already logged which + // structure was full; re-emit the + // SubscribeError at warn! here so the + // NACK and its specific cause are + // correlated in the same log line + // without allocating a String that + // would need to live across the await. + tracing::warn!("Subscription rejected: {e}"); self.send_subscribe_nack_from_view( &entry_view, sender, - &format!("subscription rejected: {e}"), + "subscription rejected", ) .await?; } From 5a804cf40d9c6102d116d7c2b37a5b8522191ee3 Mon Sep 17 00:00:00 2001 From: Justin Kovacich Date: Fri, 24 Apr 2026 18:32:14 -0400 Subject: [PATCH 11/12] PR #76 round: trim stale changelog, specific NACK reasons, truthful subscribe docs - CHANGELOG.md: drop the two "Changed" bullets about `std` becoming the default feature and `thiserror`/`tracing` moving to `default-features = false`. Those landed in 0.6.0/0.6.1 (commit f161980, already on main) and are not changes made in this PR. - server/mod.rs: when a SubscribeError rejects a subscription, match on the variant and pass a specific `&'static str` reason to `send_subscribe_nack_from_view` (`"subscribers_per_group_full"` / `"event_groups_full"`) instead of the generic `"subscription rejected"`. The NACK log line now reflects the real cause, and the static-str choice avoids any String-across-await allocation. - server/subscription_manager.rs: rewrite the `subscribe` docs to say the duplicate path is idempotent / deduplicated rather than implying TTL-refresh semantics that don't exist today. Flag the future-TTL extension point explicitly so the doc and the log wording stay in sync if that behavior is added later. Co-Authored-By: Claude Opus 4.7 (1M context) --- CHANGELOG.md | 2 -- src/server/mod.rs | 26 ++++++++++++++++---------- src/server/subscription_manager.rs | 17 ++++++++++++----- 3 files changed, 28 insertions(+), 17 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 0bb15fce..87dbc732 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,8 +9,6 @@ ### Changed -- **`std` is now the default feature** — the crate enables `std` (with `thiserror` and `tracing`) by default. Users targeting `no_std` environments must set `default-features = false` in their `Cargo.toml`. -- **`thiserror` and `tracing` use `default-features = false`** — both dependencies are always included but their `std` features are only enabled when the crate's `std` feature is active. This removes the need for `#[cfg(feature = "std")]` gating on error types and logging macros. - **Breaking: `server::SubscriptionManager::subscribe` signature change** — now returns `Result<(), server::SubscribeError>` instead of `()`. Previously, capacity rejections were silently dropped with only a `warn!` log, which let the server emit a `SubscribeAck` for a subscription that had not been recorded. Callers must now handle the `Err` path (the server's own SD loop emits `SubscribeNack` on `Err`). - **Breaking: `server::EventPublisher::register_subscriber` signature change** — now returns `Result<(), server::SubscribeError>` instead of `()`, surfacing the same capacity-rejection signal to externally managed subscription dispatchers. diff --git a/src/server/mod.rs b/src/server/mod.rs index 97f5f15d..3e2651ff 100644 --- a/src/server/mod.rs +++ b/src/server/mod.rs @@ -602,19 +602,25 @@ impl Server { Err(e) => { // Capacity-rejected subscription: NACK so // the client doesn't believe it's - // subscribed. The warn! inside - // `subscribe` already logged which - // structure was full; re-emit the - // SubscribeError at warn! here so the - // NACK and its specific cause are - // correlated in the same log line - // without allocating a String that - // would need to live across the await. - tracing::warn!("Subscription rejected: {e}"); + // subscribed. Match on the specific + // SubscribeError so the NACK log line + // carries the actual cause (which + // bounded structure was full) rather + // than the generic "subscription + // rejected" string — and pick static + // reason strings so no allocation has + // to live across the await. + let reason: &'static str = match e { + SubscribeError::SubscribersPerGroupFull => { + "subscribers_per_group_full" + } + SubscribeError::EventGroupsFull => "event_groups_full", + }; + tracing::warn!("Subscription rejected: {reason}"); self.send_subscribe_nack_from_view( &entry_view, sender, - "subscription rejected", + reason, ) .await?; } diff --git a/src/server/subscription_manager.rs b/src/server/subscription_manager.rs index 4114b95b..ca181c52 100644 --- a/src/server/subscription_manager.rs +++ b/src/server/subscription_manager.rs @@ -64,11 +64,18 @@ impl SubscriptionManager { /// Add a subscriber to an event group. /// - /// Returns `Ok(())` for a new or refreshed (deduplicated) subscription. - /// Returns `Err(SubscribeError)` when the request could not be recorded - /// because a bounded capacity was hit — the caller (typically the - /// server's `Subscribe` handler) should send a `SubscribeNack` on - /// `Err`, not a `SubscribeAck`. + /// Returns `Ok(())` both when a new subscriber is added and when the + /// given `(service_id, instance_id, event_group_id, subscriber_addr)` + /// is already subscribed — the call is idempotent / deduplicated, and + /// no stored subscriber state is modified on a duplicate. There is no + /// TTL bump or other refresh side-effect today; if TTL-refresh + /// semantics are added later, this docstring and the duplicate-log + /// wording will be updated together. + /// + /// Returns `Err(SubscribeError)` when the request could not be + /// recorded because a bounded capacity was hit — the caller + /// (typically the server's `Subscribe` handler) should send a + /// `SubscribeNack` on `Err`, not a `SubscribeAck`. /// /// # Panics /// From bceae7774104fce98d1d6dc008b119e33ff5c8dd Mon Sep 17 00:00:00 2001 From: Justin Kovacich <32140377+JustinKovacich@users.noreply.github.com> Date: Fri, 24 Apr 2026 18:37:16 -0400 Subject: [PATCH 12/12] Update src/server/mod.rs Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> --- src/server/mod.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/server/mod.rs b/src/server/mod.rs index 3e2651ff..b1293017 100644 --- a/src/server/mod.rs +++ b/src/server/mod.rs @@ -616,7 +616,7 @@ impl Server { } SubscribeError::EventGroupsFull => "event_groups_full", }; - tracing::warn!("Subscription rejected: {reason}"); + tracing::debug!("Subscription rejected: {reason}"); self.send_subscribe_nack_from_view( &entry_view, sender,