diff --git a/CHANGELOG.md b/CHANGELOG.md index 8f905c1c..f54fc90a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,16 @@ ## [Unreleased] +## [0.7.3](https://github.com/luminartech/simple_someip/compare/v0.7.2...v0.7.3) - 2026-06-23 + +### Added + +- *(client)* `ClientUpdate::Unicast` now carries the sender's `source: SocketAddr`. The SOME/IP header has no instance id, so on a shared subnet the source address is the only way to attribute a unicast event to a specific device. The source was already received but previously discarded before forwarding. + +### Fixed + +- *(e2e)* Receive E2E counter state is now keyed per source address. Previously a single per-`(service, method)` counter was shared across all senders, so several devices sending the same event under one fixed instance id collided their interleaved counters into spurious `WrongSequence` results. Profile configuration stays endpoint-agnostic; transmit (`protect`) state is unchanged. A rebooted sender's receive state is reset so its first post-reboot frame is not flagged out-of-sequence. + ## [0.7.2](https://github.com/luminartech/simple_someip/compare/v0.7.1...v0.7.2) - 2026-06-23 ### Added diff --git a/Cargo.lock b/Cargo.lock index c095ec5d..379287ba 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -154,7 +154,7 @@ dependencies = [ [[package]] name = "simple-someip" -version = "0.7.2" +version = "0.7.3" dependencies = [ "crc", "embedded-io", diff --git a/Cargo.toml b/Cargo.toml index 15543dba..06e4d236 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -3,7 +3,7 @@ members = [".", "examples/discovery_client", "examples/client_server"] [package] name = "simple-someip" -version = "0.7.2" +version = "0.7.3" edition = "2024" license = "MIT OR Apache-2.0" description = "A lightweight SOME/IP serialization and communication library" diff --git a/src/client/inner.rs b/src/client/inner.rs index e28a8cc1..d9ccbfa3 100644 --- a/src/client/inner.rs +++ b/src/client/inner.rs @@ -296,6 +296,7 @@ impl std::fmt::Debug for Inner( source: SocketAddr, transport: TransportKind, @@ -303,6 +304,7 @@ fn process_discovery

( sd_header:

::SdHeader, session_tracker: &mut SessionTracker, service_registry: &mut ServiceRegistry, + e2e_registry: &Arc>, update_sender: &mpsc::UnboundedSender>, ) where P: PayloadWireFormat + Clone + std::fmt::Debug + 'static, @@ -359,6 +361,13 @@ fn process_discovery

