fix(client): honor authenticated close before correlation

Treat a decoded, authenticated, token-matched close flag as a session-level signal after classifying the packet. Classification events now deterministically precede SessionClosed for pending, duplicate, retained-timeout, evicted-late, and untracked sequences.

Wrong-token and bad-HMAC close packets remain non-terminal. Public event documentation records the ordering contract, and focused regressions cover every correlation class.

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

@@ -145,3 +145,334 @@ fn normal_echo_reply_does_not_close_session() {
     client.close().unwrap();
     server.join();
 }
+
+#[test]
+fn close_flagged_duplicate_emits_duplicate_then_closes() {
+    let params = default_params();
+    let server = start_fake_server({
+        let params = params.clone();
+        move |socket, tx| {
+            let (_, peer) = recv_request(&socket, &tx);
+            socket
+                .send_to(
+                    &open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, &params, None),
+                    peer,
+                )
+                .unwrap();
+
+            let (request, _) = recv_request(&socket, &tx);
+            let seq = u32::from_le_bytes(request[12..16].try_into().unwrap());
+            let normal = echo_reply_packet(TOKEN, seq, &params, &TimestampFields::default(), None);
+            let close = echo_reply_packet_with_flags(
+                TOKEN,
+                seq,
+                &params,
+                &TimestampFields::default(),
+                None,
+                FLAG_REPLY | flags::FLAG_CLOSE,
+            );
+            socket.send_to(&normal, peer).unwrap();
+            socket.send_to(&close, peer).unwrap();
+        }
+    });
+    let mut client = Client::connect(ClientConfig {
+        socket_config: crate::SocketConfig {
+            recv_timeout: Some(Duration::from_millis(200)),
+            ..Default::default()
+        },
+        ..default_test_config(server.addr)
+    })
+    .unwrap();
+    assert_open_started(client.open().unwrap());
+    client.send_probe().unwrap();
+
+    assert!(matches!(
+        client.recv_once().unwrap().as_slice(),
+        [ClientEvent::EchoReply { seq: 0, .. }]
+    ));
+    assert!(matches!(
+        client.recv_once().unwrap().as_slice(),
+        [
+            ClientEvent::DuplicateReply { seq: 0, .. },
+            ClientEvent::SessionClosed { token: TOKEN, .. }
+        ]
+    ));
+    assert!(client.is_peer_closed());
+    server.join();
+}
+
+#[test]
+fn close_flagged_retained_timeout_emits_late_then_closes() {
+    let params = default_params();
+    let server = start_fake_server({
+        let params = params.clone();
+        move |socket, tx| {
+            let (_, peer) = recv_request(&socket, &tx);
+            socket
+                .send_to(
+                    &open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, &params, None),
+                    peer,
+                )
+                .unwrap();
+
+            let (request, _) = recv_request(&socket, &tx);
+            let seq = u32::from_le_bytes(request[12..16].try_into().unwrap());
+            socket
+                .send_to(
+                    &echo_reply_packet_with_flags(
+                        TOKEN,
+                        seq,
+                        &params,
+                        &TimestampFields::default(),
+                        None,
+                        FLAG_REPLY | flags::FLAG_CLOSE,
+                    ),
+                    peer,
+                )
+                .unwrap();
+        }
+    });
+    let mut client = Client::connect(ClientConfig {
+        probe_timeout: Duration::from_millis(50),
+        socket_config: crate::SocketConfig {
+            recv_timeout: Some(Duration::from_millis(200)),
+            ..Default::default()
+        },
+        ..default_test_config(server.addr)
+    })
+    .unwrap();
+    assert_open_started(client.open().unwrap());
+    let sent = client.send_probe().unwrap();
+    let ClientEvent::EchoSent { sent_at, .. } = &sent[0] else {
+        panic!("expected EchoSent");
+    };
+    assert!(matches!(
+        client
+            .poll_timeouts_at(sent_at.mono + client.probe_timeout())
+            .unwrap()
+            .as_slice(),
+        [ClientEvent::EchoLoss { seq: 0, .. }]
+    ));
+
+    assert!(matches!(
+        client.recv_once().unwrap().as_slice(),
+        [
+            ClientEvent::LateReply {
+                seq: 0,
+                sent_at: Some(_),
+                ..
+            },
+            ClientEvent::SessionClosed { token: TOKEN, .. }
+        ]
+    ));
+    assert!(client.is_peer_closed());
+    server.join();
+}
+
+#[test]
+fn close_flagged_evicted_sequence_emits_late_then_closes() {
+    let params = default_params();
+    let server = start_fake_server({
+        let params = params.clone();
+        move |socket, tx| {
+            let (_, peer) = recv_request(&socket, &tx);
+            socket
+                .send_to(
+                    &open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, &params, None),
+                    peer,
+                )
+                .unwrap();
+
+            let (first, _) = recv_request(&socket, &tx);
+            let first_seq = u32::from_le_bytes(first[12..16].try_into().unwrap());
+            socket
+                .send_to(
+                    &echo_reply_packet(
+                        TOKEN,
+                        first_seq,
+                        &params,
+                        &TimestampFields::default(),
+                        None,
+                    ),
+                    peer,
+                )
+                .unwrap();
+
+            let (second, _) = recv_request(&socket, &tx);
+            let second_seq = u32::from_le_bytes(second[12..16].try_into().unwrap());
+            socket
+                .send_to(
+                    &echo_reply_packet(
+                        TOKEN,
+                        second_seq,
+                        &params,
+                        &TimestampFields::default(),
+                        None,
+                    ),
+                    peer,
+                )
+                .unwrap();
+            socket
+                .send_to(
+                    &echo_reply_packet_with_flags(
+                        TOKEN,
+                        first_seq,
+                        &params,
+                        &TimestampFields::default(),
+                        None,
+                        FLAG_REPLY | flags::FLAG_CLOSE,
+                    ),
+                    peer,
+                )
+                .unwrap();
+        }
+    });
+    let mut client = Client::connect(ClientConfig {
+        max_pending_probes: 1,
+        socket_config: crate::SocketConfig {
+            recv_timeout: Some(Duration::from_millis(200)),
+            ..Default::default()
+        },
+        ..default_test_config(server.addr)
+    })
+    .unwrap();
+    assert_open_started(client.open().unwrap());
+
+    client.send_probe().unwrap();
+    assert!(matches!(
+        client.recv_once().unwrap().as_slice(),
+        [ClientEvent::EchoReply { seq: 0, .. }]
+    ));
+    client.send_probe().unwrap();
+    assert!(matches!(
+        client.recv_once().unwrap().as_slice(),
+        [ClientEvent::EchoReply { seq: 1, .. }]
+    ));
+    assert!(matches!(
+        client.recv_once().unwrap().as_slice(),
+        [
+            ClientEvent::LateReply {
+                seq: 0,
+                sent_at: None,
+                rtt: None,
+                ..
+            },
+            ClientEvent::SessionClosed { token: TOKEN, .. }
+        ]
+    ));
+    assert!(client.is_peer_closed());
+    server.join();
+}
+
+#[test]
+fn close_flagged_untracked_sequence_emits_warning_then_closes() {
+    let params = default_params();
+    let server = start_fake_server({
+        let params = params.clone();
+        move |socket, tx| {
+            let (_, peer) = recv_request(&socket, &tx);
+            socket
+                .send_to(
+                    &open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, &params, None),
+                    peer,
+                )
+                .unwrap();
+            socket
+                .send_to(
+                    &echo_reply_packet_with_flags(
+                        TOKEN,
+                        42,
+                        &params,
+                        &TimestampFields::default(),
+                        None,
+                        FLAG_REPLY | flags::FLAG_CLOSE,
+                    ),
+                    peer,
+                )
+                .unwrap();
+        }
+    });
+    let mut client = Client::connect(ClientConfig {
+        socket_config: crate::SocketConfig {
+            recv_timeout: Some(Duration::from_millis(200)),
+            ..Default::default()
+        },
+        ..default_test_config(server.addr)
+    })
+    .unwrap();
+    assert_open_started(client.open().unwrap());
+
+    assert!(matches!(
+        client.recv_once().unwrap().as_slice(),
+        [
+            ClientEvent::Warning {
+                kind: WarningKind::UntrackedReply,
+                ..
+            },
+            ClientEvent::SessionClosed { token: TOKEN, .. }
+        ]
+    ));
+    assert!(client.is_peer_closed());
+    server.join();
+}
+
+#[test]
+fn wrong_token_close_flag_does_not_close_session() {
+    let params = default_params();
+    let server = start_fake_server({
+        let params = params.clone();
+        move |socket, tx| {
+            let (_, peer) = recv_request(&socket, &tx);
+            socket
+                .send_to(
+                    &open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, &params, None),
+                    peer,
+                )
+                .unwrap();
+
+            let (request, _) = recv_request(&socket, &tx);
+            let seq = u32::from_le_bytes(request[12..16].try_into().unwrap());
+            socket
+                .send_to(
+                    &echo_reply_packet_with_flags(
+                        TOKEN.wrapping_add(1),
+                        seq,
+                        &params,
+                        &TimestampFields::default(),
+                        None,
+                        FLAG_REPLY | flags::FLAG_CLOSE,
+                    ),
+                    peer,
+                )
+                .unwrap();
+
+            let _ = recv_request(&socket, &tx);
+            let _ = recv_request(&socket, &tx);
+        }
+    });
+    let mut client = Client::connect(ClientConfig {
+        socket_config: crate::SocketConfig {
+            recv_timeout: Some(Duration::from_millis(200)),
+            ..Default::default()
+        },
+        ..default_test_config(server.addr)
+    })
+    .unwrap();
+    assert_open_started(client.open().unwrap());
+    client.send_probe().unwrap();
+
+    assert!(matches!(
+        client.recv_once().unwrap().as_slice(),
+        [ClientEvent::Warning {
+            kind: WarningKind::WrongToken,
+            ..
+        }]
+    ));
+    assert!(!client.is_peer_closed());
+    assert!(matches!(
+        client.send_probe().unwrap().as_slice(),
+        [ClientEvent::EchoSent { seq: 1, .. }]
+    ));
+    client.close().unwrap();
+    server.join();
+}

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

