fix(client): preserve no-test opening semantics

crates/irtt-client/src/client/tests/no_test.rs

@@ -22,13 +22,55 @@ fn no_test_rejects_non_close_open_reply() {
     let mut config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 1)));
     config.run_mode = RunMode::NoTest;
     let params = params_from_config(&config).unwrap();
-    let server = open_success_server(params);
+    let server = start_fake_server(move |socket, tx| {
+        let (_, peer) = recv_request(&socket, &tx);
+        socket
+            .send_to(
+                &open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, &params, None),
+                peer,
+            )
+            .unwrap();
+        let _ = recv_request(&socket, &tx);
+    });
     config.server_addr = server.addr.to_string();
     let mut client = Client::connect(config).unwrap();
     assert!(matches!(
         client.open(),
         Err(ClientError::UnexpectedNoTestReply)
     ));
+    let packets: Vec<_> = server.rx.iter().take(2).collect();
+    assert_eq!(packets[1][3], flags::FLAG_CLOSE);
+    assert_eq!(
+        u64::from_le_bytes(packets[1][4..12].try_into().unwrap()),
+        TOKEN
+    );
+    assert!(client.runtime.prepare_open_request().is_ok());
+    server.join();
+}
+
+#[test]
+fn no_test_cleanup_send_failure_preserves_unexpected_reply() {
+    let mut config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 1)));
+    config.run_mode = RunMode::NoTest;
+    let params = params_from_config(&config).unwrap();
+    let server = start_fake_server(move |socket, tx| {
+        let (_, peer) = recv_request(&socket, &tx);
+        socket
+            .send_to(
+                &open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, &params, None),
+                peer,
+            )
+            .unwrap();
+    });
+    config.server_addr = server.addr.to_string();
+    let mut client = Client::connect(config).unwrap();
+    client.test_hooks.fail_cleanup_send.set(true);
+
+    assert!(matches!(
+        client.open(),
+        Err(ClientError::UnexpectedNoTestReply)
+    ));
+    assert!(client.runtime.prepare_open_request().is_ok());
     server.join();
 }
 

crates/irtt-client/src/client/tests/open.rs

@@ -29,10 +29,6 @@ fn successful_open_restores_configured_receive_timeout() {
         client.test_hooks.last_restored_read_timeout.get(),
         Some(Duration::from_millis(75))
     );
-    assert_eq!(
-        client.socket.read_timeout().unwrap(),
-        Some(Duration::from_millis(75))
-    );
     server.join();
 }
 

crates/irtt-client/src/managed/group.rs