( } if rebooted { + // A rebooted sender restarts its E2E counter at zero, so drop our + // stored per-source state for it; otherwise the first post-reboot + // frame would read as out-of-sequence. + e2e_registry + .lock() + .expect("e2e registry lock poisoned") + .reset_source(source.ip()); let _ = update_sender.send(ClientUpdate::SenderRebooted(source)); } let discovery_msg = DiscoveryMessage { @@ -838,6 +847,7 @@ where request_queue, session_tracker, service_registry, + e2e_registry, run, .. } = &mut self; @@ -865,6 +875,7 @@ where sd_header, session_tracker, service_registry, + e2e_registry, update_sender, ); } @@ -887,6 +898,7 @@ where sd_header, session_tracker, service_registry, + e2e_registry, update_sender, ); } @@ -900,7 +912,7 @@ where trace!("Received unicast message: {:?}", unicast); match unicast { Ok(received) => { - let ReceivedMessage { message: received_message, e2e_status, .. } = received; + let ReceivedMessage { message: received_message, e2e_status, source } = received; // Check if this matches a pending request-response by request_id let request_id = received_message.header().request_id(); if let Some(sender) = pending_responses.remove(&request_id) { @@ -908,7 +920,7 @@ where continue; } // Not a response — forward as ClientUpdate::Unicast - let _ = update_sender.send(ClientUpdate::Unicast { message: received_message, e2e_status }); + let _ = update_sender.send(ClientUpdate::Unicast { message: received_message, e2e_status, source }); } Err(err) => { let _ = update_sender.send(ClientUpdate::Error(err)); diff --git a/src/client/mod.rs b/src/client/mod.rs index 026f2b74..a7245867 100644 --- a/src/client/mod.rs +++ b/src/client/mod.rs @@ -80,6 +80,10 @@ pub enum ClientUpdate { message: Message

, /// E2E check status, if E2E was configured for this message. e2e_status: Option, + /// The sender's source address. On a shared subnet this is the only + /// way to attribute a unicast event to a specific device, since the + /// SOME/IP header carries no instance id. + source: SocketAddr, }, /// The client encountered an error. Error(Error), @@ -93,10 +97,12 @@ impl std::fmt::Debug for ClientUpdate

{ Self::Unicast { message, e2e_status, + source, } => f .debug_struct("Unicast") .field("message", message) .field("e2e_status", e2e_status) + .field("source", source) .finish(), Self::Error(err) => f.debug_tuple("Error").field(err).finish(), } @@ -692,6 +698,7 @@ mod tests { let update: ClientUpdate = ClientUpdate::Unicast { message: msg, e2e_status: None, + source: SocketAddr::new(Ipv4Addr::LOCALHOST.into(), 30640), }; let debug_str = format!("{update:?}"); assert!(debug_str.contains("Unicast")); @@ -702,6 +709,21 @@ mod tests { assert!(debug_str.contains("Error")); } + #[test] + fn unicast_update_carries_source() { + let src = SocketAddr::new(Ipv4Addr::new(192, 168, 11, 101).into(), 30640); + let msg = crate::protocol::Message::new_sd(1, &empty_sd_header()); + let update: ClientUpdate = ClientUpdate::Unicast { + message: msg, + e2e_status: None, + source: src, + }; + match update { + ClientUpdate::Unicast { source, .. } => assert_eq!(source, src), + _ => panic!("expected Unicast"), + } + } + #[tokio::test] async fn test_subscribe_unknown_service_returns_error() { let (client, _updates) = TestClient::new(Ipv4Addr::LOCALHOST); diff --git a/src/client/socket_manager.rs b/src/client/socket_manager.rs index b6e47b31..abe2f773 100644 --- a/src/client/socket_manager.rs +++ b/src/client/socket_manager.rs @@ -271,7 +271,7 @@ where // Apply E2E check if configured let (e2e_status, effective_payload) = { let mut registry = e2e_registry.lock().expect("e2e registry lock poisoned"); - match registry.check(key, payload_bytes, upper_header) { + match registry.check(source_address.ip(), key, payload_bytes, upper_header) { Some((status, stripped)) => (Some(status), stripped), None => (None, payload_bytes), } diff --git a/src/e2e/registry.rs b/src/e2e/registry.rs index 0dcd8c86..9df88e76 100644 --- a/src/e2e/registry.rs +++ b/src/e2e/registry.rs @@ -1,13 +1,31 @@ //! E2E configuration registry for runtime E2E management. use std::collections::HashMap; +use std::net::IpAddr; use super::{E2ECheckStatus, E2EKey, E2EProfile, E2EState, Error, e2e_check, e2e_protect}; -/// Registry mapping message keys to E2E profile configurations and state. +/// Registry mapping message keys to E2E profile configurations and counter +/// state. +/// +/// On a shared subnet several devices send the same `(service, method)` under +/// the same fixed instance id. The profile *configuration* is endpoint-agnostic +/// (one per [`E2EKey`]), but the **receive** counter state must be independent +/// per device — otherwise two senders' interleaved counters collide into +/// spurious `WrongSequence` results. Receive state is therefore keyed by +/// `(source, key)` and created lazily the first time a source is seen. +/// +/// Transmit (protect) counter state stays per-key: a fan-out publish sends the +/// same protected bytes (one counter) to every subscriber, and per-recipient +/// transmit counters are handled a layer up (e.g. `iris_someip_client`). #[derive(Debug)] pub struct E2ERegistry { - map: HashMap, + /// Endpoint-agnostic profile configuration, keyed by data element. + configs: HashMap, + /// Receive counter state, per source address. + rx_states: HashMap<(IpAddr, E2EKey), E2EState>, + /// Transmit counter state, per key. + tx_states: HashMap, } impl E2ERegistry { @@ -15,39 +33,51 @@ impl E2ERegistry { #[must_use] pub fn new() -> Self { Self { - map: HashMap::new(), + configs: HashMap::new(), + rx_states: HashMap::new(), + tx_states: HashMap::new(), } } - /// Register an E2E profile for the given key, creating fresh state. + /// Register an E2E profile for the given key, creating fresh transmit state + /// and clearing any prior per-source receive state for the key. pub fn register(&mut self, key: E2EKey, profile: E2EProfile) { - let state = E2EState::from_profile(&profile); - self.map.insert(key, (profile, state)); + self.tx_states.insert(key, E2EState::from_profile(&profile)); + self.rx_states.retain(|(_, k), _| *k != key); + self.configs.insert(key, profile); } - /// Remove E2E configuration for the given key. + /// Remove E2E configuration (and all state) for the given key. pub fn unregister(&mut self, key: &E2EKey) { - self.map.remove(key); + self.configs.remove(key); + self.tx_states.remove(key); + self.rx_states.retain(|(_, k), _| k != key); } /// Returns `true` if a profile is registered for `key`. #[must_use] pub fn contains_key(&self, key: &E2EKey) -> bool { - self.map.contains_key(key) + self.configs.contains_key(key) } - /// Run E2E check for `key` if configured. + /// Run E2E check for `key` against `source`'s receive counter state, if + /// configured. /// - /// Returns `None` if no profile is registered for `key`. - /// Otherwise returns the check status and the best available payload - /// (stripped E2E header on success, original bytes on check failure). + /// Returns `None` if no profile is registered for `key`. Otherwise returns + /// the check status and the best available payload (stripped E2E header on + /// success, original bytes on check failure). pub fn check<'a>( &mut self, + source: IpAddr, key: E2EKey, payload: &'a [u8], upper_header: [u8; 8], ) -> Option<(E2ECheckStatus, &'a [u8])> { - let (profile, state) = self.map.get_mut(&key)?; + let profile = self.configs.get(&key)?; + let state = self + .rx_states + .entry((source, key)) + .or_insert_with(|| E2EState::from_profile(profile)); Some(e2e_check(profile, state, payload, upper_header)) } @@ -61,9 +91,17 @@ impl E2ERegistry { upper_header: [u8; 8], output: &mut [u8], ) -> Option> { - let (profile, state) = self.map.get_mut(&key)?; + let profile = self.configs.get(&key)?; + let state = self.tx_states.get_mut(&key)?; Some(e2e_protect(profile, state, payload, upper_header, output)) } + + /// Drop all per-source receive state for `source` (e.g. on its reboot), so + /// its next frame starts a fresh counter sequence. Configuration and + /// transmit state are untouched. + pub fn reset_source(&mut self, source: IpAddr) { + self.rx_states.retain(|(s, _), _| *s != source); + } } impl Default for E2ERegistry { @@ -76,11 +114,29 @@ impl Default for E2ERegistry { mod tests { use super::*; use crate::e2e::{Profile4Config, Profile5Config}; + use std::net::Ipv4Addr; fn make_key() -> E2EKey { E2EKey::new(0x1234, 0x5678) } + fn src() -> IpAddr { + IpAddr::V4(Ipv4Addr::LOCALHOST) + } + + fn make_profile5() -> E2EProfile { + E2EProfile::Profile5(Profile5Config::new(0x1234, 20, 15)) + } + + /// Protect a 20-byte "Hello" frame with `sender`'s next transmit counter, + /// writing into `out` and returning the protected length. Avoids `Vec` + /// because the crate's prelude is `core` (no_std-compatible). + fn protect_next(sender: &mut E2ERegistry, key: E2EKey, out: &mut [u8; 64]) -> usize { + let mut payload = [0u8; 20]; + payload[..5].copy_from_slice(b"Hello"); + sender.protect(key, &payload, [0; 8], out).unwrap().unwrap() + } + #[test] fn register_and_check_profile4() { let mut reg = E2ERegistry::new(); @@ -98,7 +154,7 @@ mod tests { .unwrap(); // Check it - let (status, stripped) = reg.check(key, &out[..len], [0; 8]).unwrap(); + let (status, stripped) = reg.check(src(), key, &out[..len], [0; 8]).unwrap(); assert_eq!(status, E2ECheckStatus::Ok); assert_eq!(stripped, payload); } @@ -107,8 +163,7 @@ mod tests { fn register_and_check_profile5() { let mut reg = E2ERegistry::new(); let key = make_key(); - let config = Profile5Config::new(0x1234, 20, 15); - reg.register(key, E2EProfile::Profile5(config)); + reg.register(key, make_profile5()); let mut payload = [0u8; 20]; payload[..5].copy_from_slice(b"Hello"); @@ -118,17 +173,84 @@ mod tests { .unwrap() .unwrap(); - let (status, stripped) = reg.check(key, &out[..len], [0; 8]).unwrap(); + let (status, stripped) = reg.check(src(), key, &out[..len], [0; 8]).unwrap(); assert_eq!(status, E2ECheckStatus::Ok); assert_eq!(stripped, &payload); } + #[test] + fn distinct_sources_have_independent_e2e_state() { + let a = IpAddr::V4(Ipv4Addr::new(192, 168, 11, 101)); + let b = IpAddr::V4(Ipv4Addr::new(192, 168, 11, 102)); + let key = make_key(); + + // A sender produces two frames carrying counters 0 then 1. + let mut sender = E2ERegistry::new(); + sender.register(key, make_profile5()); + let mut b0 = [0u8; 64]; + let l0 = protect_next(&mut sender, key, &mut b0); + let mut b1 = [0u8; 64]; + let l1 = protect_next(&mut sender, key, &mut b1); + + let mut recv = E2ERegistry::new(); + recv.register(key, make_profile5()); + + // Source A consumes counters 0 then 1. + assert_eq!( + recv.check(a, key, &b0[..l0], [0; 8]).unwrap().0, + E2ECheckStatus::Ok + ); + assert_eq!( + recv.check(a, key, &b1[..l1], [0; 8]).unwrap().0, + E2ECheckStatus::Ok + ); + // Source B, interleaved AFTER A, starts its own counter sequence at 0. + // With shared (per-key) receive state this would flag b0 as + // out-of-sequence because A already advanced the single counter past 0. + assert_eq!( + recv.check(b, key, &b0[..l0], [0; 8]).unwrap().0, + E2ECheckStatus::Ok, + "source B's receive counter must be independent of source A's" + ); + assert_eq!( + recv.check(b, key, &b1[..l1], [0; 8]).unwrap().0, + E2ECheckStatus::Ok + ); + } + + #[test] + fn reset_source_clears_only_that_source() { + let a = IpAddr::V4(Ipv4Addr::new(192, 168, 11, 101)); + let key = make_key(); + + let mut sender = E2ERegistry::new(); + sender.register(key, make_profile5()); + let mut b0 = [0u8; 64]; + let l0 = protect_next(&mut sender, key, &mut b0); + let mut b1 = [0u8; 64]; + let l1 = protect_next(&mut sender, key, &mut b1); + + let mut recv = E2ERegistry::new(); + recv.register(key, make_profile5()); + recv.check(a, key, &b0[..l0], [0; 8]); + recv.check(a, key, &b1[..l1], [0; 8]); + + // After a reboot, source A starts fresh — its counter-0 frame is Ok + // again. + recv.reset_source(a); + assert_eq!( + recv.check(a, key, &b0[..l0], [0; 8]).unwrap().0, + E2ECheckStatus::Ok, + "reset_source(a) restarts A's receive counter sequence" + ); + } + #[test] fn unregistered_key_returns_none() { let mut reg = E2ERegistry::new(); let key = make_key(); assert!(!reg.contains_key(&key)); - assert!(reg.check(key, b"test", [0; 8]).is_none()); + assert!(reg.check(src(), key, b"test", [0; 8]).is_none()); assert!(reg.protect(key, b"test", [0; 8], &mut [0; 64]).is_none()); }