Merge pull request #43 from Lochnair/refactor/transactional-opening

refactor(client): make session opening transactional

crates/irtt-client/src/client.rs

@@ -4,6 +4,9 @@ use std::{
     time::{Duration, Instant},
 };
 
+#[cfg(test)]
+use std::cell::Cell;
+
 #[cfg(test)]
 use irtt_proto::{flags, Params, TimestampFields, PROTOCOL_VERSION};
 
@@ -12,7 +15,10 @@ use crate::{
     error::ClientError,
     event::{ClientEvent, OpenOutcome},
     receive::recv_datagram,
-    session::machine::{recv_buffer_size, ProbeSent, SessionMachine, MAX_OPEN_PACKET_SIZE},
+    session::machine::{
+        recv_buffer_size, OpenDatagramDisposition, PreparedOpenAcceptance, ProbeSent,
+        SessionMachine, MAX_OPEN_PACKET_SIZE,
+    },
     socket::{connect_udp_socket, resolve_remote, validate_open_timeouts},
     socket_options::{apply_dscp_to_socket, clear_dscp_on_socket},
     timing::ClientTimestamp,
@@ -28,6 +34,34 @@ enum ProbeScheduleMode {
     Managed { scheduled_at: Instant },
 }
 
+#[derive(Debug)]
+struct PreparedClientOpen {
+    machine: PreparedOpenAcceptance,
+    schedule: Option<ProbeSchedule>,
+    recv_buffer_len: Option<usize>,
+    negotiated_dscp: Option<u8>,
+    previous_dscp: Option<u8>,
+    post_open_recv_timeout: Option<Duration>,
+}
+
+#[derive(Debug)]
+struct PreparedClientOpenFailure {
+    primary: ClientError,
+    machine: PreparedOpenAcceptance,
+}
+
+#[cfg(test)]
+#[derive(Debug, Default)]
+struct ClientTestHooks {
+    fail_open_dscp: Cell<bool>,
+    fail_open_timeout_restore: Cell<bool>,
+    fail_cleanup_send: Cell<bool>,
+    recv_buffer_len_override: Cell<Option<usize>>,
+    fail_dscp_restore: Cell<bool>,
+    last_restored_read_timeout: Cell<Option<Duration>>,
+    prepared_active_session_before_dscp: Cell<bool>,
+}
+
 #[cfg(test)]
 #[derive(Debug, Clone, Copy)]
 pub(crate) struct ProbeSendTimestamps {
@@ -60,6 +94,9 @@ pub struct Client {
     socket: UdpSocket,
     remote: SocketAddr,
     recv_buffer: Vec<u8>,
+    applied_dscp: Option<u8>,
+    #[cfg(test)]
+    test_hooks: ClientTestHooks,
     #[cfg(test)]
     probe_reported_len: Option<usize>,
     #[cfg(test)]
@@ -86,6 +123,9 @@ impl Client {
             socket,
             remote,
             recv_buffer: vec![0_u8; recv_buffer_size(false, None)?],
+            applied_dscp: None,
+            #[cfg(test)]
+            test_hooks: ClientTestHooks::default(),
             #[cfg(test)]
             probe_reported_len: None,
             #[cfg(test)]
@@ -99,66 +139,24 @@ impl Client {
     ///
     /// On success, returns the negotiated open outcome and transitions the
     /// client into either an open probe session or completed no-test state.
-    /// Open attempts use [`ClientConfig::open_timeouts`].
+    /// Open attempts use [`ClientConfig::open_timeouts`]. Malformed, unrelated,
+    /// or unauthenticated datagrams are ignored until the current attempt's
+    /// absolute deadline, so one attempt may consume several datagrams without
+    /// retransmitting. Silence or ignored traffic eventually produces
+    /// [`ClientError::OpenTimeout`], while authenticated incompatibility remains
+    /// terminal.
+    ///
+    /// When a trusted reply allocates a token but later negotiation or socket
+    /// preparation fails, the client sends a best-effort cleanup close and
+    /// preserves the original failure. Failed opening never leaves the session
+    /// machine open.
     pub fn open(&mut self) -> Result<OpenOutcome, ClientError> {
-        let outcome = (|| -> Result<OpenOutcome, ClientError> {
-            let packet = self.runtime.open_packet()?;
-            let mut buf = [0_u8; MAX_OPEN_PACKET_SIZE];
-
-            for timeout in &self.runtime.config().open_timeouts {
-                self.socket.set_read_timeout(Some(*timeout))?;
-                self.socket.send(&packet)?;
-
-                match self.socket.recv(&mut buf) {
-                    Ok(size) => {
-                        let reply = self.runtime.decode_open_reply(&buf[..size])?;
-                        let opened_at = ClientTimestamp::now();
-                        let remote = self.remote;
-                        let has_hmac = self.runtime.has_hmac();
-                        let socket = &self.socket;
-                        let recv_buffer = &mut self.recv_buffer;
-                        let mut schedule = None;
-
-                        let outcome =
-                            self.runtime
-                                .accept_open_reply(reply, opened_at, |negotiated| {
-                                    let size = recv_buffer_size(has_hmac, Some(negotiated))?;
-                                    let negotiated_dscp = u8::try_from(negotiated.params.dscp)
-                                        .map_err(|_| ClientError::InvalidConfig {
-                                            reason: "negotiated dscp must be in range 0..=63"
-                                                .to_owned(),
-                                        })?;
-                                    apply_dscp_to_socket(socket, remote, negotiated_dscp)?;
-                                    recv_buffer.resize(size, 0);
-                                    schedule =
-                                        Some(ProbeSchedule::new(opened_at.mono, negotiated)?);
-                                    Ok(())
-                                })?;
-                        self.schedule = schedule;
-                        return Ok(outcome);
-                    }
-                    Err(err)
-                        if matches!(
-                            err.kind(),
-                            io::ErrorKind::WouldBlock | io::ErrorKind::TimedOut
-                        ) => {}
-                    Err(err) => return Err(ClientError::Socket(err)),
-                }
-            }
-
-            Err(ClientError::OpenTimeout)
-        })();
-
-        let restore = self
-            .socket
-            .set_read_timeout(self.runtime.config().socket_config.recv_timeout);
-
-        match (outcome, restore) {
-            (Ok(outcome), Ok(())) => Ok(outcome),
-            (Ok(_), Err(source)) => Err(ClientError::ReadTimeoutRestore { source }),
-            (Err(err), Ok(())) => Err(err),
-            (Err(err), Err(_)) => Err(err),
+        let result = self.open_transaction();
+        if result.is_err() {
+            let _ =
+                self.restore_open_read_timeout(self.runtime.config().socket_config.recv_timeout);
         }
+        result
     }
 
     /// Send a close request and emit a [`ClientEvent::SessionClosed`] event.
@@ -417,6 +415,231 @@ impl Client {
         ));
         Ok(events)
     }
+
+    fn open_transaction(&mut self) -> Result<OpenOutcome, ClientError> {
+        let request = self.runtime.prepare_open_request()?;
+        let mut buf = [0_u8; MAX_OPEN_PACKET_SIZE];
+        let attempt_count = self.runtime.config().open_timeouts.len();
+
+        for attempt in 0..attempt_count {
+            let timeout = self.runtime.config().open_timeouts[attempt];
+            let deadline = Instant::now()
+                .checked_add(timeout)
+                .ok_or(ClientError::DurationOverflow)?;
+            self.socket.set_read_timeout(Some(timeout))?;
+            self.socket.send(&request.bytes)?;
+
+            loop {
+                let now = Instant::now();
+                let Some(remaining) = deadline.checked_duration_since(now) else {
+                    break;
+                };
+                if remaining.is_zero() {
+                    break;
+                }
+                self.socket.set_read_timeout(Some(remaining))?;
+
+                let datagram = match recv_datagram(&self.socket, &mut buf) {
+                    Ok(datagram) => datagram,
+                    Err(err)
+                        if matches!(
+                            err.kind(),
+                            io::ErrorKind::WouldBlock | io::ErrorKind::TimedOut
+                        ) =>
+                    {
+                        if Instant::now() >= deadline {
+                            break;
+                        }
+                        continue;
+                    }
+                    Err(err) => return Err(ClientError::Socket(err)),
+                };
+                if datagram.received_at.mono > deadline {
+                    break;
+                }
+
+                let reply = match self.runtime.inspect_open_datagram(&buf[..datagram.len])? {
+                    OpenDatagramDisposition::Ignore => continue,
+                    OpenDatagramDisposition::Trusted(reply) => reply,
+                };
+                let machine = match self
+                    .runtime
+                    .prepare_open_acceptance(reply, datagram.received_at)
+                {
+                    Ok(machine) => machine,
+                    Err(failure) => {
+                        self.send_cleanup_close_best_effort(failure.cleanup_close.as_deref());
+                        return Err(failure.primary);
+                    }
+                };
+                let prepared = match self.prepare_client_open(machine, datagram.received_at) {
+                    Ok(prepared) => prepared,
+                    Err(failure) => {
+                        self.send_cleanup_close_best_effort(failure.machine.cleanup_close_packet());
+                        return Err(failure.primary);
+                    }
+                };
+                return self.apply_prepared_open(prepared);
+            }
+        }
+
+        Err(ClientError::OpenTimeout)
+    }
+
+    fn prepare_client_open(
+        &self,
+        machine: PreparedOpenAcceptance,
+        opened_at: ClientTimestamp,
+    ) -> Result<PreparedClientOpen, Box<PreparedClientOpenFailure>> {
+        let Some(negotiated) = machine.normal_negotiated() else {
+            return Ok(PreparedClientOpen {
+                machine,
+                schedule: None,
+                recv_buffer_len: None,
+                negotiated_dscp: None,
+                previous_dscp: self.applied_dscp,
+                post_open_recv_timeout: self.runtime.config().socket_config.recv_timeout,
+            });
+        };
+        let schedule = match ProbeSchedule::new(opened_at.mono, negotiated) {
+            Ok(schedule) => schedule,
+            Err(primary) => return Err(Box::new(PreparedClientOpenFailure { primary, machine })),
+        };
+        let recv_buffer_len = match recv_buffer_size(self.runtime.has_hmac(), Some(negotiated)) {
+            Ok(size) => size,
+            Err(primary) => return Err(Box::new(PreparedClientOpenFailure { primary, machine })),
+        };
+        #[cfg(test)]
+        let recv_buffer_len = self
+            .test_hooks
+            .recv_buffer_len_override
+            .get()
+            .unwrap_or(recv_buffer_len);
+        let negotiated_dscp = match u8::try_from(negotiated.params.dscp) {
+            Ok(dscp) => dscp,
+            Err(_) => {
+                return Err(Box::new(PreparedClientOpenFailure {
+                    primary: ClientError::InvalidConfig {
+                        reason: "negotiated dscp must be in range 0..=63".to_owned(),
+                    },
+                    machine,
+                }));
+            }
+        };
+
+        Ok(PreparedClientOpen {
+            machine,
+            schedule: Some(schedule),
+            recv_buffer_len: Some(recv_buffer_len),
+            negotiated_dscp: Some(negotiated_dscp),
+            previous_dscp: self.applied_dscp,
+            post_open_recv_timeout: self.runtime.config().socket_config.recv_timeout,
+        })
+    }
+
+    fn apply_prepared_open(
+        &mut self,
+        prepared: PreparedClientOpen,
+    ) -> Result<OpenOutcome, ClientError> {
+        let PreparedClientOpen {
+            machine,
+            schedule,
+            recv_buffer_len,
+            negotiated_dscp,
+            previous_dscp,
+            post_open_recv_timeout,
+        } = prepared;
+
+        if let (Some(recv_buffer_len), Some(negotiated_dscp)) = (recv_buffer_len, negotiated_dscp) {
+            let previous_len = self.recv_buffer.len();
+            let additional = recv_buffer_len.saturating_sub(previous_len);
+            if let Err(source) = self.recv_buffer.try_reserve(additional) {
+                self.send_cleanup_close_best_effort(machine.cleanup_close_packet());
+                return Err(ClientError::AllocationFailed {
+                    operation: "negotiated receive buffer",
+                    source,
+                });
+            }
+            self.recv_buffer.resize(recv_buffer_len, 0);
+
+            #[cfg(test)]
+            self.test_hooks
+                .prepared_active_session_before_dscp
+                .set(machine.has_prepared_active_session());
+            if let Err(primary) = self.apply_open_dscp(negotiated_dscp) {
+                self.recv_buffer.truncate(previous_len);
+                self.restore_dscp_best_effort(previous_dscp);
+                self.send_cleanup_close_best_effort(machine.cleanup_close_packet());
+                return Err(primary);
+            }
+            if let Err(source) = self.restore_open_read_timeout(post_open_recv_timeout) {
+                let primary = ClientError::ReadTimeoutRestore { source };
+                self.recv_buffer.truncate(previous_len);
+                self.restore_dscp_best_effort(previous_dscp);
+                self.send_cleanup_close_best_effort(machine.cleanup_close_packet());
+                return Err(primary);
+            }
+
+            let outcome = self.runtime.commit_open(machine);
+            self.schedule = schedule;
+            self.applied_dscp = Some(negotiated_dscp);
+            Ok(outcome)
+        } else {
+            debug_assert!(schedule.is_none());
+            if let Err(source) = self.restore_open_read_timeout(post_open_recv_timeout) {
+                return Err(ClientError::ReadTimeoutRestore { source });
+            }
+            let outcome = self.runtime.commit_open(machine);
+            self.schedule = None;
+            self.applied_dscp = None;
+            Ok(outcome)
+        }
+    }
+
+    fn restore_dscp_best_effort(&self, previous_dscp: Option<u8>) {
+        #[cfg(test)]
+        if self.test_hooks.fail_dscp_restore.replace(false) {
+            return;
+        }
+        let _ = match previous_dscp {
+            Some(dscp) => apply_dscp_to_socket(&self.socket, self.remote, dscp),
+            None => clear_dscp_on_socket(&self.socket, self.remote),
+        };
+    }
+
+    fn send_cleanup_close_best_effort(&self, packet: Option<&[u8]>) {
+        if let Some(packet) = packet {
+            #[cfg(test)]
+            if self.test_hooks.fail_cleanup_send.replace(false) {
+                return;
+            }
+            let _ = self.socket.send(packet);
+        }
+    }
+
+    fn apply_open_dscp(&self, dscp: u8) -> Result<(), ClientError> {
+        #[cfg(test)]
+        if self.test_hooks.fail_open_dscp.replace(false) {
+            return Err(ClientError::SocketOption {
+                operation: "set negotiated DSCP",
+                remote: self.remote,
+                source: io::Error::other("injected negotiated DSCP failure"),
+            });
+        }
+        apply_dscp_to_socket(&self.socket, self.remote, dscp)
+    }
+
+    fn restore_open_read_timeout(&self, timeout: Option<Duration>) -> io::Result<()> {
+        #[cfg(test)]
+        self.test_hooks.last_restored_read_timeout.set(timeout);
+        #[cfg(test)]
+        if self.test_hooks.fail_open_timeout_restore.replace(false) {
+            return Err(io::Error::other(
+                "injected configured read timeout restoration failure",
+            ));
+        }
+        self.socket.set_read_timeout(timeout)
+    }
 }
 
 pub(crate) fn echo_sent_event(

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

@@ -19,45 +19,77 @@ fn hmac_open_success() {
 }
 
 #[test]
-fn hmac_open_rejects_missing_hmac() {
+fn hmac_open_ignores_missing_hmac_before_valid_reply() {
     let key = b"secret".to_vec();
     let mut config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 1)));
-    config.hmac_key = Some(key);
+    config.hmac_key = Some(key.clone());
     let params = params_from_config(&config).unwrap();
     let server = start_fake_server(move |socket, tx| {
         let (_, peer) = recv_request(&socket, &tx);
-        let reply = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, &params, None);
-        socket.send_to(&reply, peer).unwrap();
+        let missing = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, &params, None);
+        let valid = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, &params, Some(&key));
+        socket.send_to(&missing, peer).unwrap();
+        socket.send_to(&valid, peer).unwrap();
     });
     config.server_addr = server.addr.to_string();
     let mut client = Client::connect(config).unwrap();
