feat: forward-port per-source RX demux (v0.7.3) onto 0.8.0

src/bare_metal_tasks.rs

@@ -131,7 +131,7 @@ pub async fn event_rx_dispatch_future<'a, S, R>(
             continue;
         };
         let (status, body) = if e2e_enabled {
-            check_parsed_e2e(e2e, &parsed)
+            check_parsed_e2e(e2e, core::net::IpAddr::V4(*source.ip()), &parsed)
         } else {
             (E2ECheckStatus::Unchecked, parsed.payload)
         };

src/client/inner.rs

@@ -1057,6 +1057,7 @@ where
                     request_queue,
                     session_tracker,
                     service_registry,
+                    e2e_registry,
                     run,
                     timer,
                     ..
@@ -1194,6 +1195,11 @@ where
                             });
 
                             if rebooted {
+                                // A rebooted sender restarts its E2E counter at
+                                // zero, so drop our stored per-source receive
+                                // state for it; otherwise its first post-reboot
+                                // frame would read as out-of-sequence.
+                                e2e_registry.reset_source(source.ip());
                                 let _ = update_sender.send_now(ClientUpdate::SenderRebooted(source));
                             }
 
@@ -1214,15 +1220,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_now(ClientUpdate::Unicast { message: received_message, e2e_status });
+                             let _ = update_sender.send_now(ClientUpdate::Unicast { message: received_message, e2e_status, source });
                          }
                          Err(err) => {
                              let _ = update_sender.send_now(ClientUpdate::Error(err));

src/client/mod.rs

@@ -213,6 +213,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),
@@ -226,10 +230,12 @@ impl<P: PayloadWireFormat> core::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(),
         }
@@ -1484,6 +1490,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"));
@@ -1494,6 +1501,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, run_fut) = TestClient::new(Ipv4Addr::LOCALHOST);

src/client/socket_manager.rs

@@ -794,9 +794,16 @@ where
                             let key = E2EKey::from_message_id(header.message_id());
                             let payload_bytes = view.payload_bytes();
 
-                            // Apply E2E check if configured
+                            // Apply E2E check if configured. The source IP keys
+                            // the receive counter state so interleaved senders
+                            // on a shared subnet don't collide (see `E2ERegistry`).
                             let (e2e_status, effective_payload) =
-                                match e2e_registry.check(key, payload_bytes, upper_header) {
+                                match e2e_registry.check(
+                                    source_address.ip(),
+                                    key,
+                                    payload_bytes,
+                                    upper_header,
+                                ) {
                                     Some((status, stripped)) => (Some(status), stripped),
                                     None => (None, payload_bytes),
                                 };