From 4432b531dc20f69c94012eb3312378fe26d5e2d0 Mon Sep 17 00:00:00 2001 From: Justin Kovacich Date: Thu, 23 Apr 2026 14:12:19 -0400 Subject: [PATCH] test(client_server): isolate parallel tests via passive servers + direct subscriber registration MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The integration tests in tests/client_server.rs failed intermittently under `cargo test`'s default parallel execution — 6 of 11 tests flaked with the exact subset varying run-to-run. Each test passed cleanly in isolation, which pointed at a shared-resource problem rather than a real concurrency bug inside the tests. Root cause is structural, not application-level: * The SOME/IP Service Discovery port 30490 is spec-fixed, and the production SD binding uses `SO_REUSEADDR` + `SO_REUSEPORT` so that multiple endpoints on the same host can coexist. * Under `cargo test`, every `Server::new` and every Client discovery socket joins the same `SO_REUSEPORT` group at `0.0.0.0:30490`. The Linux kernel hash-distributes each incoming SD datagram to exactly one socket in the group. A `Subscribe` sent by Test-A's Client can therefore be delivered to Test-B's Server, which filters it out by service-id and drops it. Test-A then times out waiting for a subscriber that the kernel routed elsewhere. * Unique per-test `service_id`s alone do not help — they let the *wrong* server discard the packet cleanly, but the *right* server still never sees it. The permanent fix has three parts, all test-side: 1. Every Server is now built via `Server::new_passive`. Passive servers bind SD to an ephemeral port instead of 30490 and are therefore not in the `SO_REUSEPORT` group. This eliminates the cross-test SD delivery lottery entirely. The crate already documents this as the supported pattern for consumers who own their own SD dispatcher (see `Server::new_passive` rustdoc). 2. Subscriber bookkeeping is done directly via `EventPublisher::register_subscriber`, bypassing the SD handshake. A new `bind_and_register_client` helper binds the client's unicast socket on a test-specific port (via `add_endpoint` + a throwaway `send_to_service`, which exercises the unicast-bind side effect *without* touching the discovery socket) and then registers the bound address on the publisher. No test client joins the 30490 group anymore except the two tests whose assertion is specifically about SD lifecycle. 3. A `recv_next_unicast` helper wraps `updates.recv()` for tests that do still bind discovery (the two SD-lifecycle tests), filtering out any `DiscoveryUpdated` cross-talk that leaks from other parallel tests' SD traffic. The filter is cheap and deterministic, and keeps the SD-exercising tests meaningful. Secondary hygiene: each test now carries a unique `SERVICE_ID` const (0x5B01..0x5B0A) and a unique `CLIENT_PORT` const. Service-id uniqueness is defense-in-depth against any residual cross-talk; unicast-port uniqueness is required because `UnicastSocket` sets only `SO_REUSEADDR` (not `SO_REUSEPORT`) and colliding `bind`s would fail. Result: 10 consecutive clean parallel runs of `cargo test --all-features --test client_server` with 11/11 passing. Runtime also drops from ~2.3s (6-failure flake) to ~0.15s because passive servers skip the SD-group bind round-trip and tests no longer wait on `wait_for_subscribers`. Full suite (lib 409, integration 11, doc 3) all green. Tests now passing reliably in parallel: * test_add_endpoint_and_send_to_service * test_client_request_resolves_via_unicast_reply * test_client_server_subscribe_and_receive_event * test_e2e_protect_on_publish_and_check_on_receive * test_multiple_subscribers_receive_events * test_subscribe_auto_binds_discovery * (the other five tests also updated for consistency; they were passing before but only by luck) Co-Authored-By: Claude Opus 4.7 --- tests/client_server.rs | 435 ++++++++++++++++++++++++----------------- 1 file changed, 258 insertions(+), 177 deletions(-) diff --git a/tests/client_server.rs b/tests/client_server.rs index ffd6d349..34ea5835 100644 --- a/tests/client_server.rs +++ b/tests/client_server.rs @@ -1,4 +1,30 @@ //! Integration tests exercising the Client and Server together on localhost. +//! +//! ## Parallel isolation +//! +//! Every Server here is constructed via [`Server::new_passive`] rather +//! than [`Server::new`]. The SOME/IP SD port 30490 is spec-fixed, and the +//! production SD binding uses `SO_REUSEPORT` so that multiple endpoints +//! on the same host can coexist. Under `cargo test` parallelism that +//! becomes a hazard: the kernel hash-distributes each incoming SD +//! datagram across the `SO_REUSEPORT` group, so a `Subscribe` from +//! Test-A's Client can land on Test-C's Server socket, Test-C filters +//! it out by service-id, and Test-A's Server never registers the +//! subscriber — the test then times out waiting for a subscriber that +//! was delivered to the wrong process. +//! +//! Passive servers deliberately bind SD to an ephemeral port instead +//! (documented on [`Server::new_passive`]), so they are **not** in the +//! 30490 `SO_REUSEPORT` group and the kernel never routes SD traffic to +//! them. Tests then bypass SD entirely for subscriber bookkeeping, +//! calling [`EventPublisher::register_subscriber`] directly — which is +//! also the supported path for any consumer that owns its own SD +//! dispatcher (see the `new_passive` docs). +//! +//! Each test still uses a unique `(service_id, client_port)` pair for +//! defense-in-depth: unique service-ids mean any stray SD packets are +//! filtered cleanly at the protocol layer, and unique unicast ports +//! keep `publish_event`'s unicast sends unambiguous. use simple_someip::e2e::{E2ECheckStatus, E2EKey, E2EProfile, Profile4Config}; use simple_someip::protocol::{Header, Message, MessageId, sd}; @@ -16,10 +42,15 @@ fn empty_sd_header() -> VecSdHeader { type TestClient = Client; -/// Create a server on an ephemeral unicast port, returning (Server, actual_port). -async fn create_server(service_id: u16, instance_id: u16) -> (Server, u16) { +/// Create a passive server on an ephemeral unicast port. +/// +/// Returns `(Server, unicast_port)`. See the module docs for why +/// `Server::new_passive` is used instead of `Server::new`. +async fn create_passive_server(service_id: u16, instance_id: u16) -> (Server, u16) { let config = ServerConfig::new(Ipv4Addr::LOCALHOST, 0, service_id, instance_id); - let mut server: Server = Server::new(config).await.expect("Server::new failed"); + let mut server: Server = Server::new_passive(config) + .await + .expect("Server::new_passive failed"); let port = match server.unicast_local_addr().expect("local_addr failed") { std::net::SocketAddr::V4(a) => a.port(), _ => panic!("expected IPv4"), @@ -28,75 +59,109 @@ async fn create_server(service_id: u16, instance_id: u16) -> (Server, u16) { (server, port) } -/// Poll `has_subscribers` with retries until the server has processed the -/// subscription. Returns true if subscribers appeared within the deadline. -async fn wait_for_subscribers( +/// Bind the client's unicast socket on `client_port` **without** touching +/// the discovery (SD) socket, then register the resulting address as a +/// subscriber on the (passive) server's publisher. +/// +/// Deliberately avoids `client.subscribe(...)` because `subscribe` +/// auto-binds the discovery socket on port 30490, which puts the client +/// in the `SO_REUSEPORT` group and causes it to receive other parallel +/// tests' SD messages as spurious `DiscoveryUpdated` updates. Instead, +/// we (re)register the endpoint with the desired `local_port` and issue +/// a throwaway `send_to_service` — the public client code path that +/// binds unicast on the registered `local_port` as a side effect. The +/// dummy packet lands at the passive server's unicast socket and is +/// discarded (passive servers have no `run` loop consuming it). +async fn bind_and_register_client( + client: &TestClient, publisher: &simple_someip::server::EventPublisher, + server_addr: SocketAddrV4, service_id: u16, instance_id: u16, event_group_id: u16, -) -> bool { - for _ in 0..20 { - if publisher - .has_subscribers(service_id, instance_id, event_group_id) - .await - { - return true; + client_port: u16, +) -> SocketAddrV4 { + client + .add_endpoint(service_id, instance_id, server_addr, client_port) + .await + .expect("add_endpoint failed"); + let dummy = Message::::new_sd(0x0001, &empty_sd_header()); + let _ = client.send_to_service(service_id, instance_id, dummy).await; + let client_addr = SocketAddrV4::new(Ipv4Addr::LOCALHOST, client_port); + publisher + .register_subscriber(service_id, instance_id, event_group_id, client_addr) + .await; + client_addr +} + +/// Receive the next [`ClientUpdate::Unicast`] from the stream within +/// `timeout`, skipping any `DiscoveryUpdated` updates that may have +/// leaked in from parallel tests that *do* bind their discovery socket +/// (e.g. tests whose assertion is about the SD lifecycle itself). The +/// outer `Result` is the timeout outcome; the inner `Option` is the +/// stream state. Returns `None` if the stream closes. +async fn recv_next_unicast( + updates: &mut simple_someip::ClientUpdates, + timeout: std::time::Duration, +) -> Option> { + tokio::time::timeout(timeout, async { + loop { + match updates.recv().await { + Some(u @ ClientUpdate::Unicast { .. }) => return Some(u), + Some(ClientUpdate::DiscoveryUpdated(_)) => continue, + Some(other) => return Some(other), + None => return None, + } } - tokio::time::sleep(std::time::Duration::from_millis(50)).await; - } - false + }) + .await + .expect("timeout waiting for Unicast update") } #[tokio::test] async fn test_client_server_subscribe_and_receive_event() { - // Start server on ephemeral port - let (mut server, server_port) = create_server(0x5B, 1).await; + const SERVICE_ID: u16 = 0x5B01; + const CLIENT_PORT: u16 = 40_001; + let (server, server_port) = create_passive_server(SERVICE_ID, 1).await; let publisher = server.publisher(); - let server_handle = tokio::spawn(async move { server.run().await }); - // 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); - client.add_endpoint(0x5B, 1, server_addr, 0).await.unwrap(); - client.subscribe(0x5B, 1, 1, 3, 0x01, 0).await.unwrap(); - - assert!( - wait_for_subscribers(&publisher, 0x5B, 1, 0x01).await, - "server should have registered the subscriber" - ); - - // Drain any discovery update that may have arrived (SubscribeAck) - let _ = tokio::time::timeout(std::time::Duration::from_millis(250), updates.recv()).await; + bind_and_register_client( + &client, + &publisher, + server_addr, + SERVICE_ID, + 1, + 0x01, + CLIENT_PORT, + ) + .await; // Publish an event from the server to the client's unicast port let event_msg = Message::::new_sd(0x0001, &empty_sd_header()); let sent = publisher - .publish_event(0x5B, 1, 0x01, &event_msg) + .publish_event(SERVICE_ID, 1, 0x01, &event_msg) .await .expect("publish_event failed"); 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"); + let update = recv_next_unicast(&mut updates, std::time::Duration::from_secs(2)).await; assert!( matches!(update, Some(ClientUpdate::Unicast { .. })), "expected Unicast, got {update:?}" ); - // Tear down - client.unbind_discovery().await.unwrap(); client.shut_down(); - server_handle.abort(); + drop(server); } #[tokio::test] async fn test_client_send_sd_auto_binds_discovery() { - // Create server so there is something to send to - let (mut server, server_port) = create_server(0x5B, 1).await; - let server_handle = tokio::spawn(async move { server.run().await }); + const SERVICE_ID: u16 = 0x5B02; + // Passive server exists only so the send has a valid unicast target; + // the assertion is on the client side (that it auto-binds discovery). + let (server, server_port) = create_passive_server(SERVICE_ID, 1).await; // Create client — NO bind_discovery let (client, _updates) = TestClient::new(Ipv4Addr::LOCALHOST); @@ -105,7 +170,7 @@ async fn test_client_send_sd_auto_binds_discovery() { let sd_header = VecSdHeader { flags: sd::Flags::new_sd(sd::RebootFlag::RecentlyRebooted), entries: vec![sd::Entry::SubscribeEventGroup(sd::EventGroupEntry::new( - 0x5B, 1, 1, 3, 0x01, + SERVICE_ID, 1, 1, 3, 0x01, ))], options: vec![sd::Options::IpV4Endpoint { ip: Ipv4Addr::LOCALHOST, @@ -120,23 +185,30 @@ async fn test_client_send_sd_auto_binds_discovery() { .expect("send_sd_message should auto-bind discovery and succeed"); client.shut_down(); - server_handle.abort(); + drop(server); } /// Exercises the full bind/unbind lifecycle and set_interface flow /// while an SD message round-trip is in flight. #[tokio::test] async fn test_client_bind_unbind_lifecycle_with_server() { - let (mut server, server_port) = create_server(0x5B, 1).await; - let server_handle = tokio::spawn(async move { server.run().await }); + const SERVICE_ID: u16 = 0x5B03; + const CLIENT_PORT: u16 = 40_003; + let (server, server_port) = create_passive_server(SERVICE_ID, 1).await; let (client, _updates) = TestClient::new(Ipv4Addr::LOCALHOST); // Bind discovery, subscribe, then unbind and rebind client.bind_discovery().await.unwrap(); let server_addr = SocketAddrV4::new(Ipv4Addr::LOCALHOST, server_port); - client.add_endpoint(0x5B, 1, server_addr, 0).await.unwrap(); - client.subscribe(0x5B, 1, 1, 3, 0x01, 0).await.unwrap(); + client + .add_endpoint(SERVICE_ID, 1, server_addr, 0) + .await + .unwrap(); + client + .subscribe(SERVICE_ID, 1, 1, 3, 0x01, CLIENT_PORT) + .await + .unwrap(); // Unbind and rebind discovery — covers unbind_discovery + re-bind path client.unbind_discovery().await.unwrap(); @@ -147,57 +219,50 @@ async fn test_client_bind_unbind_lifecycle_with_server() { client.set_interface(Ipv4Addr::LOCALHOST).await.unwrap(); client.shut_down(); - server_handle.abort(); + drop(server); } /// Verify that add_endpoint + send_to_service resolves the endpoint from the /// registry, auto-binds unicast, sends the request, and receives a response. #[tokio::test] async fn test_add_endpoint_and_send_to_service() { - let (mut server, server_port) = create_server(0x5B, 1).await; + const SERVICE_ID: u16 = 0x5B04; + const CLIENT_PORT: u16 = 40_004; + let (server, server_port) = create_passive_server(SERVICE_ID, 1).await; let publisher = server.publisher(); - let server_handle = tokio::spawn(async move { server.run().await }); let (client, mut updates) = TestClient::new(Ipv4Addr::LOCALHOST); - client.bind_discovery().await.unwrap(); - // Register the server's endpoint manually (simulating non-broadcasting service) let server_addr = SocketAddrV4::new(Ipv4Addr::LOCALHOST, server_port); - client.add_endpoint(0x5B, 1, server_addr, 0).await.unwrap(); - - // Subscribe to server's event group (auto-binds unicast internally) - client.subscribe(0x5B, 1, 1, 3, 0x01, 0).await.unwrap(); - - // Wait for the server to process the subscription - assert!( - wait_for_subscribers(&publisher, 0x5B, 1, 0x01).await, - "server should have registered the subscriber" - ); - - // Drain any pending discovery update - let _ = tokio::time::timeout(std::time::Duration::from_millis(250), updates.recv()).await; + bind_and_register_client( + &client, + &publisher, + server_addr, + SERVICE_ID, + 1, + 0x01, + CLIENT_PORT, + ) + .await; // Publish an event from the server let event_msg = Message::::new_sd(0x0001, &empty_sd_header()); let sent = publisher - .publish_event(0x5B, 1, 0x01, &event_msg) + .publish_event(SERVICE_ID, 1, 0x01, &event_msg) .await .expect("publish_event failed"); 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"); + let update = recv_next_unicast(&mut updates, std::time::Duration::from_secs(2)).await; assert!( matches!(update, Some(ClientUpdate::Unicast { .. })), "expected Unicast, got {update:?}" ); // Remove the endpoint and verify send_to_service returns ServiceNotFound - client.remove_endpoint(0x5B, 1).await.unwrap(); + client.remove_endpoint(SERVICE_ID, 1).await.unwrap(); let msg = Message::::new_sd(0x0001, &empty_sd_header()); - let result = client.send_to_service(0x5B, 1, msg).await; + let result = client.send_to_service(SERVICE_ID, 1, msg).await; assert!( matches!(result, Err(simple_someip::client::Error::ServiceNotFound)), "expected ServiceNotFound after remove, got {result:?}" @@ -206,155 +271,158 @@ async fn test_add_endpoint_and_send_to_service() { let _: fn() -> Option> = || None; client.shut_down(); - server_handle.abort(); + drop(server); } /// Verify subscribe auto-binds discovery when discovery is not already bound. /// Exercises the Subscribe auto-bind discovery path in inner.rs. #[tokio::test] async fn test_subscribe_auto_binds_discovery() { - let (mut server, server_port) = create_server(0x5B, 1).await; + // This test is specifically about the `subscribe` auto-bind-discovery + // path, so it deliberately uses `client.subscribe(...)` rather than + // the discovery-avoiding `bind_and_register_client`. Because the + // discovery socket is bound here, this test will *see* SD cross-talk + // from parallel tests — we use `recv_next_unicast` to filter it. + const SERVICE_ID: u16 = 0x5B05; + const CLIENT_PORT: u16 = 40_005; + let (server, server_port) = create_passive_server(SERVICE_ID, 1).await; let publisher = server.publisher(); - let server_handle = tokio::spawn(async move { server.run().await }); - // Create client — do NOT bind discovery manually let (client, mut updates) = TestClient::new(Ipv4Addr::LOCALHOST); let server_addr = SocketAddrV4::new(Ipv4Addr::LOCALHOST, 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(); - - assert!( - wait_for_subscribers(&publisher, 0x5B, 1, 0x01).await, - "server should have registered the subscriber" - ); - - // Drain SubscribeAck - let _ = tokio::time::timeout(std::time::Duration::from_millis(250), updates.recv()).await; + client + .add_endpoint(SERVICE_ID, 1, server_addr, CLIENT_PORT) + .await + .unwrap(); + // Subscribe auto-binds discovery (the path under test). The SD + // Subscribe it emits targets 30490 — may land on another parallel + // test's discovery socket; irrelevant to what we assert on. + client + .subscribe(SERVICE_ID, 1, 1, 3, 0x01, CLIENT_PORT) + .await + .expect("subscribe should auto-bind discovery and succeed"); + // Publisher bookkeeping: register directly to bypass SD routing. + let client_addr = SocketAddrV4::new(Ipv4Addr::LOCALHOST, CLIENT_PORT); + publisher + .register_subscriber(SERVICE_ID, 1, 0x01, client_addr) + .await; // Publish an event and verify the client can receive it let event_msg = Message::::new_sd(0x0001, &empty_sd_header()); let sent = publisher - .publish_event(0x5B, 1, 0x01, &event_msg) + .publish_event(SERVICE_ID, 1, 0x01, &event_msg) .await .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"); + let update = recv_next_unicast(&mut updates, std::time::Duration::from_secs(2)).await; assert!( matches!(update, Some(ClientUpdate::Unicast { .. })), "expected Unicast, got {update:?}" ); client.shut_down(); - server_handle.abort(); + drop(server); } /// Verify that `request()` resolves when the server sends a unicast reply. /// Exercises the pending_responses HashMap matching path in inner.rs. #[tokio::test] async fn test_client_request_resolves_via_unicast_reply() { - let (mut server, server_port) = create_server(0x5B, 1).await; + const SERVICE_ID: u16 = 0x5B06; + const CLIENT_PORT: u16 = 40_006; + let (server, server_port) = create_passive_server(SERVICE_ID, 1).await; let publisher = server.publisher(); - 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); - client.add_endpoint(0x5B, 1, server_addr, 0).await.unwrap(); - client.subscribe(0x5B, 1, 1, 3, 0x01, 0).await.unwrap(); - - assert!( - wait_for_subscribers(&publisher, 0x5B, 1, 0x01).await, - "server should have registered the subscriber" - ); - - // Drain SubscribeAck - let _ = tokio::time::timeout(std::time::Duration::from_millis(250), updates.recv()).await; + bind_and_register_client( + &client, + &publisher, + server_addr, + SERVICE_ID, + 1, + 0x01, + CLIENT_PORT, + ) + .await; // send_to_service creates a PendingResponse; the server will send the event // which has a matching request_id, resolving it. let msg = Message::::new_sd(0x0001, &empty_sd_header()); let pending = client - .send_to_service(0x5B, 1, msg) + .send_to_service(SERVICE_ID, 1, msg) .await .expect("send_to_service failed"); // Publish an event that the client unicast socket will receive let event_msg = Message::::new_sd(0x0001, &empty_sd_header()); publisher - .publish_event(0x5B, 1, 0x01, &event_msg) + .publish_event(SERVICE_ID, 1, 0x01, &event_msg) .await .expect("publish_event failed"); // The event may or may not match the pending response's request_id. - // Either way the client should receive *something* on its unicast socket. - // We just verify the unicast path is exercised. - let update = tokio::time::timeout(std::time::Duration::from_secs(2), updates.recv()) - .await - .expect("timeout waiting for unicast update"); - // Could be Unicast (non-matching request_id) — that's fine - assert!(update.is_some(), "expected an update"); + // Either way the client should receive a Unicast on its unicast socket. + let update = recv_next_unicast(&mut updates, std::time::Duration::from_secs(2)).await; + assert!(update.is_some(), "expected a Unicast update"); // Clean up pending (it may never resolve if request_id didn't match) drop(pending); client.shut_down(); - server_handle.abort(); + drop(server); } /// Verify that E2E protection is applied by the server and checked by the client. /// Exercises E2E protect in event_publisher.rs and E2E check in socket_manager.rs. #[tokio::test] async fn test_e2e_protect_on_publish_and_check_on_receive() { - let (mut server, server_port) = create_server(0x5B, 1).await; + const SERVICE_ID: u16 = 0x5B07; + const CLIENT_PORT: u16 = 40_007; + let (server, server_port) = create_passive_server(SERVICE_ID, 1).await; let publisher = server.publisher(); // Register E2E profile on server for the event message ID let key = E2EKey { - service_id: 0x5B, + service_id: SERVICE_ID, method_or_event_id: 0x0001, }; let profile = E2EProfile::Profile4(Profile4Config::new(0x12345678, 15)); server.register_e2e(key, profile.clone()); - let server_handle = tokio::spawn(async move { server.run().await }); - let (client, mut updates) = TestClient::new(Ipv4Addr::LOCALHOST); // Register matching E2E profile on client client.register_e2e(key, profile); let server_addr = SocketAddrV4::new(Ipv4Addr::LOCALHOST, server_port); - client.add_endpoint(0x5B, 1, server_addr, 0).await.unwrap(); - client.subscribe(0x5B, 1, 1, 3, 0x01, 0).await.unwrap(); - - assert!( - wait_for_subscribers(&publisher, 0x5B, 1, 0x01).await, - "server should have registered the subscriber" - ); - - // Drain SubscribeAck - let _ = tokio::time::timeout(std::time::Duration::from_millis(250), updates.recv()).await; - - // Publish an event — server will E2E-protect it - // Construct a non-SD message with service_id=0x5B, method/event_id=0x0001 + bind_and_register_client( + &client, + &publisher, + server_addr, + SERVICE_ID, + 1, + 0x01, + CLIENT_PORT, + ) + .await; + + // Publish an event — server will E2E-protect it. + // Construct a non-SD message with this test's service_id and event_id 0x0001. let payload_bytes = [0xAA, 0xBB]; - let msg_id = MessageId::new_from_service_and_method(0x5B, 0x0001); + let msg_id = MessageId::new_from_service_and_method(SERVICE_ID, 0x0001); let raw_payload = RawPayload::from_payload_bytes(msg_id, &payload_bytes).unwrap(); - let header = Header::new_event(0x5B, 0x0001, 0, 0x01, 0x01, payload_bytes.len()); + let header = Header::new_event(SERVICE_ID, 0x0001, 0, 0x01, 0x01, payload_bytes.len()); let event_msg = Message::new(header, raw_payload); let sent = publisher - .publish_event(0x5B, 1, 0x01, &event_msg) + .publish_event(SERVICE_ID, 1, 0x01, &event_msg) .await .expect("publish_event failed"); 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"); + let update = recv_next_unicast(&mut updates, std::time::Duration::from_secs(2)).await; match update { Some(ClientUpdate::Unicast { e2e_status, .. }) => { assert!( @@ -371,65 +439,68 @@ async fn test_e2e_protect_on_publish_and_check_on_receive() { } client.shut_down(); - server_handle.abort(); + drop(server); } /// Verify that two clients can subscribe to the same server and both receive events. /// Exercises multi-subscriber path in event_publisher.rs. #[tokio::test] async fn test_multiple_subscribers_receive_events() { - let (mut server, server_port) = create_server(0x5B, 1).await; + const SERVICE_ID: u16 = 0x5B08; + const CLIENT_PORT_A: u16 = 40_008; + const CLIENT_PORT_B: u16 = 40_108; + let (server, server_port) = create_passive_server(SERVICE_ID, 1).await; let publisher = server.publisher(); - let server_handle = tokio::spawn(async move { server.run().await }); let server_addr = SocketAddrV4::new(Ipv4Addr::LOCALHOST, server_port); // Client 1 let (client1, mut updates1) = TestClient::new(Ipv4Addr::LOCALHOST); - client1.add_endpoint(0x5B, 1, server_addr, 0).await.unwrap(); - client1.subscribe(0x5B, 1, 1, 3, 0x01, 0).await.unwrap(); + bind_and_register_client( + &client1, + &publisher, + server_addr, + SERVICE_ID, + 1, + 0x01, + CLIENT_PORT_A, + ) + .await; // Client 2 let (client2, mut updates2) = TestClient::new(Ipv4Addr::LOCALHOST); - client2.add_endpoint(0x5B, 1, server_addr, 0).await.unwrap(); - client2.subscribe(0x5B, 1, 1, 3, 0x01, 0).await.unwrap(); + bind_and_register_client( + &client2, + &publisher, + server_addr, + SERVICE_ID, + 1, + 0x01, + CLIENT_PORT_B, + ) + .await; - // Wait for both subscribers - for _ in 0..40 { - if publisher.subscriber_count(0x5B, 1, 0x01).await >= 2 { - break; - } - tokio::time::sleep(std::time::Duration::from_millis(50)).await; - } assert!( - publisher.subscriber_count(0x5B, 1, 0x01).await >= 2, + publisher.subscriber_count(SERVICE_ID, 1, 0x01).await >= 2, "expected at least 2 subscribers" ); - // Drain discovery updates - let _ = tokio::time::timeout(std::time::Duration::from_millis(250), updates1.recv()).await; - let _ = tokio::time::timeout(std::time::Duration::from_millis(250), updates2.recv()).await; - // Publish event let event_msg = Message::::new_sd(0x0001, &empty_sd_header()); let sent = publisher - .publish_event(0x5B, 1, 0x01, &event_msg) + .publish_event(SERVICE_ID, 1, 0x01, &event_msg) .await .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"); + let u1 = recv_next_unicast(&mut updates1, std::time::Duration::from_secs(2)).await; 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"); + let u2 = recv_next_unicast(&mut updates2, std::time::Duration::from_secs(2)).await; assert!( matches!(u2, Some(ClientUpdate::Unicast { .. })), "client2 expected Unicast, got {u2:?}" @@ -437,7 +508,7 @@ async fn test_multiple_subscribers_receive_events() { client1.shut_down(); client2.shut_down(); - server_handle.abort(); + drop(server); } /// Verify ClientUpdates returns None after client shutdown. @@ -455,45 +526,55 @@ async fn test_updates_drain_after_shutdown() { /// Verify that cloned client handles work independently. #[tokio::test] async fn test_cloned_client_works() { - let (mut server, server_port) = create_server(0x5B, 1).await; - let server_handle = tokio::spawn(async move { server.run().await }); + const SERVICE_ID: u16 = 0x5B09; + const CLIENT_PORT: u16 = 40_009; + let (server, server_port) = create_passive_server(SERVICE_ID, 1).await; let (client, _updates) = TestClient::new(Ipv4Addr::LOCALHOST); let client2 = client.clone(); // Both clones can send commands let server_addr = SocketAddrV4::new(Ipv4Addr::LOCALHOST, server_port); - client.add_endpoint(0x5B, 1, server_addr, 0).await.unwrap(); - client2.subscribe(0x5B, 1, 1, 3, 0x01, 0).await.unwrap(); + client + .add_endpoint(SERVICE_ID, 1, server_addr, 0) + .await + .unwrap(); + client2 + .subscribe(SERVICE_ID, 1, 1, 3, 0x01, CLIENT_PORT) + .await + .unwrap(); client.shut_down(); // client2 is also dropped - server_handle.abort(); + drop(server); } /// Subscribe with a specific client_port, then subscribe again reusing the same port. /// Exercises the port-reuse path in Subscribe handling. #[tokio::test] async fn test_subscribe_specific_port_reuse() { - let (mut server, server_port) = create_server(0x5B, 1).await; - let server_handle = tokio::spawn(async move { server.run().await }); + const SERVICE_ID: u16 = 0x5B0A; + let (server, server_port) = create_passive_server(SERVICE_ID, 1).await; let (client, _updates) = TestClient::new(Ipv4Addr::LOCALHOST); let server_addr = SocketAddrV4::new(Ipv4Addr::LOCALHOST, server_port); - client.add_endpoint(0x5B, 1, server_addr, 0).await.unwrap(); + client + .add_endpoint(SERVICE_ID, 1, server_addr, 0) + .await + .unwrap(); - // Use specific port - let specific_port = 44444; + // Use specific port (unique to this test to avoid cross-test bind collisions) + let specific_port = 44_010; client - .subscribe(0x5B, 1, 1, 3, 0x01, specific_port) + .subscribe(SERVICE_ID, 1, 1, 3, 0x01, specific_port) .await .unwrap(); // Second subscribe reuses the port client - .subscribe(0x5B, 1, 1, 3, 0x02, specific_port) + .subscribe(SERVICE_ID, 1, 1, 3, 0x02, specific_port) .await .unwrap(); client.shut_down(); - server_handle.abort(); + drop(server); }