-    assert!(matches!(
-        client.open(),
-        Err(ClientError::Protocol(
-            irtt_proto::ProtoError::HmacPresenceMismatch
-        ))
-    ));
+    assert_open_started(client.open().unwrap());
+    assert_eq!(server.rx.iter().take(1).count(), 1);
+    assert!(server.rx.try_recv().is_err());
     server.join();
 }
 
 #[test]
-fn hmac_open_rejects_bad_hmac() {
+fn hmac_open_ignores_bad_hmac_before_valid_reply() {
     let key = b"secret".to_vec();
     let wrong_key = b"wrong".to_vec();
     let mut config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 1)));
-    config.hmac_key = Some(key);
+    config.hmac_key = Some(key.clone());
     let params = params_from_config(&config).unwrap();
     let server = start_fake_server(move |socket, tx| {
         let (_, peer) = recv_request(&socket, &tx);
-        let reply = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, &params, Some(&wrong_key));
-        socket.send_to(&reply, peer).unwrap();
+        let bad = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, &params, Some(&wrong_key));
+        let valid = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, &params, Some(&key));
+        socket.send_to(&bad, peer).unwrap();
+        socket.send_to(&valid, peer).unwrap();
     });
     config.server_addr = server.addr.to_string();
     let mut client = Client::connect(config).unwrap();