@@ -112,7 +112,14 @@ fn bad_hmac_reply_is_dropped() {
                     .unwrap(),
             );
             let ts = TimestampFields::default();
-            let reply_packet = echo_reply_packet(TOKEN, seq, &params, &ts, Some(&wrong_key));
+            let reply_packet = echo_reply_packet_with_flags(
+                TOKEN,
+                seq,
+                &params,
+                &ts,
+                Some(&wrong_key),
+                FLAG_REPLY | flags::FLAG_CLOSE,
+            );
             socket.send_to(&reply_packet, peer).unwrap();
         }
     });
@@ -131,6 +138,8 @@ fn bad_hmac_reply_is_dropped() {
     let events = client.recv_once().unwrap();
     assert_eq!(events.len(), 1);
     assert!(matches!(&events[0], ClientEvent::Warning { .. }));
+    assert!(!client.is_peer_closed());
+    client.close().unwrap();
     server.join();
 }
 

crates/irtt-client/src/event.rs

@@ -66,9 +66,10 @@ pub enum ClientEvent {
         /// Client timestamp when the open reply was accepted.
         at: ClientTimestamp,
     },
-    /// The client sent a close request and considers the session closed.
+    /// The session closed locally or through an authenticated peer close.
     ///
-    /// This is a local lifecycle event, not an acknowledgement from the server.
+    /// For a peer close-flagged echo reply, the reply's measurement or
+    /// classification event is emitted immediately before this event.
     /// Managed sessions disconnect event subscribers after the worker exits,
     /// while leaving already queued events available to drain.
     SessionClosed {

crates/irtt-client/src/runtime.rs

@@ -503,30 +503,74 @@ impl SessionRuntime {
             }]);
         }
 
