feat: per-source RX demux — source on Unicast + per-source E2E state

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

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"

src/client/inner.rs

@@ -296,13 +296,15 @@ impl<PayloadDefinitions: PayloadWireFormat> std::fmt::Debug for Inner<PayloadDef
 /// unicast discovery receive arms so each transport's SD session counter is
 /// tracked on its own key — without this split the sensor's interleaved
 /// multicast/unicast session counters look like perpetual reboots.
+#[allow(clippy::too_many_arguments)]
 fn process_discovery<P>(
     source: SocketAddr,
     transport: TransportKind,
     someip_header: protocol::Header,
     sd_header: <P as PayloadWireFormat>::SdHeader,
     session_tracker: &mut SessionTracker,
     service_registry: &mut ServiceRegistry,
+    e2e_registry: &Arc<Mutex<E2ERegistry>>,
     update_sender: &mpsc::UnboundedSender<ClientUpdate<P>>,
 ) where
     P: PayloadWireFormat + Clone + std::fmt::Debug + 'static,
@@ -359,6 +361,13 @@ fn process_discovery<P>(
     }
 
     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,15 +912,15 @@ 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) {
                                      let _ = sender.send(Ok(received_message.payload().clone()));
                                      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));

src/client/mod.rs

@@ -80,6 +80,10 @@ pub enum ClientUpdate<P: PayloadWireFormat> {
         message: Message<P>,
         /// E2E check status, if E2E was configured for this message.
         e2e_status: Option<E2ECheckStatus>,
+        /// 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<P: PayloadWireFormat> std::fmt::Debug for ClientUpdate<P> {
             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<TestPayload> = 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<TestPayload> = 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);

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),
                                             }