+    assert_open_started(client.open().unwrap());
+    assert_eq!(server.rx.iter().take(1).count(), 1);
+    assert!(server.rx.try_recv().is_err());
+    server.join();
+}
+
+#[test]
+fn post_token_hmac_negotiation_failure_sends_authenticated_cleanup_close() {
+    let key = b"secret".to_vec();
+    let mut config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 1)));
+    config.hmac_key = Some(key.clone());
+    let mut returned = params_from_config(&config).unwrap();
+    returned.interval_ns += 1;
+    let server_key = key.clone();
+    let server = start_fake_server(move |socket, tx| {
+        let (_, peer) = recv_request(&socket, &tx);
+        let reply = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, &returned, Some(&server_key));
+        socket.send_to(&reply, 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::Protocol(irtt_proto::ProtoError::BadHmac))
+        Err(ClientError::NegotiationRejected { .. })
     ));
+    let packets: Vec<_> = server.rx.iter().take(2).collect();
+    let cleanup = &packets[1];
+    assert_eq!(cleanup[3], flags::FLAG_CLOSE | FLAG_HMAC);
+    verify_hmac(&key, cleanup, HMAC_OFFSET).unwrap();
+    assert_eq!(
+        u64::from_le_bytes(cleanup[4 + HMAC_SIZE..12 + HMAC_SIZE].try_into().unwrap()),
+        TOKEN
+    );
     server.join();
 }
 

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

@@ -13,6 +13,7 @@ fn no_test_success_validates_params() {
     assert_eq!(negotiated.params, params);
     assert!(client.next_send_deadline().is_none());
     assert!(client.schedule.is_none());
+    assert_eq!(client.applied_dscp, None);
     server.join();
 }
 
@@ -21,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

@@ -9,6 +9,26 @@ fn successful_open_handshake() {
 
     let negotiated = assert_open_started(client.open().unwrap());
     assert_eq!(negotiated.params, params);
+    assert!(client.schedule.is_some());
+    assert_eq!(client.applied_dscp, Some(0));
+    server.join();
+}
+
+#[test]
+fn successful_open_restores_configured_receive_timeout() {
+    let config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 1)));
+    let params = params_from_config(&config).unwrap();
+    let server = open_success_server(params);
+    let mut config = default_test_config(server.addr);
+    config.socket_config.recv_timeout = Some(Duration::from_millis(75));
+    let mut client = Client::connect(config).unwrap();
+
+    assert_open_started(client.open().unwrap());
+
+    assert_eq!(
+        client.test_hooks.last_restored_read_timeout.get(),
+        Some(Duration::from_millis(75))
+    );
     server.join();
 }
 
@@ -49,11 +69,19 @@ fn open_timeout_after_all_timeouts() {
     let server = timeout_server(Duration::from_millis(700));
     let config = ClientConfig {
         open_timeouts: vec![Duration::from_millis(200), Duration::from_millis(200)],
+        socket_config: crate::SocketConfig {
+            recv_timeout: Some(Duration::from_millis(50)),
+            ..Default::default()
+        },
         ..default_test_config(server.addr)
     };
     let mut client = Client::connect(config).unwrap();
     assert!(matches!(client.open(), Err(ClientError::OpenTimeout)));
     assert_eq!(server.rx.iter().take(2).count(), 2);
+    assert_eq!(
+        client.test_hooks.last_restored_read_timeout.get(),
+        Some(Duration::from_millis(50))
+    );
     server.join();
 }
 
