From 1bf44a174641c8607ed3f4263f0e24852031b315 Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Mon, 27 Jul 2026 23:58:16 +0700 Subject: [PATCH] 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 | 331 +++++++++++++++++++ crates/irtt-client/src/client/tests/hmac.rs | 11 +- crates/irtt-client/src/event.rs | 5 +- crates/irtt-client/src/runtime.rs | 174 +++++----- 4 files changed, 430 insertions(+), 91 deletions(-) diff --git a/crates/irtt-client/src/client/tests/close.rs b/crates/irtt-client/src/client/tests/close.rs index 4b3d73e..b7152f0 100644 --- a/crates/irtt-client/src/client/tests/close.rs +++ b/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, ¶ms, 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, ¶ms, &TimestampFields::default(), None); + let close = echo_reply_packet_with_flags( + TOKEN, + seq, + ¶ms, + &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, ¶ms, 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, + ¶ms, + &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, ¶ms, 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, + ¶ms, + &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, + ¶ms, + &TimestampFields::default(), + None, + ), + peer, + ) + .unwrap(); + socket + .send_to( + &echo_reply_packet_with_flags( + TOKEN, + first_seq, + ¶ms, + &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, ¶ms, None), + peer, + ) + .unwrap(); + socket + .send_to( + &echo_reply_packet_with_flags( + TOKEN, + 42, + ¶ms, + &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, ¶ms, 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, + ¶ms, + &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(); +} diff --git a/crates/irtt-client/src/client/tests/hmac.rs b/crates/irtt-client/src/client/tests/hmac.rs index b2d3361..a49d106 100644 --- a/crates/irtt-client/src/client/tests/hmac.rs +++ b/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, ¶ms, &ts, Some(&wrong_key)); + let reply_packet = echo_reply_packet_with_flags( + TOKEN, + seq, + ¶ms, + &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(); } diff --git a/crates/irtt-client/src/event.rs b/crates/irtt-client/src/event.rs index 705062b..2bb6803 100644 --- a/crates/irtt-client/src/event.rs +++ b/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 { diff --git a/crates/irtt-client/src/runtime.rs b/crates/irtt-client/src/runtime.rs index 9e679c8..41d819a 100644 --- a/crates/irtt-client/src/runtime.rs +++ b/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) {