-        let session = self.session.as_mut().expect("session must exist when Open");
-
         let wire_seq = reply.sequence;
+        let should_close = flags::has(reply.flags, flags::FLAG_CLOSE);
+        let mut events = {
+            let session = self.session.as_mut().expect("session must exist when Open");
 
-        if let Some(pending) = session.pending.remove(wire_seq) {
-            let rtt = compute_rtt(&pending.sent_at, &now, &reply.timestamps);
-            let server_timing = build_server_timing(&reply.timestamps);
-            let one_way = compute_one_way(&pending.sent_at, &now, &reply.timestamps);
-            let received_stats = build_received_stats(&reply);
-            let is_late = session
-                .highest_received_seq
-                .is_some_and(|h| sequence_is_before(wire_seq, h));
-            let highest_seen = session.highest_received_seq.unwrap_or(wire_seq);
+            if let Some(pending) = session.pending.remove(wire_seq) {
+                let rtt = compute_rtt(&pending.sent_at, &now, &reply.timestamps);
+                let server_timing = build_server_timing(&reply.timestamps);
+                let one_way = compute_one_way(&pending.sent_at, &now, &reply.timestamps);
+                let received_stats = build_received_stats(&reply);
+                let is_late = session
+                    .highest_received_seq
+                    .is_some_and(|h| sequence_is_before(wire_seq, h));
+                let highest_seen = session.highest_received_seq.unwrap_or(wire_seq);
 
-            update_highest_received(&mut session.highest_received_seq, wire_seq);
-            session.completed.insert(wire_seq);
+                update_highest_received(&mut session.highest_received_seq, wire_seq);
+                session.completed.insert(wire_seq);
 
-            let mut events = Vec::new();
-            if is_late {
-                events.push(ClientEvent::LateReply {
+                if is_late {
+                    vec![ClientEvent::LateReply {
+                        seq: wire_seq,
+                        highest_seen,
+                        remote: self.remote,
+                        sent_at: Some(pending.sent_at),
+                        received_at: now,
+                        rtt: Some(rtt),
+                        server_timing,
+                        one_way,
+                        received_stats,
+                        bytes: packet_len,
+                        packet_meta: meta.into(),
+                    }]
+                } else {
+                    vec![ClientEvent::EchoReply {
+                        seq: wire_seq,
+                        remote: self.remote,
+                        sent_at: pending.sent_at,
+                        received_at: now,
+                        rtt,
+                        server_timing,
+                        one_way,
+                        received_stats,
+                        bytes: packet_len,
+                        packet_meta: meta.into(),
+                    }]
+                }
+            } else if session.completed.contains(wire_seq) {
+                update_highest_received(&mut session.highest_received_seq, wire_seq);
+                vec![ClientEvent::DuplicateReply {
+                    seq: wire_seq,
+                    remote: self.remote,
+                    received_at: now,
+                    bytes: packet_len,
+                }]
+            } else if let Some(timed_out) = session.timed_out.remove(wire_seq) {
+                let rtt = compute_rtt(&timed_out.sent_at, &now, &reply.timestamps);
+                let server_timing = build_server_timing(&reply.timestamps);
+                let one_way = compute_one_way(&timed_out.sent_at, &now, &reply.timestamps);
+                let received_stats = build_received_stats(&reply);
+                let highest_seen = session.highest_received_seq.unwrap_or(wire_seq);
+                update_highest_received(&mut session.highest_received_seq, wire_seq);
+                session.completed.insert(wire_seq);
+
+                vec![ClientEvent::LateReply {
                     seq: wire_seq,
                     highest_seen,
                     remote: self.remote,
-                    sent_at: Some(pending.sent_at),
+                    sent_at: Some(timed_out.sent_at),
                     received_at: now,
                     rtt: Some(rtt),
                     server_timing,
@@ -534,85 +578,39 @@ impl SessionRuntime {
                     received_stats,
                     bytes: packet_len,
                     packet_meta: meta.into(),
-                });
-            } else {
-                events.push(ClientEvent::EchoReply {
+                }]
+            } else if session
+                .highest_received_seq
+                .is_some_and(|h| sequence_is_before(wire_seq, h))
+            {
+                vec![ClientEvent::LateReply {
                     seq: wire_seq,
+                    highest_seen: session.highest_received_seq.unwrap(),
                     remote: self.remote,
-                    sent_at: pending.sent_at,
+                    sent_at: None,
                     received_at: now,
-                    rtt,
-                    server_timing,
-                    one_way,
-                    received_stats,
+                    rtt: None,
+                    server_timing: build_server_timing(&reply.timestamps),
+                    one_way: None,
+                    received_stats: build_received_stats(&reply),
                     bytes: packet_len,
                     packet_meta: meta.into(),
-                });
+                }]
+            } else {
+                vec![ClientEvent::Warning {
+                    kind: WarningKind::UntrackedReply,
+                    message: format!(
+                        "dropped reply with untracked seq {wire_seq} (no pending or completed entry)"
+                    ),
+                    at: now,
+                }]
             }
-            if flags::has(reply.flags, flags::FLAG_CLOSE) {
-                self.close_from_peer(token, now, &mut events);
-            }
-            Ok(events)
-        } else if session.completed.contains(wire_seq) {
-            update_highest_received(&mut session.highest_received_seq, wire_seq);
-            Ok(vec![ClientEvent::DuplicateReply {
-                seq: wire_seq,
-                remote: self.remote,
-                received_at: now,
-                bytes: packet_len,
-            }])
-        } else if let Some(timed_out) = session.timed_out.remove(wire_seq) {
-            let rtt = compute_rtt(&timed_out.sent_at, &now, &reply.timestamps);
-            let server_timing = build_server_timing(&reply.timestamps);
-            let one_way = compute_one_way(&timed_out.sent_at, &now, &reply.timestamps);
-            let received_stats = build_received_stats(&reply);
-            let highest_seen = session.highest_received_seq.unwrap_or(wire_seq);
-            update_highest_received(&mut session.highest_received_seq, wire_seq);
-            session.completed.insert(wire_seq);
+        };
 
-            let mut events = vec![ClientEvent::LateReply {
-                seq: wire_seq,
-                highest_seen,
-                remote: self.remote,
-                sent_at: Some(timed_out.sent_at),
-                received_at: now,
-                rtt: Some(rtt),
-                server_timing,
-                one_way,
-                received_stats,
-                bytes: packet_len,
-                packet_meta: meta.into(),
-            }];
-            if flags::has(reply.flags, flags::FLAG_CLOSE) {
-                self.close_from_peer(token, now, &mut events);
-            }
-            Ok(events)
-        } else if session
-            .highest_received_seq
-            .is_some_and(|h| sequence_is_before(wire_seq, h))
-        {
-            Ok(vec![ClientEvent::LateReply {
-                seq: wire_seq,
-                highest_seen: session.highest_received_seq.unwrap(),
-                remote: self.remote,
-                sent_at: None,
-                received_at: now,
-                rtt: None,
-                server_timing: build_server_timing(&reply.timestamps),
-                one_way: None,
-                received_stats: build_received_stats(&reply),
-                bytes: packet_len,
-                packet_meta: meta.into(),
-            }])
-        } else {
-            Ok(vec![ClientEvent::Warning {
-                kind: WarningKind::UntrackedReply,
-                message: format!(
-                    "dropped reply with untracked seq {wire_seq} (no pending or completed entry)"
-                ),
-                at: now,
-            }])
+        if should_close {
+            self.close_from_peer(token, now, &mut events);
         }
+        Ok(events)
     }
 
     fn close_from_peer(&mut self, token: u64, now: ClientTimestamp, events: &mut Vec<ClientEvent>) {