@@ -69,6 +97,8 @@ fn protocol_version_mismatch_fails() {
         client.open(),
         Err(ClientError::ProtocolVersionMismatch { received: 2, .. })
     ));
+    assert_eq!(server.rx.iter().take(1).count(), 1);
+    assert!(client.runtime.prepare_open_request().is_ok());
     server.join();
 }
 
@@ -83,5 +113,348 @@ fn server_rejection_fails_in_normal_mode() {
     });
     let mut client = Client::connect(default_test_config(server.addr)).unwrap();
     assert!(matches!(client.open(), Err(ClientError::ServerRejected)));
+    assert_eq!(server.rx.iter().take(1).count(), 1);
+    assert!(client.runtime.prepare_open_request().is_ok());
+    server.join();
+}
+
+#[test]
+fn multiple_untrusted_datagrams_before_valid_reply_use_one_attempt() {
+    let params = default_params();
+    let server = start_fake_server(move |socket, tx| {
+        let (_, peer) = recv_request(&socket, &tx);
+        let valid = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, &params, None);
+        let mut wrong_type = valid.clone();
+        wrong_type[3] = FLAG_REPLY;
+        let mut reserved_flags = valid.clone();
+        reserved_flags[3] |= 0x10;
+        let mut bad_magic = valid.clone();
+        bad_magic[0] ^= 0xff;
+        let unexpected_hmac =
+            open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, &params, Some(b"unexpected"));
+
+        for packet in [
+            vec![0_u8],
+            bad_magic,
+            wrong_type,
+            reserved_flags,
+            unexpected_hmac,
+            valid,
+        ] {
+            socket.send_to(&packet, peer).unwrap();
+        }
+    });
+    let mut client = Client::connect(default_test_config(server.addr)).unwrap();
+
+    assert_open_started(client.open().unwrap());
+    assert_eq!(server.rx.iter().take(1).count(), 1);
+    assert!(server.rx.try_recv().is_err());
+    assert!(client.schedule.is_some());
+    server.join();
+}
+
+#[test]
+fn ignored_datagrams_do_not_restart_the_attempt_deadline() {
+    let server = start_fake_server(move |socket, tx| {
+        let (_, peer) = recv_request(&socket, &tx);
+        thread::sleep(Duration::from_millis(150));
+        socket.send_to(&[0_u8], peer).unwrap();
+        thread::sleep(Duration::from_millis(150));
+        socket.send_to(&[0_u8], peer).unwrap();
+        let _ = recv_request(&socket, &tx);
+    });
+    let config = ClientConfig {
+        open_timeouts: vec![Duration::from_millis(250), Duration::from_millis(250)],
+        ..default_test_config(server.addr)
+    };
+    let mut client = Client::connect(config).unwrap();
+    let started = Instant::now();
+
+    assert!(matches!(client.open(), Err(ClientError::OpenTimeout)));
+    let requests: Vec<_> = server.rx.iter().take(2).collect();
+    assert_eq!(requests.len(), 2);
+    assert!(
+        started.elapsed() < Duration::from_millis(700),
+        "ignored datagrams restarted an attempt deadline"
+    );
+    server.join();
+}
+
+#[test]
+fn only_ignored_open_datagrams_eventually_time_out() {
+    let server = start_fake_server(move |socket, tx| {
+        for _ in 0..2 {
+            let (_, peer) = recv_request(&socket, &tx);
+            socket.send_to(&[0_u8], peer).unwrap();
+            socket.send_to(&MAGIC, peer).unwrap();
+        }
+        thread::sleep(Duration::from_millis(300));
+    });
+    let config = ClientConfig {
+        open_timeouts: vec![Duration::from_millis(200), Duration::from_millis(200)],
+        socket_config: crate::SocketConfig {
+            recv_timeout: Some(Duration::from_millis(50)),
+            ..Default::default()
+        },
+        ..default_test_config(server.addr)
+    };
+    let mut client = Client::connect(config).unwrap();
+
+    assert!(matches!(client.open(), Err(ClientError::OpenTimeout)));
+    assert_eq!(server.rx.iter().take(2).count(), 2);
+    assert!(server.rx.try_recv().is_err());
+    assert_eq!(
+        client.test_hooks.last_restored_read_timeout.get(),
+        Some(Duration::from_millis(50))
+    );
+    server.join();
+}
+
+#[test]
+fn trusted_zero_token_normal_reply_is_terminal() {
+    let params = default_params();
+    let server = start_fake_server(move |socket, tx| {
+        let (_, peer) = recv_request(&socket, &tx);
+        socket
+            .send_to(&open_reply(FLAG_OPEN | FLAG_REPLY, 0, &params, None), peer)
+            .unwrap();
+    });
+    let mut client = Client::connect(default_test_config(server.addr)).unwrap();
+
+    assert!(matches!(
+        client.open(),
+        Err(ClientError::Protocol(irtt_proto::ProtoError::ZeroToken))
+    ));
+    assert_eq!(server.rx.iter().take(1).count(), 1);
+    assert!(client.runtime.prepare_open_request().is_ok());
+    assert!(client.schedule.is_none());
+    server.join();
+}
+
+#[test]
+fn opening_deadline_overflow_occurs_before_send() {
+    let server = timeout_server(Duration::from_millis(250));
+    let config = ClientConfig {
+        open_timeouts: vec![Duration::MAX],
+        ..default_test_config(server.addr)
+    };
+    let mut client = Client::connect(config).unwrap();
+
+    assert!(matches!(client.open(), Err(ClientError::DurationOverflow)));
+    assert!(server.rx.try_recv().is_err());
+    server.join();
+}
+
+#[test]
+fn schedule_duration_overflow_cannot_partially_open() {
+    let server = start_fake_server(|_socket, _tx| {});
+    let client = Client::connect(default_test_config(server.addr)).unwrap();
+    let reply = irtt_proto::OpenReply {
+        flags: FLAG_OPEN | FLAG_REPLY,
+        token: TOKEN,
+        params: params_from_config(client.runtime.config()).unwrap(),
+    };
+    let mut latest = Instant::now();
+    let mut step = Duration::MAX;
+    while !step.is_zero() {
+        if let Some(candidate) = latest.checked_add(step) {
+            latest = candidate;
+        }
+        step /= 2;
+    }
+    let opened_at = ClientTimestamp {
+        mono: latest,
+        wall: SystemTime::now(),
+    };
+    let machine = client
+        .runtime
+        .prepare_open_acceptance(reply, opened_at)
+        .unwrap();
+    let previous_recv_len = client.recv_buffer.len();
+
+    let failure = client.prepare_client_open(machine, opened_at).unwrap_err();
+
+    assert!(matches!(
+        failure.primary,
+        ClientError::NegotiationRejected { .. }
+    ));
+    assert!(failure.machine.cleanup_close_packet().is_some());
+    assert!(client.runtime.prepare_open_request().is_ok());
+    assert!(client.schedule.is_none());
+    assert_eq!(client.applied_dscp, None);
+    assert_eq!(client.recv_buffer.len(), previous_recv_len);
+    server.join();
+}
+
+#[test]
+fn post_token_negotiation_failure_sends_cleanup_close() {
+    let mut returned = default_params();
+    returned.interval_ns += 1;
+    let server = start_fake_server(move |socket, tx| {
+        let (_, peer) = recv_request(&socket, &tx);
+        socket
+            .send_to(
+                &open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, &returned, None),
+                peer,
+            )
+            .unwrap();
+        let _ = recv_request(&socket, &tx);
+    });
+    let mut client = Client::connect(default_test_config(server.addr)).unwrap();
+
+    assert!(matches!(
+        client.open(),
+        Err(ClientError::NegotiationRejected { .. })
+    ));
+    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 cleanup_send_failure_does_not_replace_primary_open_error() {
+    let mut returned = default_params();
+    returned.interval_ns += 1;
+    let server = start_fake_server(move |socket, tx| {
+        let (_, peer) = recv_request(&socket, &tx);
+        socket
+            .send_to(
+                &open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, &returned, None),
+                peer,
+            )
+            .unwrap();
+    });
+    let mut client = Client::connect(default_test_config(server.addr)).unwrap();
+    client.test_hooks.fail_cleanup_send.set(true);
+
+    assert!(matches!(
+        client.open(),
+        Err(ClientError::NegotiationRejected { .. })
+    ));
+    assert!(client.runtime.prepare_open_request().is_ok());
+    server.join();
+}
+
+#[test]
+fn receive_buffer_reservation_failure_precedes_dscp_application() {
+    let mut returned = default_params();
+    returned.dscp = 46;
+    let server = start_fake_server(move |socket, tx| {
+        let (_, peer) = recv_request(&socket, &tx);
+        socket
+            .send_to(
+                &open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, &returned, None),
+                peer,
+            )
+            .unwrap();
+        let _ = recv_request(&socket, &tx);
+    });
+    let mut config = default_test_config(server.addr);
+    config.dscp = 46;
+    let mut client = Client::connect(config).unwrap();
+    client
+        .test_hooks
+        .recv_buffer_len_override
+        .set(Some(usize::MAX));
+    client.test_hooks.fail_open_dscp.set(true);
+
+    assert!(matches!(
+        client.open(),
+        Err(ClientError::AllocationFailed {
+            operation: "negotiated receive buffer",
+            ..
+        })
+    ));
+    assert!(client.test_hooks.fail_open_dscp.get());
+    assert!(client.runtime.prepare_open_request().is_ok());
+    assert!(client.schedule.is_none());
+    assert_eq!(client.applied_dscp, None);
+    server.join();
+}
+
+#[test]
+fn dscp_application_failure_leaves_machine_connected() {
+    let params = default_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);
+    });
+    let mut client = Client::connect(default_test_config(server.addr)).unwrap();
+    client.test_hooks.fail_open_dscp.set(true);
+
+    assert!(matches!(
+        client.open(),
+        Err(ClientError::SocketOption {
+            operation: "set negotiated DSCP",
+            ..
+        })
+    ));
+    assert!(client.test_hooks.prepared_active_session_before_dscp.get());
+    assert!(client.runtime.prepare_open_request().is_ok());
+    assert!(client.schedule.is_none());
+    assert_eq!(client.applied_dscp, None);
+    server.join();
+}
+
+#[test]
+fn read_timeout_restoration_failure_leaves_machine_connected() {
+    let params = default_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);
+    });
+    let mut client = Client::connect(default_test_config(server.addr)).unwrap();
+    client.test_hooks.fail_open_timeout_restore.set(true);
+
+    assert!(matches!(
+        client.open(),
+        Err(ClientError::ReadTimeoutRestore { .. })
+    ));
+    assert!(client.runtime.prepare_open_request().is_ok());
+    assert!(client.schedule.is_none());
+    assert_eq!(client.applied_dscp, None);
+    server.join();
+}
+
+#[test]
+fn dscp_rollback_failure_keeps_timeout_restoration_primary() {
+    let params = default_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);
+    });
+    let mut client = Client::connect(default_test_config(server.addr)).unwrap();
+    client.test_hooks.fail_open_timeout_restore.set(true);
+    client.test_hooks.fail_dscp_restore.set(true);
+
+    let error = client.open().unwrap_err();
+
+    assert!(matches!(error, ClientError::ReadTimeoutRestore { .. }));
+    assert!(client.runtime.prepare_open_request().is_ok());
+    assert!(client.schedule.is_none());
+    assert_eq!(client.applied_dscp, None);
     server.join();
 }

