diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index fa278bd8..eda01259 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -80,3 +80,29 @@ 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. 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 --no-run + - run: cargo test --all-features --lib 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/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 + ); + } } diff --git a/src/client/socket_manager.rs b/src/client/socket_manager.rs index 13a8f6cb..b6e47b31 100644 --- a/src/client/socket_manager.rs +++ b/src/client/socket_manager.rs @@ -115,6 +115,49 @@ 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. + /// + /// 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>, + ) -> 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); @@ -339,6 +382,139 @@ 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 — 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; + use std::net::{Ipv4Addr, SocketAddr, SocketAddrV4, UdpSocket}; + use std::time::Duration; + use std::vec::Vec; + + let group = crate::protocol::sd::MULTICAST_IP; + + 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)?; + #[cfg(unix)] + 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 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((Ipv4Addr::UNSPECIFIED, 0)))?; + 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; + } + }; + // 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 = { + 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); + // 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); + 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 INADDR_ANY) 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(); diff --git a/src/server/mod.rs b/src/server/mod.rs index 1532b982..d8a2efce 100644 --- a/src/server/mod.rs +++ b/src/server/mod.rs @@ -1857,26 +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(); - let sd_addr = match server.sd_socket.local_addr().unwrap() { - std::net::SocketAddr::V4(v4) => v4, - std::net::SocketAddr::V6(_) => panic!("expected v4 sd socket"), - }; - 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(); diff --git a/tests/client_server.rs b/tests/client_server.rs index ffd6d349..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 { @@ -16,9 +18,20 @@ 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 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. +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(), @@ -48,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 @@ -57,7 +90,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(); @@ -78,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(); @@ -113,7 +140,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 +161,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 +189,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) @@ -186,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(); @@ -219,7 +240,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(); @@ -240,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(); @@ -261,7 +276,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 +341,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(); @@ -352,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" @@ -367,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(); @@ -382,7 +394,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); @@ -418,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(); @@ -462,7 +461,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 +478,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