From de12b9f5e581765df9c3b7b737688c81860de7c8 Mon Sep 17 00:00:00 2001 From: Justin Kovacich Date: Mon, 15 Jun 2026 15:56:06 -0400 Subject: [PATCH 01/15] test(session): lock per-transport keying; document caller mis-tag bug Pins the SessionTracker contract the SD transport-attribution fix relies on: interleaved multicast/unicast session domains for the same (service,instance) must not look like a reboot, while a real counter reset within a domain must. Co-Authored-By: Claude Opus 4.8 (1M context) --- src/client/session.rs | 44 +++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 44 insertions(+) diff --git a/src/client/session.rs b/src/client/session.rs index 9aa63663..e2ff9ffe 100644 --- a/src/client/session.rs +++ b/src/client/session.rs @@ -310,4 +310,48 @@ mod tests { let verdict = tracker.check(addr(1000), TransportKind::Multicast, SVC, INST, 2, CONT); assert_eq!(verdict, SessionVerdict::Ok); } + + #[test] + fn interleaved_transports_for_same_instance_do_not_false_reboot() { + // A sensor keeps independent SD session-id domains per transport + // (multicast ~1468, unicast ~739). Tracked under distinct keys they + // never look like a reboot when interleaved; a real counter reset + // within one domain still does. + let mut t = SessionTracker::default(); + let a = addr(30490); + assert_eq!( + t.check(a, TransportKind::Multicast, SVC, INST, 1468, RB), + SessionVerdict::Initial + ); + assert_eq!( + t.check(a, TransportKind::Unicast, SVC, INST, 739, RB), + SessionVerdict::Initial + ); + assert_eq!( + t.check(a, TransportKind::Multicast, SVC, INST, 1469, RB), + SessionVerdict::Ok + ); + assert_eq!( + t.check(a, TransportKind::Unicast, SVC, INST, 740, RB), + SessionVerdict::Ok + ); + assert_eq!( + t.check(a, TransportKind::Multicast, SVC, INST, 3, RB), + SessionVerdict::Reboot + ); + } + + #[test] + fn same_transport_mis_tag_false_reboots() { + // Documents the caller bug this fix targets: a unicast datagram + // mis-tagged Multicast collapses two domains onto one key, so its low + // session id looks like a decrease and is wrongly reported as a reboot. + let mut t = SessionTracker::default(); + let a = addr(30490); + t.check(a, TransportKind::Multicast, SVC, INST, 1468, RB); + assert_eq!( + t.check(a, TransportKind::Multicast, SVC, INST, 739, RB), + SessionVerdict::Reboot + ); + } } From f32782c7b8f5841e7692f883a8bc97570391fabe Mon Sep 17 00:00:00 2001 From: Justin Kovacich Date: Tue, 16 Jun 2026 10:16:55 -0400 Subject: [PATCH 02/15] =?UTF-8?q?test(socket=5Fmanager):=20spike=20?= =?UTF-8?q?=E2=80=94=20dual-socket=20multicast/unicast=20SD=20split?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Validates the per-transport SD fix's core assumption: a socket bound to the MULTICAST GROUP address (joined) receives only multicast, and a socket bound to the SPECIFIC host interface IP (not joined) receives only unicast — no cross-delivery. NOTE the spike also found the naive variant fails: binding the unicast socket to INADDR_ANY double-delivers (it receives the multicast too via SO_REUSEPORT), so the unicast socket MUST bind the specific host IP. Skips on hosts with no multicast route (e.g. lo-only CI). Co-Authored-By: Claude Opus 4.8 (1M context) --- src/client/socket_manager.rs | 107 +++++++++++++++++++++++++++++++++++ 1 file changed, 107 insertions(+) diff --git a/src/client/socket_manager.rs b/src/client/socket_manager.rs index 13a8f6cb..2fc2d133 100644 --- a/src/client/socket_manager.rs +++ b/src/client/socket_manager.rs @@ -339,6 +339,113 @@ mod tests { Arc::new(Mutex::new(E2ERegistry::new())) } + /// Spike for the per-transport SD fix: prove the kernel splits SD + /// multicast from unicast across two sockets sharing the SD port — + /// one bound to the group address (joined), one bound to INADDR_ANY + /// (not joined). The reboot fix relies on this so each transport's + /// session counter lands on its own `SessionTracker` key instead of + /// colliding. Skips if the host has no usable multicast route (e.g. + /// `lo`-only CI) — the authoritative check is the live-sensor run. + #[test] + fn dual_socket_splits_multicast_from_unicast() { + use std::eprintln; + use std::net::{Ipv4Addr, SocketAddr, SocketAddrV4, UdpSocket}; + use std::time::Duration; + use std::vec::Vec; + + let group = crate::protocol::sd::MULTICAST_IP; + let port = 51900u16; // test-only; not the real SD port + + let bind_reuse = |addr: SocketAddr| -> std::io::Result { + let s = socket2::Socket::new( + socket2::Domain::IPV4, + socket2::Type::DGRAM, + Some(socket2::Protocol::UDP), + )?; + s.set_reuse_address(true)?; + s.set_reuse_port(true)?; + s.bind(&addr.into())?; + s.set_read_timeout(Some(Duration::from_millis(400)))?; + Ok(s) + }; + let drain = |s: &UdpSocket| -> Vec> { + let mut out = Vec::new(); + let mut buf = [0u8; 64]; + while let Ok((n, _)) = s.recv_from(&mut buf) { + out.push(buf[..n].to_vec()); + } + out + }; + + // Multicast socket: bound to the GROUP address, joined, loopback on. + let mc: UdpSocket = match (|| -> std::io::Result { + let s = bind_reuse(SocketAddr::from((group, port)))?; + s.set_multicast_loop_v4(true)?; + let s: UdpSocket = s.into(); + s.join_multicast_v4(&group, &Ipv4Addr::UNSPECIFIED)?; + Ok(s) + })() { + Ok(s) => s, + Err(e) => { + eprintln!("SKIP dual_socket_splits: multicast setup failed ({e})"); + return; + } + }; + // This host's egress IPv4 for the multicast route — the analogue of + // the real `interface` arg the discovery socket is bound against. + let local_ip = { + let probe = UdpSocket::bind("0.0.0.0:0").expect("probe bind"); + let _ = probe.connect(SocketAddrV4::new(group, port)); + match probe.local_addr() { + Ok(SocketAddr::V4(a)) => *a.ip(), + _ => Ipv4Addr::UNSPECIFIED, + } + }; + if local_ip.is_unspecified() { + eprintln!("SKIP dual_socket_splits: no egress IPv4"); + return; + } + + // Unicast socket: bound to the SPECIFIC host IP (not wildcard), NOT + // joined to the group — so it must not receive the group multicast. + let uc: UdpSocket = bind_reuse(SocketAddr::from((local_ip, port))) + .expect("bind unicast socket") + .into(); + + let tx: UdpSocket = bind_reuse(SocketAddr::from((Ipv4Addr::UNSPECIFIED, 0))) + .expect("bind sender") + .into(); + let _ = tx.set_multicast_loop_v4(true); + let _ = tx.set_multicast_ttl_v4(1); + let _ = tx.send_to(b"MCAST", SocketAddrV4::new(group, port)); + let _ = tx.send_to(b"UCAST", SocketAddrV4::new(local_ip, port)); + std::thread::sleep(Duration::from_millis(60)); + + let mc_got = drain(&mc); + let uc_got = drain(&uc); + + if mc_got.is_empty() { + eprintln!("SKIP dual_socket_splits: no multicast route on this host"); + return; + } + assert!( + mc_got.iter().any(|p| p == b"MCAST"), + "mc socket must get the multicast" + ); + assert!( + !mc_got.iter().any(|p| p == b"UCAST"), + "mc socket (bound to group addr) must NOT get the unicast" + ); + assert!( + uc_got.iter().any(|p| p == b"UCAST"), + "uc socket must get the unicast" + ); + assert!( + !uc_got.iter().any(|p| p == b"MCAST"), + "uc socket (never joined the group) must NOT get the multicast" + ); + } + #[tokio::test] async fn test_bind_ephemeral_port() { let sm = TestSocketManager::bind(0, test_registry()).unwrap(); From 5b33749e2babd46c38055856e9811f17bb684012 Mon Sep 17 00:00:00 2001 From: Justin Kovacich Date: Tue, 16 Jun 2026 10:39:30 -0400 Subject: [PATCH 03/15] test(socket_manager): use Windows-portable dual-socket split (INADDR_ANY mc + interface-IP unicast) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit recvmsg/PKTINFO is per-platform cmsg (no Windows/bare-metal story), so dual-socket is the portable mechanism. Bind-to-group (my first spike) is rejected on Windows, so validate the portable variant instead: keep the multicast socket on INADDR_ANY + join (unchanged, still sends) and add a more-specific interface-IP socket for unicast. 'Most-specific bind wins' diverts the sensor's unicast SD to it, so the wildcard socket sees only multicast — no bind-to-group, no send-path change. Co-Authored-By: Claude Opus 4.8 (1M context) --- src/client/socket_manager.rs | 23 +++++++++++++++-------- 1 file changed, 15 insertions(+), 8 deletions(-) diff --git a/src/client/socket_manager.rs b/src/client/socket_manager.rs index 2fc2d133..9702a878 100644 --- a/src/client/socket_manager.rs +++ b/src/client/socket_manager.rs @@ -340,12 +340,17 @@ mod tests { } /// Spike for the per-transport SD fix: prove the kernel splits SD - /// multicast from unicast across two sockets sharing the SD port — - /// one bound to the group address (joined), one bound to INADDR_ANY - /// (not joined). The reboot fix relies on this so each transport's - /// session counter lands on its own `SessionTracker` key instead of - /// colliding. Skips if the host has no usable multicast route (e.g. - /// `lo`-only CI) — the authoritative check is the live-sensor run. + /// multicast from unicast across two sockets sharing the SD port — the + /// multicast socket bound to `INADDR_ANY` + joined (Windows-portable, and + /// what the real discovery socket already does), and a more-specific + /// socket bound to the host interface IP (not joined). "Most-specific bind + /// wins" must divert the sensor's unicast SD to the interface-IP socket, + /// leaving the wildcard multicast socket seeing only multicast — so each + /// transport's session counter lands on its own `SessionTracker` key + /// instead of colliding (the false-reboot bug). No bind-to-group (Windows + /// rejects it) and no send-path change required. Skips if the host has no + /// usable multicast route (e.g. `lo`-only CI) — the authoritative check is + /// the live-sensor run. #[test] fn dual_socket_splits_multicast_from_unicast() { use std::eprintln; @@ -377,9 +382,11 @@ mod tests { out }; - // Multicast socket: bound to the GROUP address, joined, loopback on. + // Multicast socket: bound to INADDR_ANY (Windows-portable; NOT the + // group address) + joined. Tagged Multicast. The more-specific + // interface-IP unicast socket below must divert unicast away from it. let mc: UdpSocket = match (|| -> std::io::Result { - let s = bind_reuse(SocketAddr::from((group, port)))?; + let s = bind_reuse(SocketAddr::from((Ipv4Addr::UNSPECIFIED, port)))?; s.set_multicast_loop_v4(true)?; let s: UdpSocket = s.into(); s.join_multicast_v4(&group, &Ipv4Addr::UNSPECIFIED)?; From e94937826a71ade88a4377db20f5382c3264b749 Mon Sep 17 00:00:00 2001 From: Justin Kovacich Date: Tue, 16 Jun 2026 10:47:08 -0400 Subject: [PATCH 04/15] fix(client): per-transport SD session tracking via dual discovery sockets MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Splits the sensor's multicast and unicast SD session/reboot domains onto separate SessionTracker keys, fixing the false-reboot churn (interleaved counters looked like perpetual reboots). Mechanism (validated by dual_socket_splits test, portable to Windows + bare-metal Transport trait): - multicast discovery socket: unchanged (INADDR_ANY:30490 + group join, still sends outgoing SD); its receives are now tagged TransportKind::Multicast. - NEW receive-only unicast discovery socket bound to the interface IP — by most-specific-bind-wins the kernel diverts the sensor's unicast SD here; tagged TransportKind::Unicast. The shared process_discovery() feeds both arms, so each transport's session counter is tracked independently. No send-path change. SessionTracker was already correct (keys by (sender, transport, svc, inst)); this stops feeding it a hardcoded Multicast for every datagram. Co-Authored-By: Claude Opus 4.8 (1M context) --- src/client/inner.rs | 200 +++++++++++++++++++++++------------ src/client/socket_manager.rs | 39 +++++++ 2 files changed, 169 insertions(+), 70 deletions(-) diff --git a/src/client/inner.rs b/src/client/inner.rs index 412c6c6e..e28a8cc1 100644 --- a/src/client/inner.rs +++ b/src/client/inner.rs @@ -246,8 +246,14 @@ pub(super) struct Inner { update_sender: mpsc::UnboundedSender>, /// Target interface for sockets interface: Ipv4Addr, - /// Socket manager for service discovery if bound + /// Socket manager for service discovery if bound (multicast: `INADDR_ANY` + /// + group join; also sends outgoing SD) discovery_socket: Option>, + /// Receive-only UNICAST service-discovery socket (interface-IP bound) if + /// bound. Diverts the sensor's unicast SD off `discovery_socket` so the + /// unicast and multicast SD session domains get separate `SessionTracker` + /// keys (prevents interleaved-counter false reboots). + discovery_unicast_socket: Option>, /// Socket managers for unicast messages, keyed by local port unicast_sockets: HashMap>, /// Per-sender SD session state for reboot detection @@ -284,6 +290,85 @@ impl std::fmt::Debug for Inner( + source: SocketAddr, + transport: TransportKind, + someip_header: protocol::Header, + sd_header:

