refactor(managed): transact group session opening

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

@@ -225,7 +225,10 @@ fn trusted_zero_token_normal_reply_is_terminal() {
     });
     let mut client = Client::connect(default_test_config(server.addr)).unwrap();
 
-    assert!(matches!(client.open(), Err(ClientError::ZeroToken)));
+    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());

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

@@ -4,7 +4,7 @@ use std::{
     hash::{Hash, Hasher},
     net::{SocketAddr, UdpSocket},
     sync::{
-        atomic::{AtomicU64, Ordering},
+        atomic::{AtomicBool, AtomicU64, Ordering},
         mpsc, Arc, Mutex,
     },
     thread::{self, JoinHandle},
@@ -22,7 +22,10 @@ use crate::{
     event::{ClientEvent, OpenOutcome},
     metadata::ReceiveMeta,
     receive::recv_datagram_from,
-    session::machine::{params_from_config, SessionMachine},
+    session::machine::{
+        params_from_config, OpenDatagramDisposition, PreparedOpenAcceptance, PreparedOpenRequest,
+        SessionMachine,
+    },
     socket::{bind_unconnected_udp_socket, validate_open_timeouts},
     socket_options::apply_dscp_to_socket,
     timing::ClientTimestamp,
@@ -190,6 +193,12 @@ impl ManagedGroupEvent {
 pub type TargetEventSubscription = EventSubscription<ManagedGroupEvent>;
 
 /// Entry point for running a shared-socket multi-target managed client group.
+///
+/// Each opening attempt sends one reusable request and may inspect several
+/// datagrams until its absolute deadline. Malformed, unrelated, and
+/// unauthenticated traffic is ignored; authenticated incompatibility is
+/// terminal. A trusted token followed by group-policy or schedule rejection
+/// triggers a best-effort cleanup close without replacing the primary failure.
 #[derive(Debug)]
 pub struct ManagedClientGroup;
 
@@ -473,6 +482,8 @@ impl ManagedClientGroup {
             cancellation: cancellation.clone(),
             control_tx: control_tx.clone(),
             peer_closed_target_count: peer_closed_target_count.clone(),
+            receiver_processed_through: Mutex::new(now),
+            receiver_progress_waited_on: AtomicBool::new(false),
             requested_interval_ns: requested.interval_ns,
             requested_dscp: requested.dscp,
             family_remote,
@@ -604,11 +615,63 @@ struct GroupShared {
     cancellation: CancellationToken,
     control_tx: mpsc::Sender<ControlMessage>,
     peer_closed_target_count: Arc<AtomicU64>,
+    receiver_processed_through: Mutex<Instant>,
+    receiver_progress_waited_on: AtomicBool,
     requested_interval_ns: i64,
     requested_dscp: i64,
     family_remote: SocketAddr,
 }
 
+struct ReceiverProgressGuard<'a> {
+    shared: &'a GroupShared,
+    processed_through: Instant,
+}
+
+impl<'a> ReceiverProgressGuard<'a> {
+    fn new(shared: &'a GroupShared, processed_through: Instant) -> Self {
+        Self {
+            shared,
+            processed_through,
+        }
+    }
+}
+
+impl Drop for ReceiverProgressGuard<'_> {
+    fn drop(&mut self) {
+        record_receiver_progress(self.shared, self.processed_through);
+    }
+}
+
+fn record_receiver_progress(shared: &GroupShared, processed_through: Instant) {
+    let advanced = {
+        let mut current = shared
+            .receiver_processed_through
+            .lock()
+            .expect("receiver progress mutex poisoned");
+        if processed_through > *current {
+            *current = processed_through;
+            true
+        } else {
+            false
+        }
+    };
+    if advanced
+        && shared
+            .receiver_progress_waited_on
+            .swap(false, Ordering::AcqRel)
+    {
+        let _ = shared.control_tx.send(ControlMessage::Wake);
+    }
+}
+
+fn receiver_processed_deadline(shared: &GroupShared, deadline: Instant) -> bool {
+    *shared
+        .receiver_processed_through
+        .lock()
+        .expect("receiver progress mutex poisoned")
+        >= deadline
+}
+
 struct GroupSchedulerCleanup {
     shared: Arc<GroupShared>,
 }
@@ -636,7 +699,7 @@ struct TargetState {
     runtime: SessionMachine,
     schedule: Option<ProbeSchedule>,
     status: TargetStatus,
-    open_packet: Vec<u8>,
+    open_request: PreparedOpenRequest,
     counters: TargetCounters,
     order: u64,
     final_reason: Option<ManagedTargetEndReason>,
@@ -655,6 +718,8 @@ enum TargetStatus {
     Opening {
         attempt: usize,
         next_send_at: Instant,
+        active_window: Option<OpenAttemptWindow>,
+        awaiting_receiver: bool,
     },
     Active,
     Draining {
@@ -663,6 +728,30 @@ enum TargetStatus {
     Finished,
 }
 
+#[derive(Debug, Clone, Copy)]
+struct OpenAttemptWindow {
+    sent_at: Instant,
+    deadline: Instant,
+}
+
+impl OpenAttemptWindow {
+    fn contains(self, received_at: Instant) -> bool {
+        received_at >= self.sent_at && received_at <= self.deadline
+    }
+}
+
+#[derive(Debug)]
+struct PreparedGroupOpen {
+    machine: PreparedOpenAcceptance,
+    schedule: Option<ProbeSchedule>,
+}
+
+#[derive(Debug)]
+struct PreparedGroupOpenFailure {
+    primary: ClientError,
+    machine: PreparedOpenAcceptance,
+}
+
 #[derive(Debug, Default, Clone, PartialEq, Eq)]
 struct TargetCounters {
     replies_received: u64,
@@ -725,7 +814,7 @@ impl TargetState {
             config.hmac_key = auth.hmac_key.clone();
         }
         let runtime = SessionMachine::new(config, target.remote)?;
-        let open_packet = runtime.open_packet()?;
+        let open_request = runtime.prepare_open_request()?;
         Ok(Self {
             id: target.id,
             remote: target.remote,
@@ -735,8 +824,10 @@ impl TargetState {
             status: TargetStatus::Opening {
                 attempt: 0,
                 next_send_at: now,
+                active_window: None,
+                awaiting_receiver: false,
             },
-            open_packet,
+            open_request,
             counters: TargetCounters::default(),
             order,
             final_reason: None,
@@ -867,6 +958,7 @@ fn run_group_receiver(socket: UdpSocket, shared: Arc<GroupShared>) -> Result<(),
                     std::io::ErrorKind::WouldBlock | std::io::ErrorKind::TimedOut
                 ) =>
             {
+                record_receiver_progress(&shared, Instant::now());
                 continue;
             }
             Err(err) => {
@@ -881,6 +973,7 @@ fn run_group_receiver(socket: UdpSocket, shared: Arc<GroupShared>) -> Result<(),
                 return Err(ClientError::Socket(err));
             }
         };
+        let _progress = ReceiverProgressGuard::new(&shared, datagram.received_at.mono);
 
         let target = {
             let registry = shared
@@ -901,29 +994,56 @@ fn run_group_receiver(socket: UdpSocket, shared: Arc<GroupShared>) -> Result<(),
         {
             let mut target = target.lock().expect("target mutex poisoned");
             match target.status {
-                TargetStatus::Opening { .. } => {
+                TargetStatus::Opening { active_window, .. } => {
+                    if !active_window
+                        .is_some_and(|window| window.contains(datagram.received_at.mono))
+                    {
+                        continue;
+                    }
                     let opened_at = datagram.received_at;
-                    let mut schedule = None;
+                    let reply = match target.runtime.inspect_open_datagram(packet) {
-                    let result = target.runtime.decode_open_reply(packet).and_then(|reply| {
+                        Ok(OpenDatagramDisposition::Ignore) => continue,
-                        target
+                        Ok(OpenDatagramDisposition::Trusted(reply)) => reply,
-                            .runtime
-                            .accept_open_reply(reply, opened_at, |negotiated| {
-                                validate_group_negotiation(&shared, negotiated)?;
-                                schedule = Some(ProbeSchedule::new(opened_at.mono, negotiated)?);
-                                Ok(())
-                            })
-                    });
-
-                    match result {
-                        Ok(outcome) => {
-                            target.schedule = schedule;
-                            publish_open_outcome(&shared.hub, &mut target, outcome);
-                            wake_scheduler = true;
-                        }
                         Err(err) => {
                             target.mark_finished(ManagedTargetEndReason::OpenFailed(
                                 ManagedTargetFailure::opening(&err),
                             ));
+                            let _ = shared.control_tx.send(ControlMessage::Wake);
+                            continue;
+                        }
+                    };
+                    let machine = match target.runtime.prepare_open_acceptance(reply, opened_at) {
+                        Ok(machine) => machine,
+                        Err(failure) => {
+                            send_group_cleanup_close_best_effort(
+                                &socket,
+                                target.remote,
+                                failure.cleanup_close.as_deref(),
+                            );
+                            target.mark_finished(ManagedTargetEndReason::OpenFailed(
+                                ManagedTargetFailure::opening(&failure.primary),
+                            ));
+                            let _ = shared.control_tx.send(ControlMessage::Wake);
+                            continue;
+                        }
+                    };
+
+                    match prepare_group_open(&shared, machine, opened_at) {
+                        Ok(prepared) => {
+                            let outcome = target.runtime.commit_open(prepared.machine);
+                            target.schedule = prepared.schedule;
+                            publish_open_outcome(&shared.hub, &mut target, outcome);
+                            wake_scheduler = true;
+                        }
+                        Err(failure) => {
+                            send_group_cleanup_close_best_effort(
+                                &socket,
+                                target.remote,
+                                failure.machine.cleanup_close_packet(),
+                            );
+                            target.mark_finished(ManagedTargetEndReason::OpenFailed(
+                                ManagedTargetFailure::opening(&failure.primary),
+                            ));
                             wake_scheduler = true;
                         }
                     }
@@ -967,6 +1087,35 @@ fn validate_group_negotiation(
     Ok(())
 }
 
+fn prepare_group_open(
+    shared: &GroupShared,
+    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() {
+        match ProbeSchedule::new(opened_at.mono, negotiated) {
+            Ok(schedule) => Some(schedule),
+            Err(primary) => return Err(Box::new(PreparedGroupOpenFailure { primary, machine })),
+        }
+    } else {
+        None
+    };
+    Ok(PreparedGroupOpen { machine, schedule })
+}
+
+fn send_group_cleanup_close_best_effort(
+    socket: &UdpSocket,
+    remote: SocketAddr,
+    packet: Option<&[u8]>,
+) {
+    if let Some(packet) = packet {
+        let _ = socket.send_to(packet, remote);
+    }
+}
+
 fn publish_open_outcome(
     hub: &EventHub<ManagedGroupEvent>,
     target: &mut TargetState,
@@ -1229,6 +1378,8 @@ fn drive_open_attempts(
         let TargetStatus::Opening {
             attempt,
             next_send_at,
+            active_window,
+            ..
         } = target.status
         else {
             continue;
@@ -1238,6 +1389,21 @@ fn drive_open_attempts(
             continue;
         }
 
+        if attempt > 0 && !receiver_processed_deadline(shared, next_send_at) {
+            shared
+                .receiver_progress_waited_on
+                .store(true, Ordering::Release);
+            if !receiver_processed_deadline(shared, next_send_at) {
+                target.status = TargetStatus::Opening {
+                    attempt,
+                    next_send_at,
+                    active_window,
+                    awaiting_receiver: true,
+                };
+                continue;
+            }
+        }
+
         if attempt >= config.open_timeouts.len() {
             target.mark_finished(ManagedTargetEndReason::OpenFailed(
                 ManagedTargetFailure::opening(&ClientError::OpenTimeout),
@@ -1245,11 +1411,24 @@ fn drive_open_attempts(
             continue;
         }
 
-        match socket.send_to(&target.open_packet, target.remote) {
+        let sent_at = Instant::now();
+        let deadline = match sent_at.checked_add(config.open_timeouts[attempt]) {
+            Some(deadline) => deadline,
+            None => {
+                target.mark_finished(ManagedTargetEndReason::OpenFailed(
+                    ManagedTargetFailure::opening(&ClientError::DurationOverflow),
+                ));
+                continue;
+            }
+        };
+
+        match socket.send_to(&target.open_request.bytes, target.remote) {
             Ok(_) => {
                 target.status = TargetStatus::Opening {
                     attempt: attempt + 1,
-                    next_send_at: now + config.open_timeouts[attempt],
+                    next_send_at: deadline,
+                    active_window: Some(OpenAttemptWindow { sent_at, deadline }),
+                    awaiting_receiver: false,
                 };
             }
             Err(err) => {
@@ -1811,7 +1990,11 @@ fn next_open_deadline(shared: &GroupShared) -> Option<Instant> {
         .filter_map(|target| {
             let target = target.lock().expect("target mutex poisoned");
             match target.status {
-                TargetStatus::Opening { next_send_at, .. } => Some(next_send_at),
+                TargetStatus::Opening {
+                    next_send_at,
+                    awaiting_receiver: false,
+                    ..
+                } => Some(next_send_at),
                 _ => None,
             }
         })
@@ -1846,10 +2029,10 @@ mod tests {
         config::NegotiationPolicy, managed::SubscriberOverflow, EventSubscriptionError, WarningKind,
     };
     use irtt_proto::{
-        echo_packet_len,
+        compute_hmac_in_place, echo_packet_len,
         flags::{self, FLAG_OPEN, FLAG_REPLY},
         layout::PacketLayout,
-        Clock, Params, ReceivedStats, StampAt, TimestampFields, MAGIC, PROTOCOL_VERSION,
+        Clock, Params, ReceivedStats, StampAt, TimestampFields, HMAC_SIZE, MAGIC, PROTOCOL_VERSION,
     };
     use std::sync::mpsc;
 
@@ -1951,16 +2134,19 @@ mod tests {
             TargetState::new(&client_config, target("peer", remote), 0, Instant::now()).unwrap();
 
         let open_packet = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, &params);
-        let open_reply = target.runtime.decode_open_reply(&open_packet).unwrap();
+        let open_reply = match target.runtime.inspect_open_datagram(&open_packet).unwrap() {
+            OpenDatagramDisposition::Trusted(reply) => reply,
+            OpenDatagramDisposition::Ignore => panic!("fixture open reply must be trusted"),
+        };
         let opened_at = ClientTimestamp::now();
-        let mut opened_schedule = None;
+        let prepared = target
-        let open_outcome = target
             .runtime
-            .accept_open_reply(open_reply, opened_at, |negotiated| {
+            .prepare_open_acceptance(open_reply, opened_at)
-                opened_schedule = Some(ProbeSchedule::new(opened_at.mono, negotiated)?);
-                Ok(())
-            })
             .unwrap();
+        let opened_schedule = Some(
+            ProbeSchedule::new(opened_at.mono, prepared.normal_negotiated().unwrap()).unwrap(),
+        );
+        let open_outcome = target.runtime.commit_open(prepared);
         target.schedule = opened_schedule;
         publish_open_outcome(&hub, &mut target, open_outcome);
 
@@ -2033,6 +2219,8 @@ mod tests {
             cancellation: CancellationToken::new(),
             control_tx,
             peer_closed_target_count: Arc::new(AtomicU64::new(0)),
+            receiver_processed_through: Mutex::new(Instant::now()),
+            receiver_progress_waited_on: AtomicBool::new(false),
             requested_interval_ns: requested.interval_ns,
             requested_dscp: requested.dscp,
             family_remote: remote,
@@ -2066,16 +2254,18 @@ mod tests {
         let mut target =
             TargetState::new(config, target("peer", remote), 0, opened_at.mono).unwrap();
         let packet = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, params);
-        let reply = target.runtime.decode_open_reply(&packet).unwrap();
+        let reply = match target.runtime.inspect_open_datagram(&packet).unwrap() {
-        let mut schedule = None;
+            OpenDatagramDisposition::Trusted(reply) => reply,
-        target
+            OpenDatagramDisposition::Ignore => panic!("test open reply must be trusted"),
+        };
+        let prepared = target
             .runtime
-            .accept_open_reply(reply, opened_at, |negotiated| {
+            .prepare_open_acceptance(reply, opened_at)
-                schedule = Some(ProbeSchedule::new(opened_at.mono, negotiated)?);
-                Ok(())
-            })
             .unwrap();
-        target.schedule = schedule;
+        target.schedule = Some(
+            ProbeSchedule::new(opened_at.mono, prepared.normal_negotiated().unwrap()).unwrap(),
+        );
+        target.runtime.commit_open(prepared);
         target.status = TargetStatus::Active;
         target
     }
@@ -2348,6 +2538,17 @@ mod tests {
         packet
     }
 
+    fn hmac_open_reply(flags: u8, token: u64, params: &Params, key: &[u8]) -> Vec<u8> {
+        let mut packet = Vec::new();
+        packet.extend_from_slice(&MAGIC);
+        packet.push(flags | flags::FLAG_HMAC);
+        packet.extend_from_slice(&[0_u8; HMAC_SIZE]);
+        packet.extend_from_slice(&token.to_le_bytes());
+        packet.extend_from_slice(&params.encode());
+        compute_hmac_in_place(key, &mut packet, 4).unwrap();
+        packet
+    }
+
     fn echo_reply_packet(
         token: u64,
         seq: u32,
@@ -3029,6 +3230,393 @@ mod tests {
             .all(|target| target.end_reason.failure().is_some()));
     }
 
+    #[test]
+    fn bad_open_packet_followed_by_valid_reply_succeeds_without_retry() {
+        let interval = Duration::from_millis(10);
+        let params = test_params(None, interval);
+        let socket = UdpSocket::bind("127.0.0.1:0").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] & FLAG_OPEN, 0);
+            tx.send(ServerObservation::Open { at: Instant::now() })
+                .unwrap();
+            socket.send_to(&[0_u8], peer).unwrap();
+            socket
+                .send_to(
+                    &open_reply(FLAG_OPEN | FLAG_REPLY | flags::FLAG_CLOSE, 0, &params),
+                    peer,
+                )
+                .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.successful_target_outcomes, 1);
+        assert_eq!(server._observations.try_iter().collect::<Vec<_>>().len(), 1);
+        server.join();
+    }
+
+    #[test]
+    fn bad_hmac_open_packet_followed_by_valid_reply_succeeds_without_retry() {
+        let interval = Duration::from_millis(10);
+        let params = test_params(None, interval);
+        let key = b"group-secret".to_vec();
+        let wrong_key = b"wrong".to_vec();
+        let socket = UdpSocket::bind("127.0.0.1:0").unwrap();
+        let addr = socket.local_addr().unwrap();
+        let (tx, rx) = mpsc::channel();
+        let server_key = key.clone();
+        let done = thread::spawn(move || {
+            let (open, peer) = recv_request_timeout(&socket).expect("missing open request");
+            assert_ne!(open[3] & FLAG_OPEN, 0);
+            tx.send(ServerObservation::Open { at: Instant::now() })
+                .unwrap();
+            socket
+                .send_to(
+                    &hmac_open_reply(
+                        FLAG_OPEN | FLAG_REPLY | flags::FLAG_CLOSE,
+                        0,
+                        &params,
+                        &wrong_key,
+                    ),
+                    peer,
+                )
+                .unwrap();
+            socket
+                .send_to(
+                    &hmac_open_reply(
+                        FLAG_OPEN | FLAG_REPLY | flags::FLAG_CLOSE,
+                        0,
+                        &params,
+                        &server_key,
+                    ),
+                    peer,
+                )
+                .unwrap();
+        });
+        let server = FakeServer {
+            addr,
+            _observations: rx,
+            done,
+        };
+        let mut config = group_config(None, interval, ManagedGroupPacing::Staggered);
+        config.client.run_mode = crate::RunMode::NoTest;
+        config.client.hmac_key = Some(key);
+
+        let outcome = ManagedClientGroup::start(config, vec![target("peer", server.addr)])
+            .unwrap()
+            .join()
+            .unwrap();
+
+        assert_eq!(outcome.successful_target_outcomes, 1);
+        assert_eq!(server._observations.try_iter().collect::<Vec<_>>().len(), 1);
+        server.join();
+    }
+
+    #[test]
+    fn trusted_group_incompatibility_sends_cleanup_and_fails_without_retry() {
+        let interval = Duration::from_millis(10);
+        let mut returned = test_params(None, interval);
+        returned.interval_ns = duration_ns_i64(Duration::from_millis(20));
+        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] & FLAG_OPEN, 0);
+            tx.send(ServerObservation::Open { at: Instant::now() })
+                .unwrap();
+            socket
+                .send_to(&open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, &returned), peer)
+                .unwrap();
+            let (cleanup, _) =
+                recv_request_timeout(&socket).expect("missing post-token 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.negotiation_policy = NegotiationPolicy::Loose;
+
+        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 group_open_deadline_overflow_fails_before_send() {
+        let silent_peer = UdpSocket::bind("127.0.0.1:0").unwrap();
+        let remote = silent_peer.local_addr().unwrap();
+        let mut config = group_config(
+            None,
+            Duration::from_millis(10),
+            ManagedGroupPacing::Staggered,
+        );
+        config.client.open_timeouts = vec![Duration::MAX];
+
+        let outcome = ManagedClientGroup::start(config, vec![target("peer", remote)])
+            .unwrap()
+            .join()
+            .unwrap();
+
+        assert_eq!(outcome.failed_target_outcomes, 1);
+        let failure = outcome.targets[0].end_reason.failure().unwrap();
+        assert!(failure.message.contains("duration"));
+        silent_peer.set_nonblocking(true).unwrap();
+        let mut packet = [0_u8; 512];
+        assert!(matches!(
+            silent_peer.recv_from(&mut packet),
+            Err(err) if err.kind() == std::io::ErrorKind::WouldBlock
+        ));
+    }
+
+    #[test]
+    fn receiver_progress_is_monotonic_and_wakes_only_waiters() {
+        let interval = Duration::from_millis(10);
+        let config = group_config(None, interval, ManagedGroupPacing::Staggered).client;
+        let peer = UdpSocket::bind("127.0.0.1:0").unwrap();
+        let remote = peer.local_addr().unwrap();
+        let now = Instant::now();
+        let target = Arc::new(Mutex::new(
+            TargetState::new(&config, target("peer", remote), 0, now).unwrap(),
+        ));
+        let shared = shared_with_target(&config, EventHub::new(), target);
+        let first = now + Duration::from_millis(10);
+        let later = now + Duration::from_millis(20);
+
+        record_receiver_progress(&shared, later);
+        record_receiver_progress(&shared, first);
+
+        assert!(receiver_processed_deadline(&shared, later));
+        assert!(!shared.receiver_progress_waited_on.load(Ordering::Acquire));
+    }
+
+    #[test]
+    fn expired_attempt_waits_for_receiver_before_retrying() {
+        let interval = Duration::from_millis(10);
+        let mut config = group_config(None, interval, ManagedGroupPacing::Staggered).client;
+        config.open_timeouts = vec![Duration::from_millis(200), Duration::from_millis(200)];
+        let peer = UdpSocket::bind("127.0.0.1:0").unwrap();
+        peer.set_read_timeout(Some(Duration::from_millis(100)))
+            .unwrap();
+        let remote = peer.local_addr().unwrap();
+        let now = Instant::now();
+        let target = Arc::new(Mutex::new(
+            TargetState::new(&config, target("peer", remote), 0, now).unwrap(),
+        ));
+        let shared = shared_with_target(&config, EventHub::new(), target.clone());
+        let socket = UdpSocket::bind("127.0.0.1:0").unwrap();
+
+        drive_open_attempts(&config, &socket, &shared, now);
+        let deadline = {
+            let target = target.lock().expect("target mutex poisoned");
+            match target.status {
+                TargetStatus::Opening {
+                    attempt: 1,
+                    next_send_at,
+                    ..
+                } => next_send_at,
+                ref status => panic!("unexpected target status after open send: {status:?}"),
+            }
+        };
+        assert!(recv_request_timeout(&peer).is_some());
+
+        drive_open_attempts(&config, &socket, &shared, deadline);
+        assert!(matches!(
+            target.lock().expect("target mutex poisoned").status,
+            TargetStatus::Opening {
+                attempt: 1,
+                awaiting_receiver: true,
+                ..
+            }
+        ));
+        assert!(recv_request_timeout(&peer).is_none());
+
+        record_receiver_progress(&shared, deadline);
+        drive_open_attempts(&config, &socket, &shared, deadline);
+        assert!(matches!(
+            target.lock().expect("target mutex poisoned").status,
+            TargetStatus::Opening {
+                attempt: 2,
+                awaiting_receiver: false,
+                ..
+            }
+        ));
+        assert!(recv_request_timeout(&peer).is_some());
+    }
+
+    #[test]
+    fn in_window_reply_remains_eligible_after_scheduler_deadline() {
+        let interval = Duration::from_millis(10);
+        let config = group_config(None, interval, ManagedGroupPacing::Staggered).client;
+        let params = test_params(None, interval);
+        let peer = UdpSocket::bind("127.0.0.1:0").unwrap();
+        let remote = peer.local_addr().unwrap();
+        let now = Instant::now();
+        let target = Arc::new(Mutex::new(
+            TargetState::new(&config, target("peer", remote), 0, now).unwrap(),
+        ));
+        let shared = shared_with_target(&config, EventHub::new(), target.clone());
+        let socket = UdpSocket::bind("127.0.0.1:0").unwrap();
+
+        drive_open_attempts(&config, &socket, &shared, now);
+        let (window, deadline) = {
+            let target = target.lock().expect("target mutex poisoned");
+            match target.status {
+                TargetStatus::Opening {
+                    active_window: Some(window),
+                    next_send_at,
+                    ..
+                } => (window, next_send_at),
+                ref status => panic!("unexpected opening status: {status:?}"),
+            }
+        };
+        let received_at = ClientTimestamp {
+            wall: std::time::SystemTime::now(),
+            mono: window.deadline,
+        };
+        let progress = ReceiverProgressGuard::new(&shared, received_at.mono);
+
+        drive_open_attempts(&config, &socket, &shared, deadline);
+        assert!(matches!(
+            target.lock().expect("target mutex poisoned").status,
+            TargetStatus::Opening {
+                awaiting_receiver: true,
+                ..
+            }
+        ));
+
+        {
+            let mut target = target.lock().expect("target mutex poisoned");
+            assert!(window.contains(received_at.mono));
+            let reply = match target
+                .runtime
+                .inspect_open_datagram(&open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, &params))
+                .unwrap()
+            {
+                OpenDatagramDisposition::Trusted(reply) => reply,
+                OpenDatagramDisposition::Ignore => panic!("in-window reply must be trusted"),
+            };
+            let machine = target
+                .runtime
+                .prepare_open_acceptance(reply, received_at)
+                .unwrap();
+            let prepared = prepare_group_open(&shared, machine, received_at).unwrap();
+            let outcome = target.runtime.commit_open(prepared.machine);
+            target.schedule = prepared.schedule;
+            publish_open_outcome(&shared.hub, &mut target, outcome);
+        }
+        drop(progress);
+
+        assert!(matches!(
+            target.lock().expect("target mutex poisoned").status,
+            TargetStatus::Active
+        ));
+        assert!(receiver_processed_deadline(&shared, deadline));
+    }
+
+    #[test]
+    fn post_deadline_reply_is_not_eligible_for_expired_attempt() {
+        let now = Instant::now();
+        let window = OpenAttemptWindow {
+            sent_at: now,
+            deadline: now + Duration::from_millis(200),
+        };
+
+        assert!(window.contains(window.sent_at));
+        assert!(window.contains(window.deadline));
+        assert!(!window.contains(window.deadline + Duration::from_nanos(1)));
+    }
+
+    #[test]
+    fn final_open_timeout_waits_until_receiver_processes_through_deadline() {
+        let interval = Duration::from_millis(10);
+        let config = group_config(None, interval, ManagedGroupPacing::Staggered).client;
+        let peer = UdpSocket::bind("127.0.0.1:0").unwrap();
+        let remote = peer.local_addr().unwrap();
+        let now = Instant::now();
+        let target = Arc::new(Mutex::new(
+            TargetState::new(&config, target("peer", remote), 0, now).unwrap(),
+        ));
+        let shared = shared_with_target(&config, EventHub::new(), target.clone());
+        let socket = UdpSocket::bind("127.0.0.1:0").unwrap();
+
+        drive_open_attempts(&config, &socket, &shared, now);
+        let deadline = {
+            let target = target.lock().expect("target mutex poisoned");
+            match target.status {
+                TargetStatus::Opening {
+                    attempt: 1,
+                    next_send_at,
+                    ..
+                } => next_send_at,
+                ref status => panic!("unexpected target status after open send: {status:?}"),
+            }
+        };
+
+        drive_open_attempts(&config, &socket, &shared, deadline);
+        {
+            let target = target.lock().expect("target mutex poisoned");
+            assert!(matches!(
+                target.status,
+                TargetStatus::Opening {
+                    attempt: 1,
+                    awaiting_receiver: true,
+                    ..
+                }
+            ));
+            assert!(target.final_reason.is_none());
+        }
+
+        record_receiver_progress(&shared, deadline);
+        drive_open_attempts(&config, &socket, &shared, deadline);
+        let target = target.lock().expect("target mutex poisoned");
+        assert!(matches!(target.status, TargetStatus::Finished));
+        assert!(matches!(
+            target.final_reason,
+            Some(ManagedTargetEndReason::OpenFailed(ManagedTargetFailure {
+                kind: ManagedTargetFailureKind::OpeningTimeout,
+                ..
+            }))
+        ));
+    }
+
     #[test]
     fn group_probe_separates_permission_committed_send_and_send_call_timing() {
         let duration = Duration::from_secs(1);

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

@@ -205,7 +205,7 @@ impl SessionMachine {
         self.ensure_connected()?;
         match irtt_proto::decode_open_reply(packet, self.config.hmac_key.as_deref()) {
             Ok(reply) => Ok(OpenDatagramDisposition::Trusted(reply)),
-            Err(irtt_proto::ProtoError::ZeroToken) => Err(ClientError::ZeroToken),
+            Err(error @ irtt_proto::ProtoError::ZeroToken) => Err(ClientError::Protocol(error)),
             Err(
                 error @ (irtt_proto::ProtoError::TruncatedVarint
                 | irtt_proto::ProtoError::VarintOverflow
@@ -282,37 +282,6 @@ impl SessionMachine {
         prepared.outcome
     }
 
-    // Temporary private bridges keep this first commit independently compiling.
-    // The blocking and managed migrations remove them before the PR is complete.
-    pub(crate) fn open_packet(&self) -> Result<Vec<u8>, ClientError> {
-        Ok(self.prepare_open_request()?.bytes.into_vec())
-    }
-
-    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 accept_open_reply<F>(
-        &mut self,
-        reply: OpenReply,
-        now: ClientTimestamp,
-        before_normal_open: F,
-    ) -> Result<OpenOutcome, ClientError>
-    where
-        F: FnOnce(&NegotiatedParams) -> Result<(), ClientError>,
-    {
-        let prepared = self
-            .prepare_open_acceptance(reply, now)
-            .map_err(|failure| failure.primary)?;
-        if let Some(negotiated) = prepared.normal_negotiated() {
-            before_normal_open(negotiated)?;
-        }
-        Ok(self.commit_open(prepared))
-    }
-
     pub(crate) fn probe_timeout(&self) -> Duration {
         self.config.probe_timeout
     }
@@ -1151,7 +1120,7 @@ mod tests {
 
         assert!(matches!(
             machine.inspect_open_datagram(&packet),
-            Err(ClientError::ZeroToken)
+            Err(ClientError::Protocol(irtt_proto::ProtoError::ZeroToken))
         ));
         assert!(matches!(machine.state, MachineState::Connected));
     }