diff --git a/CHANGELOG.md b/CHANGELOG.md index 128fece8..87dbc732 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,10 +2,15 @@ ## [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. +- **`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/client/error.rs b/src/client/error.rs index 64af3814..1e5c3045 100644 --- a/src/client/error.rs +++ b/src/client/error.rs @@ -1,6 +1,21 @@ use thiserror::Error; /// Errors that can occur during SOME/IP client operations. +/// +/// # 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. @@ -24,4 +39,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..9c41526a 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>), @@ -232,16 +245,47 @@ 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 { /// 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 +293,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 +345,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,13 +403,76 @@ 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) } + /// 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. 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>, + ) { + 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"))); + } + } + } + async fn receive_discovery( socket_manager: &mut Option>, ) -> Result< @@ -400,7 +507,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 +543,31 @@ where self.interface ); self.unbind_discovery().await; - self.request_queue - .push_front(ControlMessage::SetInterface(interface, response)); + // 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)) + { + 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); - self.request_queue - .push_front(ControlMessage::SetInterface(interface, response)); + // See re-enqueue note above. + if let Err(rejected) = self + .request_queue + .push_front(ControlMessage::SetInterface(interface, response)) + { + error!("request_queue push_front failed after pop — invariant broken"); + rejected.reject_with_capacity("request_queue"); + } return; } info!("Binding to interface: {}", interface); @@ -474,10 +602,15 @@ 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, - )); + // 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!( @@ -609,7 +742,7 @@ where match send_result { Ok(()) => { let _ = send_complete.send(Ok(())); - self.pending_responses.insert(request_id, response); + self.track_or_reject_pending_response(request_id, response); } Err(e) => { let _ = send_complete.send(Err(e)); @@ -674,15 +807,23 @@ 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, - }); + // See re-enqueue note on SetInterface above. + if let Err(rejected) = + self.request_queue.push_front(ControlMessage::Subscribe { + service_id, + instance_id, + major_version, + ttl, + event_group_id, + client_port, + response, + }) + { + error!( + "request_queue push_front failed after pop — invariant broken" + ); + rejected.reject_with_capacity("request_queue"); + } } Err(e) => { let _ = response.send(Err(e)); @@ -746,7 +887,19 @@ where ctrl = control_receiver.recv() => { if let Some(ctrl) = ctrl { debug!("Received control message: {:?}", ctrl); - request_queue.push_back(ctrl); + 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 ({}); rejecting control message", + REQUEST_QUEUE_CAP + ); + rejected.reject_with_capacity("request_queue"); + } } else { // The sender has been dropped, so we should exit *run = false; @@ -906,6 +1059,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); @@ -946,6 +1158,198 @@ 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" + ); + } + + /// 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 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 { + let (tx, rx) = oneshot::channel::>(); + inner + .pending_responses + .insert( + u32::try_from(i).expect("PENDING_RESPONSES_CAP fits in 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:?}"), + } + } + + /// 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( diff --git a/src/client/mod.rs b/src/client/mod.rs index 026f2b74..8cb2cb89 100644 --- a/src/client/mod.rs +++ b/src/client/mod.rs @@ -1,3 +1,18 @@ +//! SOME/IP client. +//! +//! # Memory footprint +//! +//! The client's internal `Inner` state is allocated inline rather than on +//! 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); these capacity constants are +//! the primary knobs for trimming this footprint. mod error; mod inner; mod service_registry; @@ -550,6 +565,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(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 this error 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..46399890 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,14 +52,57 @@ 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 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. +/// +/// # 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, + /// 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, + } + } } 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. @@ -89,13 +139,31 @@ 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 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 \ + 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 + ); + self.saturation_warned = true; + } + } verdict } } @@ -310,4 +378,57 @@ 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); + } + + #[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); + } } diff --git a/src/server/event_publisher.rs b/src/server/event_publisher.rs index c32470f1..caecdb65 100644 --- a/src/server/event_publisher.rs +++ b/src/server/event_publisher.rs @@ -242,15 +242,32 @@ 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, 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 +362,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 +405,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 +439,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 +461,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 +484,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 +496,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 +508,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 +522,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 +535,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 +562,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 +576,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 +594,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..b1293017 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,48 @@ 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. 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::debug!("Subscription rejected: {reason}"); + self.send_subscribe_nack_from_view( + &entry_view, + sender, + reason, + ) + .await?; + } + } } else { tracing::warn!("No endpoint found in Subscribe message options"); self.send_subscribe_nack_from_view( @@ -1878,7 +1910,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 28667bc8..ca181c52 100644 --- a/src/server/subscription_manager.rs +++ b/src/server/subscription_manager.rs @@ -1,13 +1,56 @@ //! 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; + +/// 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. +/// +/// 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,35 +58,112 @@ impl SubscriptionManager { #[must_use] pub fn new() -> Self { Self { - subscriptions: HashMap::new(), + subscriptions: FnvIndexMap::new(), } } - /// Add a subscriber to an event group + /// Add a subscriber to an event group. + /// + /// 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 + /// + /// 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, instance_id: u16, event_group_id: u16, subscriber_addr: SocketAddrV4, - ) { + ) -> Result<(), SubscribeError> { 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, 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!( + "Subscriber {} already subscribed for service 0x{:04X}, instance {}, \ + event group 0x{:04X}; skipping duplicate", + subscriber_addr, + service_id, + instance_id, + event_group_id + ); + return Ok(()); + } + + 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 Err(SubscribeError::SubscribersPerGroupFull); + } + + tracing::info!( + "Subscriber {} added for service 0x{:04X}, instance {}, event group 0x{:04X}", subscriber_addr, service_id, instance_id, event_group_id ); - return; + return Ok(()); } - 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(); + // 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, + )) + .expect( + "new SubscribersList must accept the first subscriber; \ + SUBSCRIBERS_PER_GROUP must be >= 1", + ); + + 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 Err(SubscribeError::EventGroupsFull); + } tracing::info!( "Subscriber {} added for service 0x{:04X}, instance {}, event group 0x{:04X}", @@ -52,6 +172,7 @@ impl SubscriptionManager { instance_id, event_group_id ); + Ok(()) } /// Remove a subscriber from an event group @@ -90,13 +211,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() } } @@ -117,7 +241,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 @@ -135,11 +259,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); } @@ -164,4 +288,50 @@ 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).unwrap(); + } + assert_eq!(manager.subscription_count(), SUBSCRIBERS_PER_GROUP); + + // 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); + 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); + 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).unwrap(); + } + assert_eq!(manager.subscription_count(), EVENT_GROUPS_CAP); + + // 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(); + 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()); + } }