::SdHeader, + session_tracker: &mut SessionTracker, + service_registry: &mut ServiceRegistry, + update_sender: &mpsc::UnboundedSender>, +) where + P: PayloadWireFormat + Clone + std::fmt::Debug + 'static, +{ + // Session ID from the SOME/IP request_id (lower 16 bits). + let session_id = (someip_header.request_id() & 0xFFFF) as u16; + let sd_payload = P::new_sd_payload(&sd_header); + let reboot_flag = sd_payload.sd_flags().map_or( + crate::protocol::sd::RebootFlag::Continuous, + crate::protocol::sd::Flags::reboot, + ); + + // Track sender session/reboot state for every SD entry that identifies a + // service instance, keyed per transport so multicast and unicast domains + // don't collide. + let mut rebooted = false; + for (svc_id, inst_id) in sd_payload.service_instances() { + let verdict = + session_tracker.check(source, transport, svc_id, inst_id, session_id, reboot_flag); + if verdict == SessionVerdict::Reboot { + rebooted = true; + } + } + + // Auto-populate the service registry from offer / stop-offer entries. + for ep in sd_payload.offered_endpoints() { + let id = ServiceInstanceId { + service_id: ep.service_id, + instance_id: ep.instance_id, + }; + if ep.is_offer { + if let Some(addr) = ep.addr { + service_registry.insert( + id, + ServiceEndpointInfo { + addr, + local_port: 0, + major_version: ep.major_version, + minor_version: ep.minor_version, + }, + ); + trace!( + "Registry: added 0x{:04X}.0x{:04X} -> {}", + ep.service_id, ep.instance_id, addr, + ); + } + } else { + service_registry.remove(id); + trace!( + "Registry: removed 0x{:04X}.0x{:04X}", + ep.service_id, ep.instance_id, + ); + } + } + + if rebooted { + let _ = update_sender.send(ClientUpdate::SenderRebooted(source)); + } + let discovery_msg = DiscoveryMessage { + source, + someip_header, + sd_header, + }; + let _ = update_sender.send(ClientUpdate::DiscoveryUpdated(discovery_msg)); +} + impl Inner where PayloadDefinitions: PayloadWireFormat + Clone + std::fmt::Debug + 'static, @@ -306,6 +391,7 @@ where update_sender, interface, discovery_socket: None, + discovery_unicast_socket: None, unicast_sockets: HashMap::new(), session_tracker: SessionTracker::default(), service_registry: ServiceRegistry::default(), @@ -334,6 +420,17 @@ where self.multicast_loopback, )?; self.discovery_socket = Some(socket); + // Receive-only unicast SD socket bound to the interface IP — see + // `discovery_unicast_socket`. Best-effort: if the unicast bind + // fails, multicast discovery still works (we just lose the + // unicast-domain split), so don't fail the whole bind. + match SocketManager::bind_discovery_unicast( + self.interface, + Arc::clone(&self.e2e_registry), + ) { + Ok(unicast) => self.discovery_unicast_socket = Some(unicast), + Err(e) => error!("Failed to bind unicast discovery socket: {e}"), + } Ok(()) } } @@ -347,6 +444,9 @@ where socket.reboot_flag() == crate::protocol::sd::RebootFlag::Continuous; socket.shut_down().await; } + if let Some(socket) = self.discovery_unicast_socket.take() { + socket.shut_down().await; + } } fn set_interface(&mut self, interface: Ipv4Addr) { @@ -732,6 +832,7 @@ where control_receiver, pending_responses, discovery_socket, + discovery_unicast_socket, unicast_sockets, update_sender, request_queue, @@ -757,78 +858,15 @@ where trace!("Received discovery message: {:?}", discovery); match discovery { Ok((source, someip_header, sd_header)) => { - // Extract session ID from SOME/IP request_id (lower 16 bits) - let session_id = (someip_header.request_id() & 0xFFFF) as u16; - let sd_payload = PayloadDefinitions::new_sd_payload(&sd_header); - // Extract reboot flag from the SD payload flags - let reboot_flag = sd_payload - .sd_flags() - .map_or(crate::protocol::sd::RebootFlag::Continuous, |f| { - f.reboot() - }); - - // Track sender session/reboot state for every SD entry - // that identifies a service instance, not only - // offer/stop-offer entries. This ensures reboot - // detection works for all SD traffic (FindService, - // Subscribe, SubscribeAck, etc.). - let mut rebooted = false; - for (svc_id, inst_id) in sd_payload.service_instances() { - let verdict = session_tracker.check( - source, - TransportKind::Multicast, - svc_id, - inst_id, - session_id, - reboot_flag, - ); - if verdict == SessionVerdict::Reboot { - rebooted = true; - } - } - - // Auto-populate service registry from offer/stop-offer - // SD entries. - for ep in sd_payload.offered_endpoints() { - let id = ServiceInstanceId { - service_id: ep.service_id, - instance_id: ep.instance_id, - }; - if ep.is_offer { - if let Some(addr) = ep.addr { - service_registry.insert( - id, - ServiceEndpointInfo { - addr, - local_port: 0, - major_version: ep.major_version, - minor_version: ep.minor_version, - }, - ); - trace!( - "Registry: added 0x{:04X}.0x{:04X} -> {}", - ep.service_id, ep.instance_id, addr, - ); - } - } else { - service_registry.remove(id); - trace!( - "Registry: removed 0x{:04X}.0x{:04X}", - ep.service_id, ep.instance_id, - ); - } - } - - if rebooted { - let _ = update_sender.send(ClientUpdate::SenderRebooted(source)); - } - - let discovery_msg = DiscoveryMessage { + process_discovery::( source, + TransportKind::Multicast, someip_header, sd_header, - }; - let _ = update_sender.send(ClientUpdate::DiscoveryUpdated(discovery_msg)); + session_tracker, + service_registry, + update_sender, + ); } Err(err) => { error!("Error receiving discovery message: {:?}", err); @@ -836,6 +874,28 @@ where } } } + // Unicast SD arrives on the interface-IP-bound socket (the + // sensor's separate unicast SD session domain). + unicast_discovery = Inner::receive_discovery(discovery_unicast_socket) => { + trace!("Received unicast discovery message: {:?}", unicast_discovery); + match unicast_discovery { + Ok((source, someip_header, sd_header)) => { + process_discovery::( + source, + TransportKind::Unicast, + someip_header, + sd_header, + session_tracker, + service_registry, + update_sender, + ); + } + Err(err) => { + error!("Error receiving unicast discovery message: {:?}", err); + let _ = update_sender.send(ClientUpdate::Error(err)); + } + } + } unicast = Inner::receive_any_unicast(unicast_sockets) => { trace!("Received unicast message: {:?}", unicast); match unicast { diff --git a/src/client/socket_manager.rs b/src/client/socket_manager.rs index 9702a878..f6deb1d6 100644 --- a/src/client/socket_manager.rs +++ b/src/client/socket_manager.rs @@ -115,6 +115,45 @@ where }) } + /// Bind a receive-only UNICAST service-discovery socket on the SD port, + /// bound to the specific `interface` IP — more specific than the multicast + /// discovery socket's `INADDR_ANY` bind, so the kernel diverts the sensor's + /// unicast SD datagrams here ("most-specific bind wins"). This keeps the + /// unicast SD session domain on its own `SessionTracker` key, separate from + /// the multicast one, which prevents the interleaved-counter false-reboot + /// bug. No multicast group join; outgoing SD still goes via the multicast + /// discovery socket, so this socket only ever receives. + pub fn bind_discovery_unicast( + interface: Ipv4Addr, + e2e_registry: Arc>, + ) -> Result { + let (rx_tx, rx_rx) = mpsc::channel(16); + let (tx_tx, tx_rx) = mpsc::channel(16); + let bind_addr = std::net::SocketAddr::new(IpAddr::V4(interface), sd::MULTICAST_PORT); + + let socket = socket2::Socket::new( + socket2::Domain::IPV4, + socket2::Type::DGRAM, + Some(socket2::Protocol::UDP), + )?; + socket.set_reuse_address(true)?; + #[cfg(unix)] + socket.set_reuse_port(true)?; + socket.bind(&bind_addr.into())?; + socket.set_nonblocking(true)?; + let socket: std::net::UdpSocket = socket.into(); + let socket = UdpSocket::from_std(socket)?; + + Self::spawn_socket_loop(socket, rx_tx, tx_rx, e2e_registry); + Ok(Self { + receiver: rx_rx, + sender: tx_tx, + local_port: sd::MULTICAST_PORT, + session_id: 1, + session_has_wrapped: false, + }) + } + pub fn bind(port: u16, e2e_registry: Arc>) -> Result { let (rx_tx, rx_rx) = mpsc::channel(4); let (tx_tx, tx_rx) = mpsc::channel(4); From 6451559b31de1802851a38c92a83d30a2bd3cf9e Mon Sep 17 00:00:00 2001 From: Justin Kovacich Date: Tue, 16 Jun 2026 13:07:28 -0400 Subject: [PATCH 05/15] ci: add Windows job to exercise dual-socket SD bind portability MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The per-transport SD fix binds a wildcard multicast socket alongside an interface-IP unicast socket and relies on "most-specific bind wins" to divert the sensor's unicast SD to the unicast socket. That divert is OS-dependent, so add a `windows-latest` job running the full `--all-features` suite (incl. the `client` socket code + `dual_socket_splits_multicast_from_unicast`) to confirm the behavior holds on Windows. Note: the divert test SKIPs (vacuous pass) on hosts with no multicast route / no egress IPv4 — watch the runner log to confirm it actually ran. Co-Authored-By: Claude Opus 4.8 (1M context) --- .github/workflows/ci.yml | 16 ++++++++++++++++ 1 file changed, 16 insertions(+) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index fa278bd8..3fdada81 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -80,3 +80,19 @@ jobs: files: ./target/nextest/default/junit.xml token: ${{ secrets.CODECOV_TOKEN }} fail_ci_if_error: true + + test-windows: + name: Build & Test (Windows) + # Cross-platform socket-bind coverage: the SD discovery path binds a + # wildcard multicast socket alongside an interface-IP unicast socket and + # relies on "most-specific bind wins" to divert unicast SD (see + # `socket_manager::dual_socket_splits_multicast_from_unicast`). That divert + # is the portability risk, so exercise the full `--all-features` suite + # (incl. the `client` socket code) on Windows. No coverage/codecov here — + # this job is purely the OS-portability signal. + runs-on: windows-latest + steps: + - uses: actions/checkout@v6 + - uses: dtolnay/rust-toolchain@stable + - uses: Swatinem/rust-cache@v2 + - run: cargo test --all-features From bbe53034ad6d943c9ec77bb5b37b5bf2ab243ba8 Mon Sep 17 00:00:00 2001 From: Justin Kovacich Date: Tue, 16 Jun 2026 13:25:07 -0400 Subject: [PATCH 06/15] test(socket_manager): gate set_reuse_port behind cfg(unix) in dual-socket test MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `socket2::Socket::set_reuse_port` is Unix-only; the test helper called it unconditionally, breaking the Windows build (the just-added windows-latest CI job caught it). Production `bind_discovery*` already gates it `#[cfg(unix)]` and compiled fine on Windows — only the test helper was missing the gate. `set_reuse_address` (cross-platform) stays unconditional. Co-Authored-By: Claude Opus 4.8 (1M context) --- src/client/socket_manager.rs | 1 + 1 file changed, 1 insertion(+) diff --git a/src/client/socket_manager.rs b/src/client/socket_manager.rs index f6deb1d6..a032c6e5 100644 --- a/src/client/socket_manager.rs +++ b/src/client/socket_manager.rs @@ -407,6 +407,7 @@ mod tests { Some(socket2::Protocol::UDP), )?; s.set_reuse_address(true)?; + #[cfg(unix)] s.set_reuse_port(true)?; s.bind(&addr.into())?; s.set_read_timeout(Some(Duration::from_millis(400)))?; From a2c38a05494c0dc082c3727d9a23eef897df4733 Mon Sep 17 00:00:00 2001 From: Justin Kovacich Date: Tue, 16 Jun 2026 14:01:39 -0400 Subject: [PATCH 07/15] test(server): target loopback in combined-SD test for Windows portability MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `combined_sd_subscribe_uses_its_own_options_run` sent the test datagram to the SD socket's `local_addr()`, which is `0.0.0.0:` (the SD socket binds the wildcard to receive multicast). Sending unicast to `0.0.0.0` loops back on Linux but is dropped on Windows, so the server's `recv_from` timed out — the new windows-latest CI job surfaced this pre-existing Linux-only assumption. Target `127.0.0.1:` explicitly so the test is cross-platform; behavior on Linux is unchanged. Co-Authored-By: Claude Opus 4.8 (1M context) --- src/server/mod.rs | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) diff --git a/src/server/mod.rs b/src/server/mod.rs index 1532b982..9d45db91 100644 --- a/src/server/mod.rs +++ b/src/server/mod.rs @@ -1862,10 +1862,15 @@ mod tests { // datagram. We drive `handle_sd_message` directly rather than // `server.run()` so we can assert state after the call. let client_socket = UdpSocket::bind("127.0.0.1:0").await.unwrap(); - let sd_addr = match server.sd_socket.local_addr().unwrap() { - std::net::SocketAddr::V4(v4) => v4, + // The SD socket binds the wildcard (`0.0.0.0`) to receive multicast, + // so `local_addr()` reports `0.0.0.0:`. Sending unicast to + // `0.0.0.0` loops back on Linux but is dropped on Windows — target + // loopback explicitly with the bound port so the test is portable. + let sd_port = match server.sd_socket.local_addr().unwrap() { + std::net::SocketAddr::V4(v4) => v4.port(), std::net::SocketAddr::V6(_) => panic!("expected v4 sd socket"), }; + let sd_addr = std::net::SocketAddrV4::new(Ipv4Addr::LOCALHOST, sd_port); client_socket.send_to(&message, sd_addr).await.unwrap(); let mut buf = vec![0u8; 65_535]; From 14973004d6d332bce4b7f0f1bd83e75e3239642f Mon Sep 17 00:00:00 2001 From: Justin Kovacich Date: Tue, 16 Jun 2026 14:22:34 -0400 Subject: [PATCH 08/15] test(server): make combined-SD test hermetic (in-memory parse) for Windows The previous commit's loopback retarget was a no-op: the SD socket binds `config.interface:30490` and the test config's interface is already `127.0.0.1`, so `local_addr()` was never the wildcard. The real cause is that every test server binds the same fixed `127.0.0.1:30490` with `SO_REUSEADDR`; under parallel execution Windows delivers the test's unicast to a different bound socket, so `recv_from` timed out. Drop the socket round-trip entirely: `message` is already in memory and the sender addr isn't asserted (the subscriber endpoint must come from the SubscribeEventGroup's `options[1]`), so parse it directly and call `handle_sd_message` with a synthetic sender. Hermetic, no shared-port dependency, identical assertions. Verified on Linux; behavior unchanged. Co-Authored-By: Claude Opus 4.8 (1M context) --- src/server/mod.rs | 39 ++++++++++++++------------------------- 1 file changed, 14 insertions(+), 25 deletions(-) diff --git a/src/server/mod.rs b/src/server/mod.rs index 9d45db91..d8a2efce 100644 --- a/src/server/mod.rs +++ b/src/server/mod.rs @@ -1857,31 +1857,20 @@ mod tests { ); let message = build_sd_message(&sd_header); - // Send the combined SD message to the server's SD socket from a - // fresh client socket and have the server handle exactly one - // datagram. We drive `handle_sd_message` directly rather than - // `server.run()` so we can assert state after the call. - let client_socket = UdpSocket::bind("127.0.0.1:0").await.unwrap(); - // The SD socket binds the wildcard (`0.0.0.0`) to receive multicast, - // so `local_addr()` reports `0.0.0.0:`. Sending unicast to - // `0.0.0.0` loops back on Linux but is dropped on Windows — target - // loopback explicitly with the bound port so the test is portable. - let sd_port = match server.sd_socket.local_addr().unwrap() { - std::net::SocketAddr::V4(v4) => v4.port(), - std::net::SocketAddr::V6(_) => panic!("expected v4 sd socket"), - }; - let sd_addr = std::net::SocketAddrV4::new(Ipv4Addr::LOCALHOST, sd_port); - client_socket.send_to(&message, sd_addr).await.unwrap(); - - let mut buf = vec![0u8; 65_535]; - let (len, sender) = tokio::time::timeout( - std::time::Duration::from_secs(2), - server.sd_socket.recv_from(&mut buf), - ) - .await - .expect("timeout receiving combined SD packet") - .unwrap(); - let view = MessageView::parse(&buf[..len]).unwrap(); + // Parse the combined SD datagram in-memory and drive + // `handle_sd_message` directly rather than `server.run()`, so we can + // assert state after the call. + // + // This previously round-tripped `message` through the server's SD + // socket to obtain the sender addr. But every test server binds the + // same fixed `127.0.0.1:30490` with `SO_REUSEADDR`; under parallel test + // execution Windows delivers the unicast to a different bound socket, so + // the `recv_from` timed out. The sender addr is not asserted here (the + // subscriber endpoint must come from the SubscribeEventGroup's + // `options[1]`), so a synthetic sender keeps the test hermetic and + // cross-platform. + let sender = std::net::SocketAddr::from((Ipv4Addr::LOCALHOST, 54321)); + let view = MessageView::parse(&message).unwrap(); let sd_view = view.sd_header().unwrap(); server.handle_sd_message(&sd_view, sender).await.unwrap(); From c3713f5f5ce0f9df52e4f15f7bc553c2726ceb45 Mon Sep 17 00:00:00 2001 From: Justin Kovacich Date: Tue, 16 Jun 2026 14:54:37 -0400 Subject: [PATCH 09/15] ci: scope Windows job to build-all + run lib tests MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The Windows job's purpose is to confirm the dual-socket SD bind divert is portable; that assertion lives in the library unit tests (`dual_socket_splits_multicast_from_unicast`). Running the full suite also ran `tests/client_server.rs`, whose integration tests bind fixed SD ports and are flaky under parallel execution on every platform — they fail on `main` on the same host and flake on the ubuntu coverage job too (tracked separately, #84). Split into `--no-run` (compile everything incl. integration tests = build portability) + `--lib` (run unit tests incl. the divert = behavior portability). The integration suite still compiles on Windows but isn't run here, so the job reflects this fix's portability without unrelated cross-platform flakiness. Co-Authored-By: Claude Opus 4.8 (1M context) --- .github/workflows/ci.yml | 18 ++++++++++++++---- 1 file changed, 14 insertions(+), 4 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 3fdada81..eda01259 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -87,12 +87,22 @@ jobs: # wildcard multicast socket alongside an interface-IP unicast socket and # relies on "most-specific bind wins" to divert unicast SD (see # `socket_manager::dual_socket_splits_multicast_from_unicast`). That divert - # is the portability risk, so exercise the full `--all-features` suite - # (incl. the `client` socket code) on Windows. No coverage/codecov here — - # this job is purely the OS-portability signal. + # is the portability risk. No coverage/codecov here — purely the + # OS-portability signal. + # + # Two steps: + # 1. `--no-run` compiles everything (incl. the `tests/` integration suite) + # to prove the crate *builds* on Windows. + # 2. `--lib` runs the library unit tests, which include the dual-socket + # divert assertion — the behavior we actually need to confirm on Windows. + # The `tests/client_server.rs` integration suite is compiled but NOT run + # here: it binds fixed SD ports and is flaky under parallel execution on + # every platform (reproduces on `main`; tracked separately, #84), so running + # it would add cross-platform noise unrelated to this fix. runs-on: windows-latest steps: - uses: actions/checkout@v6 - uses: dtolnay/rust-toolchain@stable - uses: Swatinem/rust-cache@v2 - - run: cargo test --all-features + - run: cargo test --all-features --no-run + - run: cargo test --all-features --lib From 03a4db73334b937f2397307d5caef05273c159b1 Mon Sep 17 00:00:00 2001 From: Justin Kovacich Date: Tue, 16 Jun 2026 16:34:40 -0400 Subject: [PATCH 10/15] test(client): address Copilot review on dual-socket divert test - Bind the multicast probe socket to an ephemeral port (:0) and reuse the OS-assigned port for the unicast socket + sender target, instead of a hardcoded 51900 that could collide on a shared CI runner. - Treat probe send_to failures as an explicit, logged SKIP (env issue, not a logic regression) rather than relying on an empty drain to pass. - Fix inaccurate assertion message: the mc socket is bound to INADDR_ANY, not the group address. - Clarify bind_discovery_unicast doc: it is receive-only by usage, not by type (the returned SocketManager still carries an unused send half). Co-Authored-By: Claude Opus 4.8 (1M context) --- src/client/socket_manager.rs | 32 +++++++++++++++++++++++++++----- 1 file changed, 27 insertions(+), 5 deletions(-) diff --git a/src/client/socket_manager.rs b/src/client/socket_manager.rs index a032c6e5..b6e47b31 100644 --- a/src/client/socket_manager.rs +++ b/src/client/socket_manager.rs @@ -123,6 +123,10 @@ where /// the multicast one, which prevents the interleaved-counter false-reboot /// bug. No multicast group join; outgoing SD still goes via the multicast /// discovery socket, so this socket only ever receives. + /// + /// The returned `SocketManager` still carries a send half and session + /// counter for type uniformity, but the discovery layer never drives them + /// for this socket: it is receive-only *by usage*, not by type. pub fn bind_discovery_unicast( interface: Ipv4Addr, e2e_registry: Arc>, @@ -398,7 +402,6 @@ mod tests { use std::vec::Vec; let group = crate::protocol::sd::MULTICAST_IP; - let port = 51900u16; // test-only; not the real SD port let bind_reuse = |addr: SocketAddr| -> std::io::Result { let s = socket2::Socket::new( @@ -426,7 +429,7 @@ mod tests { // group address) + joined. Tagged Multicast. The more-specific // interface-IP unicast socket below must divert unicast away from it. let mc: UdpSocket = match (|| -> std::io::Result { - let s = bind_reuse(SocketAddr::from((Ipv4Addr::UNSPECIFIED, port)))?; + let s = bind_reuse(SocketAddr::from((Ipv4Addr::UNSPECIFIED, 0)))?; s.set_multicast_loop_v4(true)?; let s: UdpSocket = s.into(); s.join_multicast_v4(&group, &Ipv4Addr::UNSPECIFIED)?; @@ -438,6 +441,16 @@ mod tests { return; } }; + // Reuse the OS-assigned ephemeral port for the unicast socket and the + // sender target too, so the test never collides with a fixed port that + // happens to be in use on a shared CI runner. + let port = match mc.local_addr() { + Ok(SocketAddr::V4(a)) => a.port(), + _ => { + eprintln!("SKIP dual_socket_splits: multicast socket has no IPv4 local addr"); + return; + } + }; // This host's egress IPv4 for the multicast route — the analogue of // the real `interface` arg the discovery socket is bound against. let local_ip = { @@ -464,8 +477,17 @@ mod tests { .into(); let _ = tx.set_multicast_loop_v4(true); let _ = tx.set_multicast_ttl_v4(1); - let _ = tx.send_to(b"MCAST", SocketAddrV4::new(group, port)); - let _ = tx.send_to(b"UCAST", SocketAddrV4::new(local_ip, port)); + // A send failure here is an environment issue (no route / permissions), + // not a logic regression — surface it as a visible SKIP rather than + // letting an empty drain quietly pass the test. + if let Err(e) = tx.send_to(b"MCAST", SocketAddrV4::new(group, port)) { + eprintln!("SKIP dual_socket_splits: multicast send failed ({e})"); + return; + } + if let Err(e) = tx.send_to(b"UCAST", SocketAddrV4::new(local_ip, port)) { + eprintln!("SKIP dual_socket_splits: unicast send failed ({e})"); + return; + } std::thread::sleep(Duration::from_millis(60)); let mc_got = drain(&mc); @@ -481,7 +503,7 @@ mod tests { ); assert!( !mc_got.iter().any(|p| p == b"UCAST"), - "mc socket (bound to group addr) must NOT get the unicast" + "mc socket (bound to INADDR_ANY) must NOT get the unicast" ); assert!( uc_got.iter().any(|p| p == b"UCAST"), From 9a2f3b20719d6ff4775fd2891a1b6e2e086b4249 Mon Sep 17 00:00:00 2001 From: Justin Kovacich Date: Wed, 17 Jun 2026 09:44:50 -0400 Subject: [PATCH 11/15] test(client_server): put server on a distinct loopback IP to avoid SD-port self-collision MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The client's receive-only unicast discovery socket binds interface:30490 with SO_REUSEPORT. When the integration-test server bound its SD socket on the same IP:30490, the client's own SubscribeEventgroup could be SO_REUSEPORT load-balanced to the client's own discovery socket, so the server never registered the subscriber and wait_for_subscribers failed (client_server.rs:172). This is a test-only artifact: in production a full SD-announcing server is a remote sensor on its own IP, and the co-located server in iris_someip_client is new_passive (binds an ephemeral SD port, never 30490). So the collision only arises when a full Server::new is co-located on the same IP as a client, which only the integration test did. Move the test server to 127.0.0.2 (client stays 127.0.0.1), mirroring the remote-sensor topology. The serial-sd-port nextest group is retained — multiple client_server tests each bind their server's :30490 and must stay serial vs each other. Full suite passes via nextest (423/423). Co-Authored-By: Claude Opus 4.8 (1M context) --- tests/client_server.rs | 32 +++++++++++++++++++++----------- 1 file changed, 21 insertions(+), 11 deletions(-) diff --git a/tests/client_server.rs b/tests/client_server.rs index ffd6d349..7ed11582 100644 --- a/tests/client_server.rs +++ b/tests/client_server.rs @@ -16,9 +16,19 @@ fn empty_sd_header() -> VecSdHeader { type TestClient = Client; +/// The full `Server` binds the SD port (30490) on its interface. Keep it on a +/// distinct loopback IP from the client (which stays on `127.0.0.1`) so the +/// client's receive-only unicast discovery socket (`interface:30490`, +/// `SO_REUSEPORT`) does not collide with the server's SD socket on the same +/// `IP:30490` and steal the client's own SubscribeEventgroup. This mirrors +/// production, where a full SD-announcing server is a remote sensor on its own +/// IP (the co-located server in `iris_someip_client` is `new_passive`, which +/// never binds 30490). See the discussion on PR #130. +const SERVER_IP: Ipv4Addr = Ipv4Addr::new(127, 0, 0, 2); + /// Create a server on an ephemeral unicast port, returning (Server, actual_port). async fn create_server(service_id: u16, instance_id: u16) -> (Server, u16) { - let config = ServerConfig::new(Ipv4Addr::LOCALHOST, 0, service_id, instance_id); + let config = ServerConfig::new(SERVER_IP, 0, service_id, instance_id); let mut server: Server = Server::new(config).await.expect("Server::new failed"); let port = match server.unicast_local_addr().expect("local_addr failed") { std::net::SocketAddr::V4(a) => a.port(), @@ -57,7 +67,7 @@ async fn test_client_server_subscribe_and_receive_event() { // Create client and subscribe to the server's event group let (client, mut updates) = TestClient::new(Ipv4Addr::LOCALHOST); - let server_addr = SocketAddrV4::new(Ipv4Addr::LOCALHOST, server_port); + let server_addr = SocketAddrV4::new(SERVER_IP, server_port); client.add_endpoint(0x5B, 1, server_addr, 0).await.unwrap(); client.subscribe(0x5B, 1, 1, 3, 0x01, 0).await.unwrap(); @@ -113,7 +123,7 @@ async fn test_client_send_sd_auto_binds_discovery() { port: 12345, }], }; - let target = SocketAddrV4::new(Ipv4Addr::LOCALHOST, server_port); + let target = SocketAddrV4::new(SERVER_IP, server_port); client .send_sd_message(target, sd_header) .await @@ -134,7 +144,7 @@ async fn test_client_bind_unbind_lifecycle_with_server() { // Bind discovery, subscribe, then unbind and rebind client.bind_discovery().await.unwrap(); - let server_addr = SocketAddrV4::new(Ipv4Addr::LOCALHOST, server_port); + let server_addr = SocketAddrV4::new(SERVER_IP, server_port); client.add_endpoint(0x5B, 1, server_addr, 0).await.unwrap(); client.subscribe(0x5B, 1, 1, 3, 0x01, 0).await.unwrap(); @@ -162,7 +172,7 @@ async fn test_add_endpoint_and_send_to_service() { client.bind_discovery().await.unwrap(); // Register the server's endpoint manually (simulating non-broadcasting service) - let server_addr = SocketAddrV4::new(Ipv4Addr::LOCALHOST, server_port); + let server_addr = SocketAddrV4::new(SERVER_IP, server_port); client.add_endpoint(0x5B, 1, server_addr, 0).await.unwrap(); // Subscribe to server's event group (auto-binds unicast internally) @@ -219,7 +229,7 @@ async fn test_subscribe_auto_binds_discovery() { // Create client — do NOT bind discovery manually let (client, mut updates) = TestClient::new(Ipv4Addr::LOCALHOST); - let server_addr = SocketAddrV4::new(Ipv4Addr::LOCALHOST, server_port); + let server_addr = SocketAddrV4::new(SERVER_IP, server_port); client.add_endpoint(0x5B, 1, server_addr, 0).await.unwrap(); // Subscribe should auto-bind discovery internally client.subscribe(0x5B, 1, 1, 3, 0x01, 0).await.unwrap(); @@ -261,7 +271,7 @@ async fn test_client_request_resolves_via_unicast_reply() { let server_handle = tokio::spawn(async move { server.run().await }); let (client, mut updates) = TestClient::new(Ipv4Addr::LOCALHOST); - let server_addr = SocketAddrV4::new(Ipv4Addr::LOCALHOST, server_port); + let server_addr = SocketAddrV4::new(SERVER_IP, server_port); client.add_endpoint(0x5B, 1, server_addr, 0).await.unwrap(); client.subscribe(0x5B, 1, 1, 3, 0x01, 0).await.unwrap(); @@ -326,7 +336,7 @@ async fn test_e2e_protect_on_publish_and_check_on_receive() { // Register matching E2E profile on client client.register_e2e(key, profile); - let server_addr = SocketAddrV4::new(Ipv4Addr::LOCALHOST, server_port); + let server_addr = SocketAddrV4::new(SERVER_IP, server_port); client.add_endpoint(0x5B, 1, server_addr, 0).await.unwrap(); client.subscribe(0x5B, 1, 1, 3, 0x01, 0).await.unwrap(); @@ -382,7 +392,7 @@ async fn test_multiple_subscribers_receive_events() { let publisher = server.publisher(); let server_handle = tokio::spawn(async move { server.run().await }); - let server_addr = SocketAddrV4::new(Ipv4Addr::LOCALHOST, server_port); + let server_addr = SocketAddrV4::new(SERVER_IP, server_port); // Client 1 let (client1, mut updates1) = TestClient::new(Ipv4Addr::LOCALHOST); @@ -462,7 +472,7 @@ async fn test_cloned_client_works() { let client2 = client.clone(); // Both clones can send commands - let server_addr = SocketAddrV4::new(Ipv4Addr::LOCALHOST, server_port); + let server_addr = SocketAddrV4::new(SERVER_IP, server_port); client.add_endpoint(0x5B, 1, server_addr, 0).await.unwrap(); client2.subscribe(0x5B, 1, 1, 3, 0x01, 0).await.unwrap(); @@ -479,7 +489,7 @@ async fn test_subscribe_specific_port_reuse() { let server_handle = tokio::spawn(async move { server.run().await }); let (client, _updates) = TestClient::new(Ipv4Addr::LOCALHOST); - let server_addr = SocketAddrV4::new(Ipv4Addr::LOCALHOST, server_port); + let server_addr = SocketAddrV4::new(SERVER_IP, server_port); client.add_endpoint(0x5B, 1, server_addr, 0).await.unwrap(); // Use specific port From eb408e379d2d4177c9f3eb6f1d84177f8c569595 Mon Sep 17 00:00:00 2001 From: Justin Kovacich Date: Wed, 17 Jun 2026 10:25:16 -0400 Subject: [PATCH 12/15] test(socket_manager): deterministic loopback divert test + SD spelling fix MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add `most_specific_bind_diverts_unicast_on_loopback`, a deterministic companion to `dual_socket_splits_multicast_from_unicast`. The multicast variant needs a live multicast route and SKIPs on a lo-only host (CI), leaving the "most-specific bind wins" divert — the load-bearing mechanism of the per-transport SD fix — unexercised in coverage and prone to the vacuous-skip trap. The new test proves the same divert with pure unicast on loopback (ephemeral port, never the fixed SD port), so it never skips: a datagram to the interface IP lands on the interface-bound socket, not the wildcard socket sharing the port. Unix-gated (leans on SO_REUSEPORT, the discovery socket's own #[cfg(unix)] reuse path); cross-platform compilation stays covered by the build-all CI job. Also fix the lone `SubscribeEventgroup` -> `SubscribeEventGroup` (capital G) typo in the client_server doc comment to match the spec spelling used everywhere else in the crate (Copilot review). Co-Authored-By: Claude Opus 4.8 (1M context) --- src/client/socket_manager.rs | 65 ++++++++++++++++++++++++++++++++++++ tests/client_server.rs | 2 +- 2 files changed, 66 insertions(+), 1 deletion(-) diff --git a/src/client/socket_manager.rs b/src/client/socket_manager.rs index b6e47b31..e671af3c 100644 --- a/src/client/socket_manager.rs +++ b/src/client/socket_manager.rs @@ -515,6 +515,71 @@ mod tests { ); } + /// Deterministic companion to `dual_socket_splits_multicast_from_unicast`. + /// That test needs a live multicast route and SKIPs on a `lo`-only host + /// (CI), which leaves the divert mechanism — the load-bearing part of the + /// per-transport SD fix — unexercised in coverage and vulnerable to the + /// vacuous-skip trap. This proves the same "most-specific bind wins" + /// divert with pure unicast on loopback, so it never skips: a datagram + /// sent to the interface IP must land on the interface-bound socket, never + /// the wildcard socket sharing the same port. + /// + /// Unix-only: it leans on `SO_REUSEPORT` to bind both sockets to one port + /// (the discovery socket's own `#[cfg(unix)]` reuse path). The bind's + /// cross-platform compilation is covered by the build-all CI job. + #[cfg(unix)] + #[tokio::test] + async fn most_specific_bind_diverts_unicast_on_loopback() { + use std::net::{Ipv4Addr, SocketAddr, UdpSocket}; + use std::time::Duration; + + let bind_reuse = |addr: SocketAddr| -> std::io::Result { + let s = socket2::Socket::new( + socket2::Domain::IPV4, + socket2::Type::DGRAM, + Some(socket2::Protocol::UDP), + )?; + s.set_reuse_address(true)?; + s.set_reuse_port(true)?; + s.bind(&addr.into())?; + s.set_read_timeout(Some(Duration::from_millis(400)))?; + Ok(s.into()) + }; + + // Wildcard socket takes an OS-assigned ephemeral port (so this never + // collides with the fixed SD port other tests use); the more-specific + // interface socket then shares that exact port — the divert under test. + let wildcard = + bind_reuse(SocketAddr::from((Ipv4Addr::UNSPECIFIED, 0))).expect("bind wildcard socket"); + let port = match wildcard.local_addr().expect("wildcard local addr") { + SocketAddr::V4(a) => a.port(), + SocketAddr::V6(_) => unreachable!("bound IPv4"), + }; + let specific = bind_reuse(SocketAddr::from((Ipv4Addr::LOCALHOST, port))) + .expect("bind interface-specific socket"); + + let tx = UdpSocket::bind("127.0.0.1:0").expect("bind sender"); + tx.send_to(b"UCAST", SocketAddr::from((Ipv4Addr::LOCALHOST, port))) + .expect("send unicast"); + std::thread::sleep(Duration::from_millis(60)); + + let mut buf = [0u8; 64]; + // The interface bind is more specific than the wildcard, so the kernel + // delivers the unicast there — not load-balanced, because the binds + // differ in specificity. + let got = specific.recv_from(&mut buf); + assert!( + matches!(got, Ok((n, _)) if &buf[..n] == b"UCAST"), + "interface-bound socket must receive the unicast (most-specific bind wins), got {got:?}" + ); + // The wildcard must not have stolen it. + let stolen = wildcard.recv_from(&mut buf); + assert!( + stolen.is_err(), + "wildcard socket must NOT receive the diverted unicast, got {stolen:?}" + ); + } + #[tokio::test] async fn test_bind_ephemeral_port() { let sm = TestSocketManager::bind(0, test_registry()).unwrap(); diff --git a/tests/client_server.rs b/tests/client_server.rs index 7ed11582..dd44860a 100644 --- a/tests/client_server.rs +++ b/tests/client_server.rs @@ -20,7 +20,7 @@ type TestClient = Client; /// distinct loopback IP from the client (which stays on `127.0.0.1`) so the /// client's receive-only unicast discovery socket (`interface:30490`, /// `SO_REUSEPORT`) does not collide with the server's SD socket on the same -/// `IP:30490` and steal the client's own SubscribeEventgroup. This mirrors +/// `IP:30490` and steal the client's own SubscribeEventGroup. This mirrors /// production, where a full SD-announcing server is a remote sensor on its own /// IP (the co-located server in `iris_someip_client` is `new_passive`, which /// never binds 30490). See the discussion on PR #130. From 3d10bace397b696f7c69eafaaea1adef53ac511d Mon Sep 17 00:00:00 2001 From: Justin Kovacich Date: Wed, 17 Jun 2026 10:41:12 -0400 Subject: [PATCH 13/15] Revert the deterministic loopback divert test (CI interference hazard) The companion test bound `0.0.0.0:` with `SO_REUSEPORT`, which can collide with a parallel integration-test client that the OS hands the same ephemeral port: that client's `bind` (REUSEADDR only, no REUSEPORT) then fails with `EADDRINUSE` and its test fails. This surfaced as a `test_multiple_ subscribers_receive_events` failure on the coverage job. The production `bind_discovery_unicast` is already covered via the `inner.rs` discovery tests, so the marginal coverage isn't worth the cross-test hazard. The `SubscribeEventGroup` spelling fix (prior commit) stays. Co-Authored-By: Claude Opus 4.8 (1M context) --- src/client/socket_manager.rs | 65 ------------------------------------ 1 file changed, 65 deletions(-) diff --git a/src/client/socket_manager.rs b/src/client/socket_manager.rs index e671af3c..b6e47b31 100644 --- a/src/client/socket_manager.rs +++ b/src/client/socket_manager.rs @@ -515,71 +515,6 @@ mod tests { ); } - /// Deterministic companion to `dual_socket_splits_multicast_from_unicast`. - /// That test needs a live multicast route and SKIPs on a `lo`-only host - /// (CI), which leaves the divert mechanism — the load-bearing part of the - /// per-transport SD fix — unexercised in coverage and vulnerable to the - /// vacuous-skip trap. This proves the same "most-specific bind wins" - /// divert with pure unicast on loopback, so it never skips: a datagram - /// sent to the interface IP must land on the interface-bound socket, never - /// the wildcard socket sharing the same port. - /// - /// Unix-only: it leans on `SO_REUSEPORT` to bind both sockets to one port - /// (the discovery socket's own `#[cfg(unix)]` reuse path). The bind's - /// cross-platform compilation is covered by the build-all CI job. - #[cfg(unix)] - #[tokio::test] - async fn most_specific_bind_diverts_unicast_on_loopback() { - use std::net::{Ipv4Addr, SocketAddr, UdpSocket}; - use std::time::Duration; - - let bind_reuse = |addr: SocketAddr| -> std::io::Result { - let s = socket2::Socket::new( - socket2::Domain::IPV4, - socket2::Type::DGRAM, - Some(socket2::Protocol::UDP), - )?; - s.set_reuse_address(true)?; - s.set_reuse_port(true)?; - s.bind(&addr.into())?; - s.set_read_timeout(Some(Duration::from_millis(400)))?; - Ok(s.into()) - }; - - // Wildcard socket takes an OS-assigned ephemeral port (so this never - // collides with the fixed SD port other tests use); the more-specific - // interface socket then shares that exact port — the divert under test. - let wildcard = - bind_reuse(SocketAddr::from((Ipv4Addr::UNSPECIFIED, 0))).expect("bind wildcard socket"); - let port = match wildcard.local_addr().expect("wildcard local addr") { - SocketAddr::V4(a) => a.port(), - SocketAddr::V6(_) => unreachable!("bound IPv4"), - }; - let specific = bind_reuse(SocketAddr::from((Ipv4Addr::LOCALHOST, port))) - .expect("bind interface-specific socket"); - - let tx = UdpSocket::bind("127.0.0.1:0").expect("bind sender"); - tx.send_to(b"UCAST", SocketAddr::from((Ipv4Addr::LOCALHOST, port))) - .expect("send unicast"); - std::thread::sleep(Duration::from_millis(60)); - - let mut buf = [0u8; 64]; - // The interface bind is more specific than the wildcard, so the kernel - // delivers the unicast there — not load-balanced, because the binds - // differ in specificity. - let got = specific.recv_from(&mut buf); - assert!( - matches!(got, Ok((n, _)) if &buf[..n] == b"UCAST"), - "interface-bound socket must receive the unicast (most-specific bind wins), got {got:?}" - ); - // The wildcard must not have stolen it. - let stolen = wildcard.recv_from(&mut buf); - assert!( - stolen.is_err(), - "wildcard socket must NOT receive the diverted unicast, got {stolen:?}" - ); - } - #[tokio::test] async fn test_bind_ephemeral_port() { let sm = TestSocketManager::bind(0, test_registry()).unwrap(); From d362f752ec8b72ec7ce8edec69371f127c443316 Mon Sep 17 00:00:00 2001 From: Justin Kovacich Date: Wed, 17 Jun 2026 10:51:12 -0400 Subject: [PATCH 14/15] docs(client_server): describe discovery-socket reuse as cross-platform MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The SD-port-collision comment said the client's unicast discovery socket uses `SO_REUSEPORT`, but `bind_discovery_unicast` only sets it under `#[cfg(unix)]` (Windows gets `SO_REUSEADDR` only). Reword to "address/port reuse — SO_REUSEPORT on Unix" so the comment is accurate cross-platform (Copilot review). Co-Authored-By: Claude Opus 4.8 (1M context) --- tests/client_server.rs | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/tests/client_server.rs b/tests/client_server.rs index dd44860a..c7a1c685 100644 --- a/tests/client_server.rs +++ b/tests/client_server.rs @@ -18,9 +18,10 @@ type TestClient = Client; /// The full `Server` binds the SD port (30490) on its interface. Keep it on a /// distinct loopback IP from the client (which stays on `127.0.0.1`) so the -/// client's receive-only unicast discovery socket (`interface:30490`, -/// `SO_REUSEPORT`) does not collide with the server's SD socket on the same -/// `IP:30490` and steal the client's own SubscribeEventGroup. This mirrors +/// client's receive-only unicast discovery socket on `interface:30490` (bound +/// with address/port reuse — `SO_REUSEPORT` on Unix) does not collide with the +/// server's SD socket on the same `IP:30490` and steal the client's own +/// SubscribeEventGroup. This mirrors /// production, where a full SD-announcing server is a remote sensor on its own /// IP (the co-located server in `iris_someip_client` is `new_passive`, which /// never binds 30490). See the discussion on PR #130. From 752cdcec69efcd5e6d30e68290ae82b364a6f54c Mon Sep 17 00:00:00 2001 From: Justin Kovacich Date: Wed, 17 Jun 2026 11:11:08 -0400 Subject: [PATCH 15/15] test(client_server): drain until the Unicast event, not just the next update MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The coverage job intermittently failed `test_multiple_subscribers_receive_events` with `client1 expected Unicast, got DiscoveryUpdated(SubscribeAckEventGroup)`. Root cause: this PR's unicast SD socket now delivers the SubscribeAck to the client as a `DiscoveryUpdated` on the same update channel as events. The tests did a single 250ms pre-drain then asserted the *next* update was the `Unicast` event — racy, because under the slower coverage build the ack can arrive after the pre-drain and get consumed where the event is expected. Add a `recv_unicast` helper that drains the update stream until the `Unicast` event arrives, skipping interleaved discovery acks / reboot notices (bounded by a 5s deadline). Apply it to every test that asserts event receipt. The opportunistic pre-drains are kept (harmless) and the helper makes ordering no longer load-bearing. Co-Authored-By: Claude Opus 4.8 (1M context) --- tests/client_server.rs | 76 ++++++++++++++++++------------------------ 1 file changed, 32 insertions(+), 44 deletions(-) diff --git a/tests/client_server.rs b/tests/client_server.rs index c7a1c685..2a95f675 100644 --- a/tests/client_server.rs +++ b/tests/client_server.rs @@ -3,7 +3,9 @@ use simple_someip::e2e::{E2ECheckStatus, E2EKey, E2EProfile, Profile4Config}; use simple_someip::protocol::{Header, Message, MessageId, sd}; use simple_someip::server::ServerConfig; -use simple_someip::{Client, ClientUpdate, PayloadWireFormat, RawPayload, Server, VecSdHeader}; +use simple_someip::{ + Client, ClientUpdate, ClientUpdates, PayloadWireFormat, RawPayload, Server, VecSdHeader, +}; use std::net::{Ipv4Addr, SocketAddrV4}; fn empty_sd_header() -> VecSdHeader { @@ -59,6 +61,26 @@ async fn wait_for_subscribers( false } +/// Drain a client's update stream until the published `Unicast` event arrives, +/// skipping interleaved discovery traffic. A `SubscribeAck` now reaches the +/// client via the unicast SD socket (the per-transport fix in this PR) and can +/// land on the channel just before the event, so a single `recv()` that expects +/// the event outright is racy — especially under the slower coverage build. +/// Panics on timeout or a closed channel; returns the `Unicast` update so +/// callers can inspect fields like `e2e_status`. +async fn recv_unicast(updates: &mut ClientUpdates) -> ClientUpdate { + let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(5); + loop { + match tokio::time::timeout_at(deadline, updates.recv()).await { + Ok(Some(update @ ClientUpdate::Unicast { .. })) => return update, + // Discovery ack / reboot notice — keep waiting for the event. + Ok(Some(_)) => continue, + Ok(None) => panic!("update channel closed before the Unicast event"), + Err(_) => panic!("timed out waiting for the Unicast event"), + } + } +} + #[tokio::test] async fn test_client_server_subscribe_and_receive_event() { // Start server on ephemeral port @@ -89,13 +111,7 @@ async fn test_client_server_subscribe_and_receive_event() { assert_eq!(sent, 1); // Client receives the unicast event - let update = tokio::time::timeout(std::time::Duration::from_secs(2), updates.recv()) - .await - .expect("timeout waiting for Unicast"); - assert!( - matches!(update, Some(ClientUpdate::Unicast { .. })), - "expected Unicast, got {update:?}" - ); + recv_unicast(&mut updates).await; // Tear down client.unbind_discovery().await.unwrap(); @@ -197,13 +213,7 @@ async fn test_add_endpoint_and_send_to_service() { assert_eq!(sent, 1); // Client receives the unicast event - let update = tokio::time::timeout(std::time::Duration::from_secs(2), updates.recv()) - .await - .expect("timeout waiting for Unicast"); - assert!( - matches!(update, Some(ClientUpdate::Unicast { .. })), - "expected Unicast, got {update:?}" - ); + recv_unicast(&mut updates).await; // Remove the endpoint and verify send_to_service returns ServiceNotFound client.remove_endpoint(0x5B, 1).await.unwrap(); @@ -251,13 +261,7 @@ async fn test_subscribe_auto_binds_discovery() { .expect("publish_event failed"); assert_eq!(sent, 1); - let update = tokio::time::timeout(std::time::Duration::from_secs(2), updates.recv()) - .await - .expect("timeout waiting for Unicast"); - assert!( - matches!(update, Some(ClientUpdate::Unicast { .. })), - "expected Unicast, got {update:?}" - ); + recv_unicast(&mut updates).await; client.shut_down(); server_handle.abort(); @@ -363,11 +367,8 @@ async fn test_e2e_protect_on_publish_and_check_on_receive() { assert_eq!(sent, 1); // Client receives the unicast event with E2E status - let update = tokio::time::timeout(std::time::Duration::from_secs(2), updates.recv()) - .await - .expect("timeout waiting for Unicast"); - match update { - Some(ClientUpdate::Unicast { e2e_status, .. }) => { + match recv_unicast(&mut updates).await { + ClientUpdate::Unicast { e2e_status, .. } => { assert!( e2e_status.is_some(), "expected e2e_status to be populated when E2E is configured" @@ -378,7 +379,7 @@ async fn test_e2e_protect_on_publish_and_check_on_receive() { "E2E check should pass for correctly protected message" ); } - other => panic!("expected Unicast with e2e_status, got {other:?}"), + other => unreachable!("recv_unicast only returns Unicast, got {other:?}"), } client.shut_down(); @@ -429,22 +430,9 @@ async fn test_multiple_subscribers_receive_events() { .expect("publish_event failed"); assert!(sent >= 2, "expected sent >= 2, got {sent}"); - // Both clients should receive the event - let u1 = tokio::time::timeout(std::time::Duration::from_secs(2), updates1.recv()) - .await - .expect("timeout on client1"); - assert!( - matches!(u1, Some(ClientUpdate::Unicast { .. })), - "client1 expected Unicast, got {u1:?}" - ); - - let u2 = tokio::time::timeout(std::time::Duration::from_secs(2), updates2.recv()) - .await - .expect("timeout on client2"); - assert!( - matches!(u2, Some(ClientUpdate::Unicast { .. })), - "client2 expected Unicast, got {u2:?}" - ); + // Both clients should receive the event (skipping any interleaved acks). + recv_unicast(&mut updates1).await; + recv_unicast(&mut updates2).await; client1.shut_down(); client2.shut_down();