@@ -1092,10 +1092,10 @@ fn prepare_group_open(
     machine: PreparedOpenAcceptance,
     opened_at: ClientTimestamp,
 ) -> Result<PreparedGroupOpen, Box<PreparedGroupOpenFailure>> {
-    if let Err(primary) = validate_group_negotiation(shared, machine.negotiated()) {
-        return Err(Box::new(PreparedGroupOpenFailure { primary, machine }));
-    }
     let schedule = if let Some(negotiated) = machine.normal_negotiated() {
+        if let Err(primary) = validate_group_negotiation(shared, negotiated) {
+            return Err(Box::new(PreparedGroupOpenFailure { primary, machine }));
+        }
         match ProbeSchedule::new(opened_at.mono, negotiated) {
             Ok(schedule) => Some(schedule),
             Err(primary) => return Err(Box::new(PreparedGroupOpenFailure { primary, machine })),
@@ -3379,6 +3379,162 @@ mod tests {
         server.join();
     }
 
+    #[test]
+    fn no_test_non_close_reply_sends_cleanup_and_fails_without_retry() {
+        let interval = Duration::from_millis(10);
+        let params = test_params(None, interval);
+        let socket = UdpSocket::bind("127.0.0.1:0").unwrap();
+        socket
+            .set_read_timeout(Some(Duration::from_secs(1)))
+            .unwrap();
+        let addr = socket.local_addr().unwrap();
+        let (tx, rx) = mpsc::channel();
+        let done = thread::spawn(move || {
+            let (open, peer) = recv_request_timeout(&socket).expect("missing open request");
+            assert_ne!(open[3] & flags::FLAG_CLOSE, 0);
+            tx.send(ServerObservation::Open { at: Instant::now() })
+                .unwrap();
+            socket
+                .send_to(&open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, &params), peer)
+                .unwrap();
+            let (cleanup, _) =
+                recv_request_timeout(&socket).expect("missing no-test cleanup close");
+            assert_eq!(cleanup[3], flags::FLAG_CLOSE);
+            assert_eq!(
+                u64::from_le_bytes(cleanup[4..12].try_into().unwrap()),
+                TOKEN
+            );
+            tx.send(ServerObservation::Close { at: Instant::now() })
+                .unwrap();
+        });
+        let server = FakeServer {
+            addr,
+            _observations: rx,
+            done,
+        };
+        let mut config = group_config(None, interval, ManagedGroupPacing::Staggered);
+        config.client.run_mode = crate::RunMode::NoTest;
+
+        let outcome = ManagedClientGroup::start(config, vec![target("peer", server.addr)])
+            .unwrap()
+            .join()
+            .unwrap();
+
+        assert_eq!(outcome.failed_target_outcomes, 1);
+        assert!(matches!(
+            outcome.targets[0].end_reason,
+            ManagedTargetEndReason::OpenFailed(ManagedTargetFailure {
+                kind: ManagedTargetFailureKind::OpeningProtocol,
+                ..
+            })
+        ));
+        assert_eq!(server._observations.try_iter().collect::<Vec<_>>().len(), 2);
+        server.join();
+    }
+
+    #[test]
+    fn loose_no_test_group_skips_active_interval_and_dscp_policy() {
+        let interval = Duration::from_millis(10);
+        let mut changed_interval = test_params(None, interval);
+        changed_interval.interval_ns += 1;
+        let changed_dscp = test_params(None, interval);
+
+        for (requested_dscp, params) in [(0, changed_interval), (1, changed_dscp)] {
+            let peer = UdpSocket::bind("127.0.0.1:0").unwrap();
+            let remote = peer.local_addr().unwrap();
+            let mut config = group_config(None, interval, ManagedGroupPacing::Staggered).client;
+            config.run_mode = crate::RunMode::NoTest;
+            config.negotiation_policy = NegotiationPolicy::Loose;
+            config.dscp = requested_dscp;
+            let target = Arc::new(Mutex::new(
+                TargetState::new(&config, target("peer", remote), 0, Instant::now()).unwrap(),
+            ));
+            let shared = shared_with_target(&config, EventHub::new(), target.clone());
+            let opened_at = ClientTimestamp::now();
+
+            let mut target = target.lock().expect("target mutex poisoned");
+            let packet = open_reply(FLAG_OPEN | FLAG_REPLY | flags::FLAG_CLOSE, 0, &params);
+            let reply = match target.runtime.inspect_open_datagram(&packet).unwrap() {
+                OpenDatagramDisposition::Trusted(reply) => reply,
+                OpenDatagramDisposition::Ignore => panic!("no-test reply must be trusted"),
+            };
+            let machine = target
+                .runtime
+                .prepare_open_acceptance(reply, opened_at)
+                .unwrap();
+            let prepared = prepare_group_open(&shared, machine, opened_at).unwrap();
+            assert!(prepared.schedule.is_none());
+            let outcome = target.runtime.commit_open(prepared.machine);
+            target.schedule = prepared.schedule;
+            publish_open_outcome(&shared.hub, &mut target, outcome);
+
+            assert!(matches!(target.status, TargetStatus::Finished));
+            assert_eq!(
+                target.final_reason,
+                Some(ManagedTargetEndReason::NoTestComplete)
+            );
+        }
+    }
+
+    #[test]
+    fn normal_and_strict_sessions_preserve_group_negotiation_checks() {
+        let interval = Duration::from_millis(10);
+        let mut changed_interval = test_params(None, interval);
+        changed_interval.interval_ns += 1;
+        let changed_dscp = test_params(None, interval);
+
+        for (requested_dscp, params) in [(0, changed_interval.clone()), (1, changed_dscp)] {
+            let peer = UdpSocket::bind("127.0.0.1:0").unwrap();
+            let remote = peer.local_addr().unwrap();
+            let mut config = group_config(None, interval, ManagedGroupPacing::Staggered).client;
+            config.negotiation_policy = NegotiationPolicy::Loose;
+            config.dscp = requested_dscp;
+            let target = Arc::new(Mutex::new(
+                TargetState::new(&config, target("peer", remote), 0, Instant::now()).unwrap(),
+            ));
+            let shared = shared_with_target(&config, EventHub::new(), target.clone());
+            let opened_at = ClientTimestamp::now();
+            let target = target.lock().expect("target mutex poisoned");
+            let packet = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, &params);
+            let reply = match target.runtime.inspect_open_datagram(&packet).unwrap() {
+                OpenDatagramDisposition::Trusted(reply) => reply,
+                OpenDatagramDisposition::Ignore => panic!("normal reply must be trusted"),
+            };
+            let machine = target
+                .runtime
+                .prepare_open_acceptance(reply, opened_at)
+                .unwrap();
+            let failure = prepare_group_open(&shared, machine, opened_at).unwrap_err();
+            assert!(matches!(
+                failure.primary,
+                ClientError::NegotiationRejected { .. }
+            ));
+        }
+
+        let peer = UdpSocket::bind("127.0.0.1:0").unwrap();
+        let remote = peer.local_addr().unwrap();
+        let mut config = group_config(None, interval, ManagedGroupPacing::Staggered).client;
+        config.run_mode = crate::RunMode::NoTest;
+        let target = TargetState::new(&config, target("peer", remote), 0, Instant::now()).unwrap();
+        let packet = open_reply(
+            FLAG_OPEN | FLAG_REPLY | flags::FLAG_CLOSE,
+            0,
+            &changed_interval,
+        );
+        let reply = match target.runtime.inspect_open_datagram(&packet).unwrap() {
+            OpenDatagramDisposition::Trusted(reply) => reply,
+            OpenDatagramDisposition::Ignore => panic!("strict no-test reply must be trusted"),
+        };
+        let failure = target
+            .runtime
+            .prepare_open_acceptance(reply, ClientTimestamp::now())
+            .unwrap_err();
+        assert!(matches!(
+            failure.primary,
+            ClientError::NegotiationRejected { .. }
+        ));
+    }
+
     #[test]
     fn group_open_deadline_overflow_fails_before_send() {
         let silent_peer = UdpSocket::bind("127.0.0.1:0").unwrap();

crates/irtt-client/src/session/machine.rs

@@ -80,13 +80,6 @@ pub(crate) struct PreparedOpenAcceptance {
 }
 
 impl PreparedOpenAcceptance {
-    pub(crate) fn negotiated(&self) -> &NegotiatedParams {
-        match &self.outcome {
-            OpenOutcome::Started { negotiated, .. }
-            | OpenOutcome::NoTestCompleted { negotiated, .. } => negotiated,
-        }
-    }
-
     pub(crate) fn normal_negotiated(&self) -> Option<&NegotiatedParams> {
         match &self.next_state {
             MachineState::Open(session) => Some(&session.negotiated),
@@ -228,18 +221,17 @@ impl SessionMachine {
             .map_err(OpenAcceptanceFailure::without_cleanup)?;
 
         let reply_is_close = flags::has(reply.flags, flags::FLAG_CLOSE);
-        let cleanup_close =
+        let cleanup_close = if !reply_is_close && reply.token != 0 {
-            if self.config.run_mode == RunMode::Normal && !reply_is_close && reply.token != 0 {
+            let bytes = encode_close_request(
-                let bytes = encode_close_request(
+                &CloseRequest { token: reply.token },
-                    &CloseRequest { token: reply.token },
+                self.config.hmac_key.as_deref(),
-                    self.config.hmac_key.as_deref(),
+            )
-                )
+            .map_err(ClientError::from)
-                .map_err(ClientError::from)
+            .map_err(OpenAcceptanceFailure::without_cleanup)?;
-                .map_err(OpenAcceptanceFailure::without_cleanup)?;
+            Some(bytes.into_boxed_slice())
-                Some(bytes.into_boxed_slice())
+        } else {
-            } else {
+            None
-                None
+        };
-            };
 
         if reply.params.protocol_version != PROTOCOL_VERSION {
             return Err(OpenAcceptanceFailure::new(
@@ -1186,6 +1178,69 @@ mod tests {
         assert!(matches!(no_test.state, MachineState::Connected));
     }
 
+    #[test]
+    fn no_test_non_close_reply_prepares_authenticated_cleanup_without_state_change() {
+        let key = b"cleanup-key".to_vec();
+        let machine = connected_machine(ClientConfig {
+            run_mode: RunMode::NoTest,
+            hmac_key: Some(key.clone()),
+            ..ClientConfig::default()
+        });
+        let token = 0x1020_3040_5060_7080;
+
+        let failure = machine
+            .prepare_open_acceptance(
+                normal_open_reply(&machine, token),
+                timestamp(Instant::now()),
+            )
+            .unwrap_err();
+
+        assert!(matches!(
+            failure.primary,
+            ClientError::UnexpectedNoTestReply
+        ));
+        let cleanup = failure.cleanup_close.as_deref().unwrap();
+        assert_eq!(cleanup[3], flags::FLAG_CLOSE | flags::FLAG_HMAC);
+        irtt_proto::verify_hmac(&key, cleanup, 4).unwrap();
+        assert_eq!(
+            u64::from_le_bytes(
+                cleanup[4 + irtt_proto::HMAC_SIZE..12 + irtt_proto::HMAC_SIZE]
+                    .try_into()
+                    .unwrap()
+            ),
+            token
+        );
+        drop(failure);
+        assert!(matches!(machine.state, MachineState::Connected));
+    }
+
+    #[test]
+    fn no_test_close_replies_do_not_prepare_cleanup() {
+        let machine = connected_machine(ClientConfig {
+            run_mode: RunMode::NoTest,
+            ..ClientConfig::default()
+        });
+        let reply = |token| OpenReply {
+            flags: flags::FLAG_OPEN | flags::FLAG_REPLY | flags::FLAG_CLOSE,
+            token,
+            params: machine.requested.clone(),
+        };
+
+        let prepared = machine
+            .prepare_open_acceptance(reply(0), timestamp(Instant::now()))
+            .unwrap();
+        assert!(prepared.cleanup_close_packet().is_none());
+
+        let failure = machine
+            .prepare_open_acceptance(reply(0x1020_3040_5060_7080), timestamp(Instant::now()))
+            .unwrap_err();
+        assert!(matches!(
+            failure.primary,
+            ClientError::NonZeroNoTestToken { .. }
+        ));
+        assert!(failure.cleanup_close.is_none());
+    }
+
     #[test]
     fn dropping_prepared_acceptance_leaves_connected_state() {
         let machine = connected_machine(ClientConfig::default());