crates/irtt-client/src/error.rs

@@ -38,8 +38,8 @@ pub enum ClientError {
     },
     /// Restoring the configured read timeout after open negotiation failed.
     ///
-    /// The session may have opened successfully, but the socket could not be
-    /// returned to the caller's requested receive-timeout behavior.
+    /// The prepared session was not committed, and the client remains
+    /// connected rather than partially open.
     #[error("failed to restore configured socket read timeout after open negotiation: {source}")]
     ReadTimeoutRestore { source: io::Error },
     /// A packet failed IRTT protocol parsing or authentication.

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

@@ -61,6 +61,64 @@ struct ActiveSession {
     completed: CompletedSet,
 }
 
+#[derive(Debug)]
+pub(crate) struct PreparedOpenRequest {
+    pub(crate) bytes: Box<[u8]>,
+}
+
+#[derive(Debug)]
+pub(crate) enum OpenDatagramDisposition {
+    Ignore,
+    Trusted(OpenReply),
+}
+
+#[derive(Debug)]
+pub(crate) struct PreparedOpenAcceptance {
+    next_state: MachineState,
+    outcome: OpenOutcome,
+    cleanup_close: Option<Box<[u8]>>,
+}
+
+impl PreparedOpenAcceptance {
+    pub(crate) fn normal_negotiated(&self) -> Option<&NegotiatedParams> {
+        match &self.next_state {
+            MachineState::Open(session) => Some(&session.negotiated),
+            MachineState::NoTestCompleted => None,
+            MachineState::Connected | MachineState::Closed { .. } => {
+                unreachable!("open acceptance only prepares open or no-test state")
+            }
+        }
+    }
+
+    pub(crate) fn cleanup_close_packet(&self) -> Option<&[u8]> {
+        self.cleanup_close.as_deref()
+    }
+
+    #[cfg(test)]
+    pub(crate) fn has_prepared_active_session(&self) -> bool {
+        matches!(self.next_state, MachineState::Open(_))
+    }
+}
+
+#[derive(Debug)]
+pub(crate) struct OpenAcceptanceFailure {
+    pub(crate) primary: ClientError,
+    pub(crate) cleanup_close: Option<Box<[u8]>>,
+}
+
+impl OpenAcceptanceFailure {
+    fn new(primary: ClientError, cleanup_close: Option<Box<[u8]>>) -> Self {
+        Self {
+            primary,
+            cleanup_close,
+        }
+    }
+
+    fn without_cleanup(primary: ClientError) -> Self {
+        Self::new(primary, None)
+    }
+}
+
 #[derive(Debug)]
 pub(crate) struct PreparedProbe {
     pub(crate) bytes: Box<[u8]>,
@@ -121,58 +179,99 @@ impl SessionMachine {
         self.config.hmac_key.is_some()
     }
 
-    pub(crate) fn open_packet(&self) -> Result<Vec<u8>, ClientError> {
-        match self.state {
-            MachineState::Connected => {}
-            MachineState::Open(_) => return Err(ClientError::AlreadyOpen),
-            MachineState::Closed { .. } => return Err(ClientError::AlreadyClosed),
-            MachineState::NoTestCompleted => return Err(ClientError::AlreadyCompleted),
-        }
-
+    pub(crate) fn prepare_open_request(&self) -> Result<PreparedOpenRequest, ClientError> {
+        self.ensure_connected()?;
         let request = OpenRequest {
             params: self.requested.clone(),
             close: self.config.run_mode == RunMode::NoTest,
         };
-        Ok(encode_open_request(
-            &request,
-            self.config.hmac_key.as_deref(),
-        )?)
+        let bytes = encode_open_request(&request, self.config.hmac_key.as_deref())?;
+        Ok(PreparedOpenRequest {
+            bytes: bytes.into_boxed_slice(),
+        })
     }
 
-    pub(crate) fn decode_open_reply(&self, packet: &[u8]) -> Result<OpenReply, ClientError> {
-        Ok(irtt_proto::decode_open_reply(
-            packet,
-            self.config.hmac_key.as_deref(),
-        )?)
+    pub(crate) fn inspect_open_datagram(
+        &self,
+        packet: &[u8],
+    ) -> Result<OpenDatagramDisposition, ClientError> {
+        self.ensure_connected()?;
+        match irtt_proto::decode_open_reply(packet, self.config.hmac_key.as_deref()) {
+            Ok(reply) => Ok(OpenDatagramDisposition::Trusted(reply)),
+            Err(error @ irtt_proto::ProtoError::ZeroToken) => Err(ClientError::Protocol(error)),
+            Err(
+                error @ (irtt_proto::ProtoError::TruncatedVarint
+                | irtt_proto::ProtoError::VarintOverflow
+                | irtt_proto::ProtoError::InvalidUtf8
+                | irtt_proto::ProtoError::InvalidEnum { .. }
+                | irtt_proto::ProtoError::NegativePacketLength { .. }
+                | irtt_proto::ProtoError::ParameterLengthTooLarge { .. }
+                | irtt_proto::ProtoError::MalformedParams),
+            ) => Err(ClientError::Protocol(error)),
+            Err(_) => Ok(OpenDatagramDisposition::Ignore),
+        }
     }
 
-    pub(crate) fn accept_open_reply<F>(
-        &mut self,
+    pub(crate) fn prepare_open_acceptance(
+        &self,
         reply: OpenReply,
         now: ClientTimestamp,
-        before_normal_open: F,
-    ) -> Result<OpenOutcome, ClientError>
-    where
-        F: FnOnce(&NegotiatedParams) -> Result<(), ClientError>,
-    {
-        if reply.params.protocol_version != PROTOCOL_VERSION {
-            return Err(ClientError::ProtocolVersionMismatch {
-                requested: PROTOCOL_VERSION,
-                received: reply.params.protocol_version,
-            });
-        }
+    ) -> Result<PreparedOpenAcceptance, OpenAcceptanceFailure> {
+        self.ensure_connected()
+            .map_err(OpenAcceptanceFailure::without_cleanup)?;
 
         let reply_is_close = flags::has(reply.flags, flags::FLAG_CLOSE);
+        let cleanup_close = if !reply_is_close && reply.token != 0 {
+            let bytes = encode_close_request(
+                &CloseRequest { token: reply.token },
+                self.config.hmac_key.as_deref(),
+            )
+            .map_err(ClientError::from)
+            .map_err(OpenAcceptanceFailure::without_cleanup)?;
+            Some(bytes.into_boxed_slice())
+        } else {
+            None
+        };
+
+        if reply.params.protocol_version != PROTOCOL_VERSION {
+            return Err(OpenAcceptanceFailure::new(
+                ClientError::ProtocolVersionMismatch {
+                    requested: PROTOCOL_VERSION,
+                    received: reply.params.protocol_version,
+                },
+                cleanup_close,
+            ));
+        }
+
         match self.config.run_mode {
-            RunMode::Normal if reply_is_close => Err(ClientError::ServerRejected),
-            RunMode::Normal if reply.token == 0 => Err(ClientError::ZeroToken),
-            RunMode::Normal => self.accept_normal_open(reply, now, before_normal_open),
-            RunMode::NoTest if !reply_is_close => Err(ClientError::UnexpectedNoTestReply),
-            RunMode::NoTest if reply.token != 0 => {
-                Err(ClientError::NonZeroNoTestToken { token: reply.token })
+            RunMode::Normal if reply_is_close => Err(OpenAcceptanceFailure::new(
+                ClientError::ServerRejected,
+                cleanup_close,
+            )),
+            RunMode::Normal if reply.token == 0 => Err(OpenAcceptanceFailure::new(
+                ClientError::ZeroToken,
+                cleanup_close,
+            )),
+            RunMode::Normal => self.prepare_normal_open(reply, now, cleanup_close),
+            RunMode::NoTest if !reply_is_close => Err(OpenAcceptanceFailure::new(
+                ClientError::UnexpectedNoTestReply,
+                cleanup_close,
+            )),
+            RunMode::NoTest if reply.token != 0 => Err(OpenAcceptanceFailure::new(
+                ClientError::NonZeroNoTestToken { token: reply.token },
+                cleanup_close,
+            )),
+            RunMode::NoTest => self.prepare_no_test_open(reply, now),
         }
-            RunMode::NoTest => self.accept_no_test_open(reply, now),
     }
+
+    pub(crate) fn commit_open(&mut self, prepared: PreparedOpenAcceptance) -> OpenOutcome {
+        debug_assert!(
+            matches!(self.state, MachineState::Connected),
+            "open acceptance commits only from connected state"
+        );
+        self.state = prepared.next_state;
+        prepared.outcome
     }
 
     pub(crate) fn probe_timeout(&self) -> Duration {
@@ -374,24 +473,22 @@ impl SessionMachine {
         self.open_session().map(|_| ())
     }
 
-    fn accept_normal_open<F>(
-        &mut self,
+    fn prepare_normal_open(
+        &self,
         reply: OpenReply,
         now: ClientTimestamp,
-        before_normal_open: F,
-    ) -> Result<OpenOutcome, ClientError>
-    where
-        F: FnOnce(&NegotiatedParams) -> Result<(), ClientError>,
-    {
+        cleanup_close: Option<Box<[u8]>>,
+    ) -> Result<PreparedOpenAcceptance, OpenAcceptanceFailure> {
         let token = reply.token;
-        let negotiated = negotiate_params(
+        let negotiated = match negotiate_params(
             &self.requested,
             reply.params,
             self.config.negotiation_policy,
-        )?;
-        before_normal_open(&negotiated)?;
-
-        self.state = MachineState::Open(Box::new(ActiveSession {
+        ) {
+            Ok(negotiated) => negotiated,
+            Err(primary) => return Err(OpenAcceptanceFailure::new(primary, cleanup_close)),
+        };
+        let next_state = MachineState::Open(Box::new(ActiveSession {
             token,
             negotiated: negotiated.clone(),
             next_wire_seq: 0,
@@ -409,34 +506,44 @@ impl SessionMachine {
             at: now,
         };
 
-        Ok(OpenOutcome::Started {
+        let outcome = OpenOutcome::Started {
             remote: self.remote,
             token,
             negotiated,
             event,
+        };
+        Ok(PreparedOpenAcceptance {
+            next_state,
+            outcome,
+            cleanup_close,
         })
     }
 
-    fn accept_no_test_open(
-        &mut self,
+    fn prepare_no_test_open(
+        &self,
         reply: OpenReply,
         now: ClientTimestamp,
-    ) -> Result<OpenOutcome, ClientError> {
+    ) -> Result<PreparedOpenAcceptance, OpenAcceptanceFailure> {
         let negotiated = negotiate_params(
             &self.requested,
             reply.params,
             self.config.negotiation_policy,
-        )?;
-        self.state = MachineState::NoTestCompleted;
+        )
+        .map_err(OpenAcceptanceFailure::without_cleanup)?;
         let event = ClientEvent::NoTestCompleted {
             remote: self.remote,
             negotiated: negotiated.clone(),
             at: now,
         };
-        Ok(OpenOutcome::NoTestCompleted {
+        let outcome = OpenOutcome::NoTestCompleted {
             remote: self.remote,
             negotiated,
             event,
+        };
+        Ok(PreparedOpenAcceptance {
+            next_state: MachineState::NoTestCompleted,
+            outcome,
+            cleanup_close: None,
         })
     }
 
@@ -620,6 +727,15 @@ impl SessionMachine {
         }
     }
 
+    fn ensure_connected(&self) -> Result<(), ClientError> {
+        match self.state {
+            MachineState::Connected => Ok(()),
+            MachineState::Open(_) => Err(ClientError::AlreadyOpen),
+            MachineState::Closed { .. } => Err(ClientError::AlreadyClosed),
+            MachineState::NoTestCompleted => Err(ClientError::AlreadyCompleted),
+        }
+    }
+
     fn open_session_mut(&mut self) -> Result<&mut ActiveSession, ClientError> {
         match &mut self.state {
             MachineState::Open(session) => Ok(session),
@@ -896,6 +1012,331 @@ mod tests {
         }
     }
 
+    fn connected_machine(config: ClientConfig) -> SessionMachine {
+        SessionMachine::new(config, "127.0.0.1:2112".parse().unwrap()).unwrap()
+    }
+
+    fn normal_open_reply(machine: &SessionMachine, token: u64) -> OpenReply {
+        OpenReply {
+            flags: flags::FLAG_OPEN | flags::FLAG_REPLY,
+            token,
+            params: machine.requested.clone(),
+        }
+    }
+
+    fn encoded_open_reply(
+        machine: &SessionMachine,
+        reply: &OpenReply,
+    ) -> Result<Vec<u8>, irtt_proto::ProtoError> {
+        irtt_proto::encode_open_reply(reply, machine.config.hmac_key.as_deref())
+    }
+
+    #[test]
+    fn prepared_open_request_is_exact_and_inert_when_dropped() {
+        let machine = connected_machine(ClientConfig::default());
+        let first = machine.prepare_open_request().unwrap();
+        let second = machine.prepare_open_request().unwrap();
+
+        assert_eq!(first.bytes, second.bytes);
+        assert!(!first.bytes.is_empty());
+        drop(first);
+        drop(second);
+        assert!(matches!(machine.state, MachineState::Connected));
+    }
+
+    #[test]
+    fn malformed_wrong_direction_and_invalid_flags_are_ignored() {
+        let machine = connected_machine(ClientConfig::default());
+        assert!(matches!(
+            machine.inspect_open_datagram(&[0_u8]),
+            Ok(OpenDatagramDisposition::Ignore)
+        ));
+
+        let request = machine.prepare_open_request().unwrap();
+        assert!(matches!(
+            machine.inspect_open_datagram(&request.bytes),
+            Ok(OpenDatagramDisposition::Ignore)
+        ));
+
+        let reply = normal_open_reply(&machine, 0x1020_3040_5060_7080);
+        let mut reserved = encoded_open_reply(&machine, &reply).unwrap();
+        reserved[3] |= 0x10;
+        assert!(matches!(
+            machine.inspect_open_datagram(&reserved),
+            Ok(OpenDatagramDisposition::Ignore)
+        ));
+        assert!(matches!(machine.state, MachineState::Connected));
+    }
+
+    #[test]
+    fn missing_unexpected_and_bad_hmac_are_ignored() {
+        let key = b"open-key".to_vec();
+        let authenticated = connected_machine(ClientConfig {
+            hmac_key: Some(key.clone()),
+            ..ClientConfig::default()
+        });
+        let reply = normal_open_reply(&authenticated, 0x1020_3040_5060_7080);
+        let plain = irtt_proto::encode_open_reply(&reply, None).unwrap();
+        assert!(matches!(
+            authenticated.inspect_open_datagram(&plain),
+            Ok(OpenDatagramDisposition::Ignore)
+        ));
+
+        let mut bad = encoded_open_reply(&authenticated, &reply).unwrap();
+        *bad.last_mut().unwrap() ^= 0x80;
+        assert!(matches!(
+            authenticated.inspect_open_datagram(&bad),
+            Ok(OpenDatagramDisposition::Ignore)
+        ));
+
+        let plain_machine = connected_machine(ClientConfig::default());
+        let unexpected = irtt_proto::encode_open_reply(&reply, Some(&key)).unwrap();
+        assert!(matches!(
+            plain_machine.inspect_open_datagram(&unexpected),
+            Ok(OpenDatagramDisposition::Ignore)
+        ));
+    }
+
+    #[test]
+    fn authenticated_zero_token_is_trusted_and_terminal() {
+        let key = b"open-key".to_vec();
+        let machine = connected_machine(ClientConfig {
+            hmac_key: Some(key.clone()),
+            ..ClientConfig::default()
+        });
+        let reply = normal_open_reply(&machine, 0x1020_3040_5060_7080);
+        let mut packet = encoded_open_reply(&machine, &reply).unwrap();
+        let token_offset = 4 + irtt_proto::HMAC_SIZE;
+        packet[token_offset..token_offset + 8].fill(0);
+        irtt_proto::compute_hmac_in_place(&key, &mut packet, 4).unwrap();
+
+        assert!(matches!(
+            machine.inspect_open_datagram(&packet),
+            Err(ClientError::Protocol(irtt_proto::ProtoError::ZeroToken))
+        ));
+        assert!(matches!(machine.state, MachineState::Connected));
+    }
+
+    #[test]
+    fn trusted_invalid_parameter_encoding_is_terminal() {
+        let machine = connected_machine(ClientConfig::default());
+        let mut packet = irtt_proto::MAGIC.to_vec();
+        packet.push(flags::FLAG_OPEN | flags::FLAG_REPLY);
+        packet.extend_from_slice(&0x1020_3040_5060_7080_u64.to_le_bytes());
+        packet.push(0x80);
+
+        assert!(matches!(
+            machine.inspect_open_datagram(&packet),
+            Err(ClientError::Protocol(
+                irtt_proto::ProtoError::TruncatedVarint
+            ))
+        ));
+        assert!(matches!(machine.state, MachineState::Connected));
+    }
+
+    #[test]
+    fn trusted_rejection_version_and_run_mode_failures_do_not_open() {
+        let machine = connected_machine(ClientConfig::default());
+        let rejection = OpenReply {
+            flags: flags::FLAG_OPEN | flags::FLAG_REPLY | flags::FLAG_CLOSE,
+            token: 0,
+            params: machine.requested.clone(),
+        };
+        assert!(matches!(
+            machine
+                .prepare_open_acceptance(rejection, timestamp(Instant::now()))
+                .unwrap_err()
+                .primary,
+            ClientError::ServerRejected
+        ));
+
+        let mut version = normal_open_reply(&machine, 0x1020_3040_5060_7080);
+        version.params.protocol_version += 1;
+        assert!(matches!(
+            machine
+                .prepare_open_acceptance(version, timestamp(Instant::now()))
+                .unwrap_err()
+                .primary,
+            ClientError::ProtocolVersionMismatch { .. }
+        ));
+
+        let no_test = connected_machine(ClientConfig {
+            run_mode: RunMode::NoTest,
+            ..ClientConfig::default()
+        });
+        assert!(matches!(
+            no_test
+                .prepare_open_acceptance(
+                    normal_open_reply(&no_test, 0x1020_3040_5060_7080),
+                    timestamp(Instant::now()),
+                )
+                .unwrap_err()
+                .primary,
+            ClientError::UnexpectedNoTestReply
+        ));
+        assert!(matches!(machine.state, MachineState::Connected));
+        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());
+        let prepared = machine
+            .prepare_open_acceptance(
+                normal_open_reply(&machine, 0x1020_3040_5060_7080),
+                timestamp(Instant::now()),
+            )
+            .unwrap();
+
+        assert!(prepared.has_prepared_active_session());
+        assert!(prepared.cleanup_close_packet().is_some());
+        drop(prepared);
+        assert!(matches!(machine.state, MachineState::Connected));
+    }
+
+    #[test]
+    fn commit_open_assigns_prebuilt_state_once() {
+        let mut machine = connected_machine(ClientConfig::default());
+        let token = 0x1020_3040_5060_7080;
+        let prepared = machine
+            .prepare_open_acceptance(
+                normal_open_reply(&machine, token),
+                timestamp(Instant::now()),
+            )
+            .unwrap();
+
+        let outcome = machine.commit_open(prepared);
+
+        assert!(matches!(
+            outcome,
+            OpenOutcome::Started {
+                token: outcome_token,
+                ..
+            } if outcome_token == token
+        ));
+        assert!(machine.is_open());
+        assert!(matches!(
+            machine.prepare_open_request(),
+            Err(ClientError::AlreadyOpen)
+        ));
+    }
+
+    #[test]
+    fn no_test_preparation_completes_without_active_session() {
+        let config = ClientConfig {
+            run_mode: RunMode::NoTest,
+            ..ClientConfig::default()
+        };
+        let mut machine = connected_machine(config);
+        let reply = OpenReply {
+            flags: flags::FLAG_OPEN | flags::FLAG_REPLY | flags::FLAG_CLOSE,
+            token: 0,
+            params: machine.requested.clone(),
+        };
+        let prepared = machine
+            .prepare_open_acceptance(reply, timestamp(Instant::now()))
+            .unwrap();
+
+        assert!(prepared.normal_negotiated().is_none());
+        assert!(prepared.cleanup_close_packet().is_none());
+        assert!(matches!(machine.state, MachineState::Connected));
+        let outcome = machine.commit_open(prepared);
+        assert!(matches!(outcome, OpenOutcome::NoTestCompleted { .. }));
+        assert!(matches!(machine.state, MachineState::NoTestCompleted));
+    }
+
+    #[test]
+    fn cleanup_close_is_preencoded_with_token_and_hmac() {
+        let key = b"cleanup-key".to_vec();
+        let machine = connected_machine(ClientConfig {
+            hmac_key: Some(key.clone()),
+            ..ClientConfig::default()
+        });
+        let token = 0x1020_3040_5060_7080;
+        let prepared = machine
+            .prepare_open_acceptance(
+                normal_open_reply(&machine, token),
+                timestamp(Instant::now()),
+            )
+            .unwrap();
+        let cleanup = prepared.cleanup_close_packet().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
+        );
+        assert!(matches!(machine.state, MachineState::Connected));
+    }
+
     #[test]
     fn dropping_prepared_probe_changes_nothing() {
         let machine = open_machine(4, Duration::from_secs(1));

crates/irtt-client/tests/hmac.rs

@@ -5,7 +5,7 @@ use std::time::Duration;
 use irtt_client::{
     Client, ClientConfig, ClientError, ClientEvent, OpenOutcome, SocketConfig, WarningKind,
 };
-use irtt_proto::{ProtoError, TimestampFields};
+use irtt_proto::TimestampFields;
 
 use support::{
     config_for_params, default_params, params_for_modes, run_one_probe_with_config,
@@ -192,7 +192,7 @@ fn bad_hmac_echo_reply_is_rejected_without_echo_reply_event() {
 }
 
 #[test]
-fn hmac_open_reply_with_bad_hmac_fails_with_protocol_error() {
+fn hmac_open_reply_with_bad_hmac_is_ignored_until_timeout() {
     let key = b"compat-secret".to_vec();
     let wrong_key = b"wrong-secret".to_vec();
     let params = default_params();
@@ -201,10 +201,7 @@ fn hmac_open_reply_with_bad_hmac_fails_with_protocol_error() {
     config.hmac_key = Some(key);
 
     let mut client = Client::connect(config).unwrap();
-    assert!(matches!(
-        client.open(),
-        Err(ClientError::Protocol(ProtoError::BadHmac))
-    ));
+    assert!(matches!(client.open(), Err(ClientError::OpenTimeout)));
     server.join();
 }