diff --git a/crates/irtt-cli/tests/client_multi_target.rs b/crates/irtt-cli/tests/client_multi_target.rs index b85bc62..d6f39af 100644 --- a/crates/irtt-cli/tests/client_multi_target.rs +++ b/crates/irtt-cli/tests/client_multi_target.rs @@ -425,8 +425,26 @@ fn finite_multi_target_peer_close_is_accepted_as_completion() { #[test] fn continuous_all_peer_closed_targets_exit_nonzero() { let params = test_params(None, Duration::from_millis(10)); - let a = start_peer_close_server(params.clone()); - let b = start_peer_close_server(params); + let (a_seen_tx, a_seen_rx) = mpsc::channel(); + let (a_release_tx, a_release_rx) = mpsc::channel(); + let (b_seen_tx, b_seen_rx) = mpsc::channel(); + let (b_release_tx, b_release_rx) = mpsc::channel(); + let a = start_synchronized_peer_close_server(params.clone(), a_seen_tx, a_release_rx); + let b = start_synchronized_peer_close_server(params, b_seen_tx, b_release_rx); + let coordinator = thread::spawn(move || { + a_seen_rx + .recv_timeout(Duration::from_secs(2)) + .expect("timed out waiting for target A's first probe"); + b_seen_rx + .recv_timeout(Duration::from_secs(2)) + .expect("timed out waiting for target B's first probe"); + a_release_tx + .send(()) + .expect("target A stopped waiting for peer-close release"); + b_release_tx + .send(()) + .expect("target B stopped waiting for peer-close release"); + }); let mut command = Command::new(env!("CARGO_BIN_EXE_irtt-cli")); command.args([ "--duration", @@ -446,6 +464,7 @@ fn continuous_all_peer_closed_targets_exit_nonzero() { ]); let output = run_with_timeout(command, Duration::from_secs(3)); + coordinator.join().unwrap(); a.join(); b.join(); @@ -860,6 +879,55 @@ fn start_peer_close_server(params: Params) -> FakeServer { start_gated_peer_close_server(params, Vec::new()) } +fn start_synchronized_peer_close_server( + params: Params, + first_probe_seen_tx: mpsc::Sender<()>, + release_peer_close_rx: mpsc::Receiver<()>, +) -> FakeServer { + let socket = UdpSocket::bind("127.0.0.1:0").unwrap(); + socket + .set_read_timeout(Some(Duration::from_secs(2))) + .unwrap(); + let addr = socket.local_addr().unwrap(); + let done = thread::spawn(move || { + let (_, peer) = recv_request(&socket); + socket + .send_to(&open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms), peer) + .unwrap(); + let (packet, peer) = recv_request(&socket); + let flags = packet[3]; + assert_eq!( + flags, 0, + "target closed before synchronized first peer-close probe" + ); + assert_eq!( + packet.len(), + echo_packet_len(false, ¶ms).unwrap(), + "expected a complete echo probe before synchronized peer-close reply" + ); + let seq = u32::from_le_bytes(packet[12..16].try_into().unwrap()); + first_probe_seen_tx + .send(()) + .expect("peer-close coordinator stopped waiting for first probe"); + release_peer_close_rx + .recv_timeout(Duration::from_secs(2)) + .expect("timed out waiting to release synchronized peer-close reply"); + socket + .send_to( + &echo_reply_packet( + TOKEN, + seq, + ¶ms, + &TimestampFields::default(), + FLAG_REPLY | FLAG_CLOSE, + ), + peer, + ) + .unwrap(); + }); + FakeServer { addr, done } +} + fn start_gated_peer_close_server( params: Params, release_gates: Vec>, @@ -875,10 +943,19 @@ fn start_gated_peer_close_server( .send_to(&open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, ¶ms), peer) .unwrap(); let (packet, peer) = recv_request(&socket); + let flags = packet[3]; + if flags & FLAG_CLOSE != 0 { + return; + } for gate in release_gates { gate.recv_timeout(Duration::from_secs(2)) .expect("timed out waiting to release peer-close reply"); } + assert_eq!( + packet.len(), + echo_packet_len(false, ¶ms).unwrap(), + "expected a complete echo probe before peer-close reply" + ); let seq = u32::from_le_bytes(packet[12..16].try_into().unwrap()); socket .send_to( diff --git a/crates/irtt-client/src/async_client.rs b/crates/irtt-client/src/async_client.rs index ecee6c3..f693b85 100644 --- a/crates/irtt-client/src/async_client.rs +++ b/crates/irtt-client/src/async_client.rs @@ -27,7 +27,7 @@ use crate::{ receive::try_recv_tokio_datagram, session::machine::{ recv_buffer_size, OpenDatagramDisposition, PreparedOpenAcceptance, PreparedOpenRequest, - PreparedProbe, SessionMachine, MAX_OPEN_PACKET_SIZE, + PreparedProbe, SessionMachine, TimeoutBatch, MAX_OPEN_PACKET_SIZE, }, socket::{connect_tokio_udp_socket, resolve_remote_tokio, validate_open_timeouts}, socket_options::{apply_dscp_to_tokio_socket, clear_dscp_on_tokio_socket}, @@ -300,6 +300,14 @@ impl AsyncClient { self.machine.poll_timeouts_at(now) } + pub(crate) fn poll_timeouts_bounded_at( + &mut self, + now: Instant, + limit: usize, + ) -> Result { + self.machine.poll_timeouts_bounded_at(now, limit) + } + /// Send the retained close packet and commit local close exactly once. /// /// DSCP is cleared only after write readiness and immediately before the @@ -765,6 +773,19 @@ impl AsyncClient { self.socket.poll_recv_ready(cx) } + #[cfg(test)] + pub(crate) fn remove_pending_for_test( + &mut self, + wire_seq: u32, + ) -> Option { + self.machine.remove_pending_for_test(wire_seq) + } + + #[cfg(test)] + pub(crate) fn replace_pending_for_test(&mut self, probe: crate::probe::PendingProbe) { + self.machine.replace_pending_for_test(probe); + } + pub(crate) fn skip_missed_probe_slots_at(&mut self, now: Instant) -> Result<(), ClientError> { let Some(schedule) = self.schedule.as_mut() else { return Ok(()); diff --git a/crates/irtt-client/src/async_client/tests.rs b/crates/irtt-client/src/async_client/tests.rs index 6447c5a..779d84f 100644 --- a/crates/irtt-client/src/async_client/tests.rs +++ b/crates/irtt-client/src/async_client/tests.rs @@ -16,7 +16,10 @@ use irtt_proto::{ use tokio::runtime::{Builder, Runtime}; use super::*; -use crate::{socket_options::tokio_socket_traffic_class, Client, RunMode, SocketConfig}; +use crate::{ + probe::PendingProbe, socket_options::tokio_socket_traffic_class, Client, ClientTimestamp, + RunMode, SocketConfig, +}; const TOKEN: u64 = 0x1234_5678_90ab_cdef; @@ -708,6 +711,35 @@ fn probe_false_readiness_resamples_all_timestamps() { assert_eq!(server.finish().len(), 2); } +#[test] +fn public_timeout_polling_remains_exhaustive() { + let server = start_open_server(None, 0); + runtime().block_on(async { + let mut client = opened_client(server.addr, 0).await; + let now = Instant::now(); + for seq in 0..3 { + client.replace_pending_for_test(PendingProbe { + wire_seq: seq, + sent_at: ClientTimestamp { + mono: now - Duration::from_secs(2), + wall: SystemTime::UNIX_EPOCH, + }, + timeout_at: now - Duration::from_secs(1), + }); + } + + assert!(matches!( + client.poll_timeouts_at(now).unwrap().as_slice(), + [ + ClientEvent::EchoLoss { seq: 0, .. }, + ClientEvent::EchoLoss { seq: 1, .. }, + ClientEvent::EchoLoss { seq: 2, .. }, + ] + )); + }); + assert_eq!(server.finish().len(), 1); +} + #[test] fn successful_short_probe_commits_before_length_error() { let server = start_open_server(None, 0); diff --git a/crates/irtt-client/src/managed/task.rs b/crates/irtt-client/src/managed/task.rs index 88a2b95..f5571a0 100644 --- a/crates/irtt-client/src/managed/task.rs +++ b/crates/irtt-client/src/managed/task.rs @@ -37,6 +37,7 @@ use super::{ const TARGET_WORK_BUDGET: usize = 128; const COMMAND_WORK_BUDGET: usize = 32; +const TIMEOUT_WORK_BUDGET: usize = 128; const POST_DEADLINE_RECEIVE_BUDGET: usize = TARGET_WORK_BUDGET; // Tokio 1.53 broadcast asserts this exact bound before rounding capacity to a // power of two; Tokio does not expose it as a public constant. @@ -415,6 +416,12 @@ struct TargetRuntime { state: TargetState, } +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +struct TimeoutBacklogEntry { + index: usize, + generation: u64, +} + struct PlannedTarget { target: ManagedTargetConfig, client_config: ClientConfig, @@ -573,6 +580,8 @@ pub struct ManagedClientTask { wake: Option, timer: Option>>, cursor: usize, + timeout_cursor: usize, + timeout_backlog: VecDeque, scan_remaining: usize, send_cursor: usize, burst_remaining: usize, @@ -589,6 +598,10 @@ pub struct ManagedClientTask { stagger_observations: Option, #[cfg(test)] drain_test_hook: DrainTestHook, + #[cfg(test)] + timeout_inspections: usize, + #[cfg(test)] + deadline_inspections: usize, } #[cfg(test)] @@ -1046,9 +1059,11 @@ impl ManagedClientTask { let len = self.targets.len(); if len == 0 { self.cursor = 0; + self.timeout_cursor = 0; self.send_cursor = 0; } else { self.cursor %= len; + self.timeout_cursor %= len; self.send_cursor %= len; } self.scan_remaining = self.scan_remaining.min(len); @@ -1205,19 +1220,8 @@ impl ManagedClientTask { .next_probe_timeout_deadline() .is_some_and(|deadline| deadline <= now) { - match client.poll_timeouts_at(now) { - Ok(events) => self.publish_client_events(index, events), - Err(error) => { - return self.begin_open_session_failure( - index, - client, - ManagedTargetFailurePhase::Timing, - error, - now, - OpenSessionFailureCleanup::Close, - ); - } - } + self.targets[index].state = TargetState::Active { client }; + return true; } let received = match client.poll_recv(cx) { Poll::Pending => false, @@ -1277,35 +1281,26 @@ impl ManagedClientTask { return false; } + if client + .next_probe_timeout_deadline() + .is_some_and(|timeout| timeout <= now) + { + self.targets[index].state = TargetState::Draining { + client, + drain_started_at, + deadline, + primary_end, + cleanup_failure, + post_deadline_receives_remaining, + }; + return true; + } + #[cfg(test)] let injected_receive_failure = self.take_drain_receive_failure(); #[cfg(not(test))] let injected_receive_failure = None; let mut retained_state_changed = false; - if client - .next_probe_timeout_deadline() - .is_some_and(|timeout| timeout <= now) - { - match client.poll_timeouts_at(now) { - Ok(events) => { - self.publish_client_events(index, events); - retained_state_changed = true; - } - Err(error) => { - self.targets[index].counters.packets_sent = client.packets_sent(); - cleanup_failure.get_or_insert_with(|| { - classify_client_error(ManagedTargetFailurePhase::Timing, &error) - }); - return self.begin_close( - index, - client, - primary_end, - cleanup_failure, - now, - ); - } - } - } let received = match client.poll_recv(cx) { Poll::Pending => false, Poll::Ready(Ok(events)) => { @@ -1595,6 +1590,14 @@ impl ManagedClientTask { self.targets[index].state = state; return SendResult::NotAttempted; }; + if client + .next_probe_timeout_deadline() + .is_some_and(|deadline| deadline <= now) + { + self.targets[index].send_waiting = false; + self.targets[index].state = TargetState::Active { client }; + return SendResult::NotAttempted; + } if client .next_send_deadline() .is_none_or(|deadline| deadline > now) @@ -1728,13 +1731,176 @@ impl ManagedClientTask { immediate || self.scan_remaining > 0 } + fn poll_timeout_pass(&mut self, now: Instant) -> bool { + let target_len = self.targets.len(); + if target_len == 0 { + return false; + } + + let mut transitions_remaining = TIMEOUT_WORK_BUDGET; + for _ in 0..target_len.min(TIMEOUT_WORK_BUDGET) { + if transitions_remaining == 0 { + break; + } + let index = self.timeout_cursor; + self.timeout_cursor = (self.timeout_cursor + 1) % target_len; + #[cfg(test)] + { + self.timeout_inspections += 1; + } + if self.target_has_due_timeout(index, now) { + let generation = self.targets[index].instance.generation; + let step = self.poll_target_timeout(index, now); + transitions_remaining -= 1; + if step.more_due { + self.enqueue_timeout_backlog(index, generation); + } + } + } + + while transitions_remaining > 0 { + let Some(entry) = self.timeout_backlog.pop_front() else { + break; + }; + let valid = self + .targets + .get(entry.index) + .is_some_and(|target| target.instance.generation == entry.generation) + && self.target_has_due_timeout(entry.index, now); + if !valid { + continue; + } + + let step = self.poll_target_timeout(entry.index, now); + transitions_remaining -= 1; + if step.more_due { + self.enqueue_timeout_backlog(entry.index, entry.generation); + } + } + + !self.timeout_backlog.is_empty() + } + + fn enqueue_timeout_backlog(&mut self, index: usize, generation: u64) { + if self.timeout_backlog.len() == TIMEOUT_WORK_BUDGET + || self + .timeout_backlog + .iter() + .any(|entry| entry.index == index && entry.generation == generation) + { + return; + } + self.timeout_backlog + .push_back(TimeoutBacklogEntry { index, generation }); + } + + fn target_has_due_timeout(&self, index: usize, now: Instant) -> bool { + match &self.targets[index].state { + TargetState::Active { client } => { + self.state != DriverState::Stopping + && self.effective_retirement(index).is_none() + && client + .next_probe_timeout_deadline() + .is_some_and(|deadline| deadline <= now) + } + TargetState::Draining { + client, + deadline: _deadline, + .. + } => { + #[cfg(test)] + let defer_work = self.drain_test_hook.defer_work_until_deadline && now < *_deadline; + #[cfg(not(test))] + let defer_work = false; + !defer_work + && client + .next_probe_timeout_deadline() + .is_some_and(|timeout| timeout <= now) + } + _ => false, + } + } + + fn poll_target_timeout(&mut self, index: usize, now: Instant) -> TimeoutStep { + let state = mem::replace(&mut self.targets[index].state, TargetState::Terminal); + match state { + TargetState::Active { mut client } => match client.poll_timeouts_bounded_at(now, 1) { + Ok(batch) => { + self.publish_client_events(index, batch.events); + self.targets[index].state = TargetState::Active { client }; + TimeoutStep { + more_due: batch.more_due, + } + } + Err(error) => { + self.begin_open_session_failure( + index, + client, + ManagedTargetFailurePhase::Timing, + error, + now, + OpenSessionFailureCleanup::Close, + ); + TimeoutStep::NONE + } + }, + TargetState::Draining { + mut client, + drain_started_at, + deadline, + primary_end, + mut cleanup_failure, + post_deadline_receives_remaining, + } => match client.poll_timeouts_bounded_at(now, 1) { + Ok(batch) => { + self.publish_client_events(index, batch.events); + let deadline = match self.drain_deadline(&client, drain_started_at) { + Some(candidate) => deadline.min(candidate), + None => { + cleanup_failure.get_or_insert_with(duration_overflow_failure); + self.begin_close(index, client, primary_end, cleanup_failure, now); + return TimeoutStep::NONE; + } + }; + self.targets[index].state = TargetState::Draining { + client, + drain_started_at, + deadline, + primary_end, + cleanup_failure, + post_deadline_receives_remaining, + }; + TimeoutStep { + more_due: batch.more_due, + } + } + Err(error) => { + self.targets[index].counters.packets_sent = client.packets_sent(); + cleanup_failure.get_or_insert_with(|| { + classify_client_error(ManagedTargetFailurePhase::Timing, &error) + }); + self.begin_close(index, client, primary_end, cleanup_failure, now); + TimeoutStep::NONE + } + }, + state => { + self.targets[index].state = state; + TimeoutStep::NONE + } + } + } + fn all_targets_terminal(&self) -> bool { self.targets .iter() .all(|target| matches!(target.state, TargetState::Terminal)) } - fn next_deadline(&self) -> Option { + fn next_deadline(&mut self) -> Option { + #[cfg(test)] + { + self.deadline_inspections += self.targets.len(); + } let mut non_send_deadline = None; let mut send_deadline = None; for target in &self.targets { @@ -1888,6 +2054,15 @@ enum SendResult { Failed { accepted: bool }, } +#[derive(Clone, Copy)] +struct TimeoutStep { + more_due: bool, +} + +impl TimeoutStep { + const NONE: Self = Self { more_due: false }; +} + impl SendResult { fn accepted(self) -> bool { matches!( @@ -1922,6 +2097,7 @@ impl Future for ManagedClientTask { this.observe_stop(); } let now = Instant::now(); + immediate |= this.poll_timeout_pass(now); immediate |= this.poll_target_pass(cx, now); this.prune_undesired_terminal(); if this.state == DriverState::Running { @@ -1962,7 +2138,9 @@ impl Future for ManagedClientTask { } immediate |= this.register_stop_wake(cx); - immediate |= this.register_timer(cx); + if !immediate { + immediate |= this.register_timer(cx); + } if immediate { cx.waker().wake_by_ref(); } @@ -2129,6 +2307,8 @@ fn build_task( wake: Some(arm_wake(wake_receiver)), timer: None, cursor: 0, + timeout_cursor: 0, + timeout_backlog: VecDeque::with_capacity(TIMEOUT_WORK_BUDGET), scan_remaining: 0, send_cursor: 0, burst_remaining: 0, @@ -2145,6 +2325,10 @@ fn build_task( stagger_observations: None, #[cfg(test)] drain_test_hook: DrainTestHook::default(), + #[cfg(test)] + timeout_inspections: 0, + #[cfg(test)] + deadline_inspections: 0, }; let handle = ManagedClientHandle { stop, diff --git a/crates/irtt-client/src/managed/tests.rs b/crates/irtt-client/src/managed/tests.rs index f767145..dc91677 100644 --- a/crates/irtt-client/src/managed/tests.rs +++ b/crates/irtt-client/src/managed/tests.rs @@ -6,7 +6,7 @@ use std::{ sync::{Arc, Condvar, Mutex}, task::{Context, Poll, Waker}, thread::{self, JoinHandle}, - time::{Duration, Instant}, + time::{Duration, Instant, SystemTime}, }; use irtt_proto::{ @@ -16,7 +16,10 @@ use irtt_proto::{ use tokio::runtime::{Builder, Runtime}; use super::*; -use crate::{socket::resolution_call_counts, ClientAuthConfig, NegotiationPolicy, RunMode}; +use crate::{ + probe::PendingProbe, socket::resolution_call_counts, ClientAuthConfig, ClientTimestamp, + NegotiationPolicy, RunMode, +}; const TOKEN: u64 = 0x1234_5678_90ab_cdef; @@ -318,6 +321,94 @@ fn client_events( .collect() } +fn timeout_probe(seq: u32, sent_at: Instant, timeout_at: Instant) -> PendingProbe { + PendingProbe { + wire_seq: seq, + sent_at: ClientTimestamp { + mono: sent_at, + wall: SystemTime::UNIX_EPOCH, + }, + timeout_at, + } +} + +fn open_target_for_timeout_test(runtime: &Runtime, task: &mut ManagedClientTask, index: usize) { + let TargetState::Pending { client_config } = + mem::replace(&mut task.targets[index].state, TargetState::Terminal) + else { + panic!("timeout test target must start pending"); + }; + let client = runtime.block_on(async { + let mut client = AsyncClient::connect(client_config).await.unwrap(); + client.open().await.unwrap(); + client + }); + task.targets[index].desired = false; + task.install_target_state(index, TargetState::Active { client }); +} + +fn replace_pending_for_timeout_test( + task: &mut ManagedClientTask, + index: usize, + probe: PendingProbe, +) { + let client = match &mut task.targets[index].state { + TargetState::Active { client } | TargetState::Draining { client, .. } => client, + _ => panic!("timeout test target must be active or draining"), + }; + client.replace_pending_for_test(probe); +} + +fn remove_pending_for_timeout_test( + task: &mut ManagedClientTask, + index: usize, + wire_seq: u32, +) -> PendingProbe { + let client = match &mut task.targets[index].state { + TargetState::Active { client } | TargetState::Draining { client, .. } => client, + _ => panic!("timeout test target must be active or draining"), + }; + client + .remove_pending_for_test(wire_seq) + .expect("timeout test pending probe must exist") +} + +fn close_timeout_test_target(runtime: &Runtime, task: &mut ManagedClientTask, index: usize) { + let (TargetState::Active { mut client } | TargetState::Draining { mut client, .. }) = + mem::replace(&mut task.targets[index].state, TargetState::Terminal) + else { + panic!("timeout test target did not remain active or draining"); + }; + runtime.block_on(client.close()).unwrap(); +} + +fn poll_task_once(runtime: &Runtime, task: &mut Pin>) { + runtime.block_on(poll_fn(|cx| { + assert!(task.as_mut().poll(cx).is_pending()); + Poll::Ready(()) + })); +} + +fn timeout_losses_for( + observations: &Mutex)>>, + id: &str, +) -> usize { + observations + .lock() + .unwrap() + .iter() + .filter(|(event, _)| { + matches!( + event, + ManagedEvent::Client { + target, + event: ClientEvent::EchoLoss { .. }, + } if target.id.as_ref() == id + ) + }) + .count() +} + fn probes(records: &[PacketRecord]) -> Vec { records .iter() @@ -722,6 +813,7 @@ fn pruning_preserves_target_cursor_progress() { ) .unwrap(); task.cursor = 5; + task.timeout_cursor = 5; task.send_cursor = 5; task.scan_remaining = 3; task.burst_remaining = 3; @@ -729,6 +821,7 @@ fn pruning_preserves_target_cursor_progress() { task.targets.truncate(2); task.rebase_target_cursors(); assert_eq!(task.cursor, 1); + assert_eq!(task.timeout_cursor, 1); assert_eq!(task.send_cursor, 1); assert_eq!(task.scan_remaining, 2); assert_eq!(task.burst_remaining, 2); @@ -737,6 +830,7 @@ fn pruning_preserves_target_cursor_progress() { task.targets.clear(); task.rebase_target_cursors(); assert_eq!(task.cursor, 0); + assert_eq!(task.timeout_cursor, 0); assert_eq!(task.send_cursor, 0); assert_eq!(task.scan_remaining, 0); assert_eq!(task.burst_remaining, 0); @@ -1653,6 +1747,869 @@ fn stagger_gate_tracks_active_membership() { second.finish(); } +#[test] +fn timeout_discovery_inspects_at_most_one_budget_of_targets() { + let target_count = TIMEOUT_WORK_BUDGET + 1; + let mut managed = config(ManagedPacing::Burst); + managed.max_live_target_generations = target_count; + let targets = (0..target_count) + .map(|index| ManagedTargetConfig::new(format!("target-{index}"), "127.0.0.1:9")) + .collect(); + let (mut task, _) = ManagedClient::task(managed, targets).unwrap(); + + task.timeout_inspections = 0; + assert!(!task.poll_timeout_pass(Instant::now())); + + assert_eq!(task.timeout_inspections, TIMEOUT_WORK_BUDGET); + assert_eq!(task.timeout_cursor, TIMEOUT_WORK_BUDGET); +} + +#[test] +fn idle_timeout_discovery_inspects_one_target_once() { + let (mut task, _) = ManagedClient::task( + config(ManagedPacing::Burst), + vec![ManagedTargetConfig::new("idle", "127.0.0.1:9")], + ) + .unwrap(); + + task.timeout_inspections = 0; + assert!(!task.poll_timeout_pass(Instant::now())); + + assert_eq!(task.timeout_inspections, 1); + assert_eq!(task.timeout_cursor, 0); +} + +#[test] +fn idle_timeout_discovery_sweeps_small_target_set_once() { + let target_count = 8; + let mut managed = config(ManagedPacing::Burst); + managed.max_live_target_generations = target_count; + let targets = (0..target_count) + .map(|index| ManagedTargetConfig::new(format!("idle-{index}"), "127.0.0.1:9")) + .collect(); + let (mut task, _) = ManagedClient::task(managed, targets).unwrap(); + + task.timeout_inspections = 0; + assert!(!task.poll_timeout_pass(Instant::now())); + + assert_eq!(task.timeout_inspections, target_count); + assert_eq!(task.timeout_cursor, 0); +} + +#[test] +fn sparse_timeout_backlog_progresses_within_one_poll() { + let server = start_server(ServerBehavior::Echo, None); + let target_count = TIMEOUT_WORK_BUDGET + 1; + let late_index = target_count - 1; + let mut managed = config(ManagedPacing::Burst); + managed.completion = ManagedCompletionPolicy::ExplicitStop; + managed.client.duration = None; + managed.max_live_target_generations = target_count; + let targets = (0..target_count) + .map(|index| { + if index == late_index { + target("late", server.addr) + } else { + ManagedTargetConfig::new(format!("target-{index}"), "127.0.0.1:9") + } + }) + .collect(); + let (mut task, _) = ManagedClient::task(managed, targets).unwrap(); + for target in &mut task.targets[..late_index] { + target.state = TargetState::Terminal; + } + let observations = Arc::new(Mutex::new(Vec::new())); + task.event_observations = Some(Arc::clone(&observations)); + let runtime = runtime(); + open_target_for_timeout_test(&runtime, &mut task, late_index); + let now = Instant::now(); + for seq in 0..u32::try_from(TIMEOUT_WORK_BUDGET + 64).unwrap() { + replace_pending_for_timeout_test( + &mut task, + late_index, + timeout_probe( + seq, + now - Duration::from_secs(2), + now - Duration::from_secs(1), + ), + ); + } + + assert!(!task.poll_timeout_pass(now)); + assert!(task.poll_timeout_pass(now)); + + assert_eq!( + timeout_losses_for(&observations, "late"), + TIMEOUT_WORK_BUDGET + ); + assert_eq!(task.timeout_backlog.len(), 1); + close_timeout_test_target(&runtime, &mut task, late_index); + drop(task); + server.finish(); +} + +#[test] +fn sparse_timeout_backlog_persists_across_polls() { + let server = start_server(ServerBehavior::Echo, None); + let target_count = (2 * TIMEOUT_WORK_BUDGET) + 1; + let mut managed = config(ManagedPacing::Burst); + managed.completion = ManagedCompletionPolicy::ExplicitStop; + managed.client.duration = None; + managed.max_live_target_generations = target_count; + let targets = std::iter::once(target("sparse", server.addr)) + .chain( + (1..target_count) + .map(|index| ManagedTargetConfig::new(format!("target-{index}"), "127.0.0.1:9")), + ) + .collect(); + let (mut task, _) = ManagedClient::task(managed, targets).unwrap(); + for target in &mut task.targets[1..] { + target.state = TargetState::Terminal; + } + let observations = Arc::new(Mutex::new(Vec::new())); + task.event_observations = Some(Arc::clone(&observations)); + let runtime = runtime(); + open_target_for_timeout_test(&runtime, &mut task, 0); + let now = Instant::now(); + for seq in 0..u32::try_from((2 * TIMEOUT_WORK_BUDGET) + 64).unwrap() { + replace_pending_for_timeout_test( + &mut task, + 0, + timeout_probe( + seq, + now - Duration::from_secs(2), + now - Duration::from_secs(1), + ), + ); + } + + assert!(task.poll_timeout_pass(now)); + assert_eq!( + timeout_losses_for(&observations, "sparse"), + TIMEOUT_WORK_BUDGET + ); + assert_eq!(task.timeout_backlog.len(), 1); + assert_eq!(task.timeout_cursor, TIMEOUT_WORK_BUDGET); + + assert!(task.poll_timeout_pass(now)); + assert_eq!( + timeout_losses_for(&observations, "sparse"), + 2 * TIMEOUT_WORK_BUDGET + ); + assert_eq!(task.timeout_backlog.len(), 1); + assert_eq!(task.timeout_cursor, 2 * TIMEOUT_WORK_BUDGET); + close_timeout_test_target(&runtime, &mut task, 0); + drop(task); + server.finish(); +} + +#[test] +fn timeout_backlog_round_robins_known_sparse_targets() { + let first = start_server(ServerBehavior::Echo, None); + let second = start_server(ServerBehavior::Echo, None); + let target_count = TIMEOUT_WORK_BUDGET + 2; + let mut managed = config(ManagedPacing::Burst); + managed.completion = ManagedCompletionPolicy::ExplicitStop; + managed.client.duration = None; + managed.max_live_target_generations = target_count; + let targets = vec![target("first", first.addr), target("second", second.addr)] + .into_iter() + .chain( + (2..target_count) + .map(|index| ManagedTargetConfig::new(format!("target-{index}"), "127.0.0.1:9")), + ) + .collect(); + let (mut task, _) = ManagedClient::task(managed, targets).unwrap(); + for target in &mut task.targets[2..] { + target.state = TargetState::Terminal; + } + let observations = Arc::new(Mutex::new(Vec::new())); + task.event_observations = Some(Arc::clone(&observations)); + let runtime = runtime(); + open_target_for_timeout_test(&runtime, &mut task, 0); + open_target_for_timeout_test(&runtime, &mut task, 1); + let now = Instant::now(); + for index in 0..2 { + for seq in 0..u32::try_from(TIMEOUT_WORK_BUDGET + 1).unwrap() { + replace_pending_for_timeout_test( + &mut task, + index, + timeout_probe( + seq, + now - Duration::from_secs(2), + now - Duration::from_secs(1), + ), + ); + } + task.timeout_backlog.push_back(TimeoutBacklogEntry { + index, + generation: task.targets[index].instance.generation, + }); + } + task.timeout_cursor = 2; + + assert!(task.poll_timeout_pass(now)); + + assert_eq!( + timeout_losses_for(&observations, "first"), + TIMEOUT_WORK_BUDGET / 2 + ); + assert_eq!( + timeout_losses_for(&observations, "second"), + TIMEOUT_WORK_BUDGET / 2 + ); + assert_eq!(task.timeout_backlog.len(), 2); + close_timeout_test_target(&runtime, &mut task, 0); + close_timeout_test_target(&runtime, &mut task, 1); + drop(task); + first.finish(); + second.finish(); +} + +#[test] +fn timeout_backlog_does_not_starve_discovery() { + let first = start_server(ServerBehavior::Echo, None); + let second = start_server(ServerBehavior::Echo, None); + let target_count = (2 * TIMEOUT_WORK_BUDGET) + 1; + let late_index = target_count - 1; + let mut managed = config(ManagedPacing::Burst); + managed.completion = ManagedCompletionPolicy::ExplicitStop; + managed.client.duration = None; + managed.max_live_target_generations = target_count; + let targets = (0..target_count) + .map(|index| match index { + 0 => target("backlog", first.addr), + index if index == late_index => target("late", second.addr), + _ => ManagedTargetConfig::new(format!("target-{index}"), "127.0.0.1:9"), + }) + .collect(); + let (mut task, _) = ManagedClient::task(managed, targets).unwrap(); + for target in &mut task.targets[1..late_index] { + target.state = TargetState::Terminal; + } + let observations = Arc::new(Mutex::new(Vec::new())); + task.event_observations = Some(Arc::clone(&observations)); + let runtime = runtime(); + open_target_for_timeout_test(&runtime, &mut task, 0); + open_target_for_timeout_test(&runtime, &mut task, late_index); + let now = Instant::now(); + for seq in 0..u32::try_from(3 * TIMEOUT_WORK_BUDGET).unwrap() { + replace_pending_for_timeout_test( + &mut task, + 0, + timeout_probe( + seq, + now - Duration::from_secs(2), + now - Duration::from_secs(1), + ), + ); + } + replace_pending_for_timeout_test( + &mut task, + late_index, + timeout_probe( + 0, + now - Duration::from_secs(2), + now - Duration::from_secs(1), + ), + ); + task.timeout_backlog.push_back(TimeoutBacklogEntry { + index: 0, + generation: task.targets[0].instance.generation, + }); + task.timeout_cursor = 1; + + assert!(task.poll_timeout_pass(now)); + assert_eq!(timeout_losses_for(&observations, "late"), 0); + assert_eq!(task.timeout_cursor, TIMEOUT_WORK_BUDGET + 1); + + assert!(task.poll_timeout_pass(now)); + assert_eq!(timeout_losses_for(&observations, "late"), 1); + assert_eq!(task.timeout_cursor, 0); + assert_eq!(task.timeout_backlog.len(), 1); + close_timeout_test_target(&runtime, &mut task, 0); + close_timeout_test_target(&runtime, &mut task, late_index); + drop(task); + first.finish(); + second.finish(); +} + +#[test] +fn stale_timeout_backlog_entry_is_dropped_after_pruning() { + let server = start_server(ServerBehavior::Echo, None); + let target_count = TIMEOUT_WORK_BUDGET + 2; + let mut managed = config(ManagedPacing::Burst); + managed.completion = ManagedCompletionPolicy::ExplicitStop; + managed.client.duration = None; + managed.max_live_target_generations = target_count; + let targets = vec![ + ManagedTargetConfig::new("retired", "127.0.0.1:9"), + target("survivor", server.addr), + ] + .into_iter() + .chain( + (2..target_count) + .map(|index| ManagedTargetConfig::new(format!("target-{index}"), "127.0.0.1:9")), + ) + .collect(); + let (mut task, _) = ManagedClient::task(managed, targets).unwrap(); + let observations = Arc::new(Mutex::new(Vec::new())); + task.event_observations = Some(Arc::clone(&observations)); + let runtime = runtime(); + open_target_for_timeout_test(&runtime, &mut task, 1); + let now = Instant::now(); + replace_pending_for_timeout_test( + &mut task, + 1, + timeout_probe( + 0, + now - Duration::from_secs(2), + now - Duration::from_secs(1), + ), + ); + task.timeout_backlog.push_back(TimeoutBacklogEntry { + index: 0, + generation: task.targets[0].instance.generation, + }); + task.targets[0].desired = false; + task.targets[0].state = TargetState::Terminal; + task.prune_undesired_terminal(); + task.timeout_cursor = 1; + + assert!(!task.poll_timeout_pass(now)); + assert!(task.timeout_backlog.is_empty()); + assert_eq!(timeout_losses_for(&observations, "survivor"), 0); + + task.timeout_cursor = 0; + assert!(!task.poll_timeout_pass(now)); + assert_eq!(timeout_losses_for(&observations, "survivor"), 1); + close_timeout_test_target(&runtime, &mut task, 0); + drop(task); + server.finish(); +} + +#[test] +fn immediate_timeout_backlog_skips_global_deadline_scan() { + let server = start_server(ServerBehavior::Echo, None); + let mut managed = config(ManagedPacing::Burst); + managed.completion = ManagedCompletionPolicy::ExplicitStop; + managed.client.duration = None; + let (mut task, _) = ManagedClient::task(managed, vec![target("one", server.addr)]).unwrap(); + let runtime = runtime(); + open_target_for_timeout_test(&runtime, &mut task, 0); + let now = Instant::now(); + for seq in 0..u32::try_from(TIMEOUT_WORK_BUDGET + 1).unwrap() { + replace_pending_for_timeout_test( + &mut task, + 0, + timeout_probe( + seq, + now - Duration::from_secs(2), + now - Duration::from_secs(1), + ), + ); + } + task.timeout_backlog.push_back(TimeoutBacklogEntry { + index: 0, + generation: task.targets[0].instance.generation, + }); + task.deadline_inspections = 0; + + let mut task = Box::pin(task); + poll_task_once(&runtime, &mut task); + assert_eq!(task.deadline_inspections, 0); + assert_eq!(task.timeout_backlog.len(), 1); + + assert!(!task.poll_timeout_pass(now)); + assert!(task.timeout_backlog.is_empty()); + task.deadline_inspections = 0; + poll_task_once(&runtime, &mut task); + assert_eq!(task.deadline_inspections, 1); + close_timeout_test_target(&runtime, &mut task, 0); + drop(task); + server.finish(); +} + +#[test] +fn timeout_discovery_reaches_later_target_after_budget_slice() { + let server = start_server(ServerBehavior::Echo, None); + let target_count = TIMEOUT_WORK_BUDGET + 1; + let late_index = target_count - 1; + let mut managed = config(ManagedPacing::Burst); + managed.completion = ManagedCompletionPolicy::ExplicitStop; + managed.client.duration = None; + managed.max_live_target_generations = target_count; + let targets = (0..target_count) + .map(|index| { + if index == late_index { + target("late", server.addr) + } else { + ManagedTargetConfig::new(format!("target-{index}"), "127.0.0.1:9") + } + }) + .collect(); + let (mut task, _) = ManagedClient::task(managed, targets).unwrap(); + for target in &mut task.targets[..late_index] { + target.state = TargetState::Terminal; + } + let observations = Arc::new(Mutex::new(Vec::new())); + task.event_observations = Some(Arc::clone(&observations)); + let runtime = runtime(); + open_target_for_timeout_test(&runtime, &mut task, late_index); + let now = Instant::now(); + replace_pending_for_timeout_test( + &mut task, + late_index, + timeout_probe(0, now - Duration::from_secs(1), now), + ); + + task.timeout_inspections = 0; + assert!(!task.poll_timeout_pass(now)); + assert_eq!(task.timeout_inspections, TIMEOUT_WORK_BUDGET); + assert_eq!(task.timeout_cursor, late_index); + assert_eq!(timeout_losses_for(&observations, "late"), 0); + + task.timeout_inspections = 0; + assert!(!task.poll_timeout_pass(now)); + assert_eq!(task.timeout_inspections, TIMEOUT_WORK_BUDGET); + assert_eq!(timeout_losses_for(&observations, "late"), 1); + + let TargetState::Active { mut client } = + mem::replace(&mut task.targets[late_index].state, TargetState::Terminal) + else { + panic!("later timeout target did not remain active"); + }; + runtime.block_on(client.close()).unwrap(); + drop(task); + server.finish(); +} + +#[test] +fn timeout_work_is_bounded_per_task_poll_and_eventually_drains() { + let server = start_server(ServerBehavior::Echo, None); + let mut managed = config(ManagedPacing::Staggered); + managed.completion = ManagedCompletionPolicy::ExplicitStop; + managed.client.duration = None; + let (mut task, handle) = + ManagedClient::task(managed, vec![target("one", server.addr)]).unwrap(); + let observations = Arc::new(Mutex::new(Vec::new())); + task.event_observations = Some(Arc::clone(&observations)); + let runtime = runtime(); + open_target_for_timeout_test(&runtime, &mut task, 0); + let now = Instant::now(); + for seq in 0..u32::try_from(TIMEOUT_WORK_BUDGET + 1).unwrap() { + replace_pending_for_timeout_test( + &mut task, + 0, + timeout_probe( + seq, + now - Duration::from_secs(2), + now - Duration::from_secs(1), + ), + ); + } + + let mut task = Box::pin(task); + poll_task_once(&runtime, &mut task); + assert_eq!( + timeout_losses_for(&observations, "one"), + TIMEOUT_WORK_BUDGET + ); + assert!(matches!( + &task.targets[0].state, + TargetState::Active { client } + if client.next_probe_timeout_deadline().is_some() + )); + + drive_task_until(&runtime, &mut task, |task| { + matches!( + &task.targets[0].state, + TargetState::Active { client } + if client.next_probe_timeout_deadline().is_none() + ) + }); + assert_eq!( + timeout_losses_for(&observations, "one"), + TIMEOUT_WORK_BUDGET + 1 + ); + + drop(handle.stop()); + runtime.block_on(task); + server.finish(); +} + +#[test] +fn timeout_work_round_robins_simultaneous_target_backlogs() { + let first = start_server(ServerBehavior::Echo, None); + let second = start_server(ServerBehavior::Echo, None); + let mut managed = config(ManagedPacing::Staggered); + managed.completion = ManagedCompletionPolicy::ExplicitStop; + managed.client.duration = None; + let (mut task, handle) = ManagedClient::task( + managed, + vec![target("first", first.addr), target("second", second.addr)], + ) + .unwrap(); + let observations = Arc::new(Mutex::new(Vec::new())); + task.event_observations = Some(Arc::clone(&observations)); + let runtime = runtime(); + open_target_for_timeout_test(&runtime, &mut task, 0); + open_target_for_timeout_test(&runtime, &mut task, 1); + let now = Instant::now(); + let per_target = TIMEOUT_WORK_BUDGET + 16; + for index in 0..2 { + for seq in 0..u32::try_from(per_target).unwrap() { + replace_pending_for_timeout_test( + &mut task, + index, + timeout_probe( + seq, + now - Duration::from_secs(2), + now - Duration::from_secs(1), + ), + ); + } + } + + let mut task = Box::pin(task); + poll_task_once(&runtime, &mut task); + assert_eq!( + timeout_losses_for(&observations, "first"), + TIMEOUT_WORK_BUDGET / 2 + ); + assert_eq!( + timeout_losses_for(&observations, "second"), + TIMEOUT_WORK_BUDGET / 2 + ); + assert!(matches!( + &task.targets[0].state, + TargetState::Active { client } + if client.next_probe_timeout_deadline().is_some() + )); + assert!(matches!( + &task.targets[1].state, + TargetState::Active { client } + if client.next_probe_timeout_deadline().is_some() + )); + + drive_task_until(&runtime, &mut task, |task| { + task.targets.iter().all(|target| { + matches!( + &target.state, + TargetState::Active { client } + if client.next_probe_timeout_deadline().is_none() + ) + }) + }); + assert_eq!(timeout_losses_for(&observations, "first"), per_target); + assert_eq!(timeout_losses_for(&observations, "second"), per_target); + + drop(handle.stop()); + runtime.block_on(task); + first.finish(); + second.finish(); +} + +#[test] +fn budget_deferred_timeout_blocks_send_until_timeout_processing() { + let reply_gate = Arc::new(PacketGate::default()); + let server = start_server_with_gates( + ServerBehavior::Echo, + None, + None, + None, + Some(Arc::clone(&reply_gate)), + ); + let target_count = TIMEOUT_WORK_BUDGET + 1; + let late_index = target_count - 1; + let mut managed = config(ManagedPacing::Burst); + managed.completion = ManagedCompletionPolicy::ExplicitStop; + managed.client.duration = None; + managed.client.interval = Duration::from_nanos(1); + managed.client.probe_timeout = Duration::from_secs(1); + managed.client.max_pending_probes = 1; + managed.max_live_target_generations = target_count; + let targets = (0..target_count) + .map(|index| { + if index == late_index { + target("late", server.addr) + } else { + ManagedTargetConfig::new(format!("target-{index}"), "127.0.0.1:9") + } + }) + .collect(); + let (mut task, _) = ManagedClient::task(managed, targets).unwrap(); + for target in &mut task.targets[..late_index] { + target.state = TargetState::Terminal; + } + let observations = Arc::new(Mutex::new(Vec::new())); + task.event_observations = Some(Arc::clone(&observations)); + let runtime = runtime(); + open_target_for_timeout_test(&runtime, &mut task, late_index); + task.targets[late_index].desired = true; + let sent = match &mut task.targets[late_index].state { + TargetState::Active { client } => runtime.block_on(client.send_probe()).unwrap(), + _ => panic!("deferred-send target did not open"), + }; + assert!(matches!( + sent.as_slice(), + [ClientEvent::EchoSent { seq: 0, .. }] + )); + wait_flag(&server.probe_seen); + + let overdue = Instant::now(); + replace_pending_for_timeout_test( + &mut task, + late_index, + timeout_probe(0, overdue - Duration::from_secs(1), overdue), + ); + task.state = DriverState::Running; + task.lifecycle = ManagedLifecycle::Running; + task.timeout_cursor = 0; + task.send_cursor = late_index; + + let mut task = Box::pin(task); + poll_task_once(&runtime, &mut task); + assert_eq!(task.timeout_inspections, TIMEOUT_WORK_BUDGET); + assert_eq!(timeout_losses_for(&observations, "late"), 0); + assert!(matches!( + &task.targets[late_index].state, + TargetState::Active { client } + if client.packets_sent() == 1 + && client.next_probe_timeout_deadline() == Some(overdue) + )); + assert!(!task.targets[late_index].send_waiting); + + poll_task_once(&runtime, &mut task); + assert_eq!(timeout_losses_for(&observations, "late"), 1); + assert!(matches!( + &task.targets[late_index].state, + TargetState::Active { client } + if client.next_probe_timeout_deadline().is_none() + )); + + poll_task_once(&runtime, &mut task); + assert!(matches!( + &task.targets[late_index].state, + TargetState::Active { client } if client.packets_sent() == 2 + )); + + reply_gate.release(); + let TargetState::Active { mut client } = + mem::replace(&mut task.targets[late_index].state, TargetState::Terminal) + else { + panic!("deferred-send target did not remain active"); + }; + runtime.block_on(client.close()).unwrap(); + drop(task); + server.finish(); +} + +#[test] +fn partial_timeout_batch_blocks_active_receive_until_loss_commits() { + let reply_gate = Arc::new(PacketGate::default()); + let server = start_server_with_gates( + ServerBehavior::Echo, + None, + None, + None, + Some(Arc::clone(&reply_gate)), + ); + let mut managed = config(ManagedPacing::Staggered); + managed.completion = ManagedCompletionPolicy::ExplicitStop; + managed.client.duration = None; + let (mut task, handle) = + ManagedClient::task(managed, vec![target("one", server.addr)]).unwrap(); + let observations = Arc::new(Mutex::new(Vec::new())); + task.event_observations = Some(Arc::clone(&observations)); + let runtime = runtime(); + open_target_for_timeout_test(&runtime, &mut task, 0); + let sent = match &mut task.targets[0].state { + TargetState::Active { client } => runtime.block_on(client.send_probe()).unwrap(), + _ => panic!("timeout test target did not open"), + }; + assert!(matches!( + sent.as_slice(), + [ClientEvent::EchoSent { seq: 0, .. }] + )); + wait_flag(&server.probe_seen); + + let now = Instant::now(); + let _ = remove_pending_for_timeout_test(&mut task, 0, 0); + for seq in 1..=u32::try_from(TIMEOUT_WORK_BUDGET).unwrap() { + replace_pending_for_timeout_test( + &mut task, + 0, + timeout_probe( + seq, + now - Duration::from_secs(2), + now - Duration::from_secs(1), + ), + ); + } + replace_pending_for_timeout_test( + &mut task, + 0, + timeout_probe(0, now - Duration::from_secs(1), now), + ); + reply_gate.release(); + wait_flag(&server.reply_sent); + + let mut task = Box::pin(task); + poll_task_once(&runtime, &mut task); + assert_eq!( + timeout_losses_for(&observations, "one"), + TIMEOUT_WORK_BUDGET + ); + assert!(!client_events(&observations).iter().any(|event| { + matches!( + event, + ClientEvent::EchoLoss { seq: 0, .. } + | ClientEvent::LateReply { seq: 0, .. } + | ClientEvent::EchoReply { seq: 0, .. } + ) + })); + + drive_task_until(&runtime, &mut task, |_| { + client_events(&observations) + .iter() + .any(|event| matches!(event, ClientEvent::LateReply { seq: 0, .. })) + }); + let classified = client_events(&observations) + .into_iter() + .filter(|event| { + matches!( + event, + ClientEvent::EchoLoss { seq: 0, .. } + | ClientEvent::LateReply { seq: 0, .. } + | ClientEvent::EchoReply { seq: 0, .. } + ) + }) + .collect::>(); + assert!(matches!( + classified.as_slice(), + [ + ClientEvent::EchoLoss { seq: 0, .. }, + ClientEvent::LateReply { seq: 0, .. } + ] + )); + + drop(handle.stop()); + runtime.block_on(task); + server.finish(); +} + +#[test] +fn partial_timeout_batch_blocks_draining_receive_until_loss_commits() { + let reply_gate = Arc::new(PacketGate::default()); + let server = start_server_with_gates( + ServerBehavior::Echo, + None, + None, + None, + Some(Arc::clone(&reply_gate)), + ); + let mut managed = config(ManagedPacing::Staggered); + managed.completion = ManagedCompletionPolicy::ExplicitStop; + managed.client.duration = None; + let (mut task, handle) = + ManagedClient::task(managed, vec![target("one", server.addr)]).unwrap(); + let observations = Arc::new(Mutex::new(Vec::new())); + task.event_observations = Some(Arc::clone(&observations)); + let runtime = runtime(); + open_target_for_timeout_test(&runtime, &mut task, 0); + let sent = match &mut task.targets[0].state { + TargetState::Active { client } => runtime.block_on(client.send_probe()).unwrap(), + _ => panic!("timeout test target did not open"), + }; + assert!(matches!( + sent.as_slice(), + [ClientEvent::EchoSent { seq: 0, .. }] + )); + wait_flag(&server.probe_seen); + + let now = Instant::now(); + let _ = remove_pending_for_timeout_test(&mut task, 0, 0); + for seq in 1..=u32::try_from(TIMEOUT_WORK_BUDGET).unwrap() { + replace_pending_for_timeout_test( + &mut task, + 0, + timeout_probe( + seq, + now - Duration::from_secs(2), + now - Duration::from_secs(1), + ), + ); + } + replace_pending_for_timeout_test( + &mut task, + 0, + timeout_probe(0, now - Duration::from_secs(1), now), + ); + let TargetState::Active { client } = + mem::replace(&mut task.targets[0].state, TargetState::Terminal) + else { + panic!("timeout test target did not remain active"); + }; + assert!(task.begin_drain(0, client, ManagedTargetEndReason::Stopped, now)); + reply_gate.release(); + wait_flag(&server.reply_sent); + + let mut task = Box::pin(task); + poll_task_once(&runtime, &mut task); + assert_eq!( + timeout_losses_for(&observations, "one"), + TIMEOUT_WORK_BUDGET + ); + assert!(matches!( + &task.targets[0].state, + TargetState::Draining { + primary_end: ManagedTargetEndReason::Stopped, + cleanup_failure: None, + .. + } + )); + assert!(!client_events(&observations).iter().any(|event| { + matches!( + event, + ClientEvent::EchoLoss { seq: 0, .. } + | ClientEvent::LateReply { seq: 0, .. } + | ClientEvent::EchoReply { seq: 0, .. } + ) + })); + + drive_task_until(&runtime, &mut task, |_| { + client_events(&observations) + .iter() + .any(|event| matches!(event, ClientEvent::LateReply { seq: 0, .. })) + }); + let classified = client_events(&observations) + .into_iter() + .filter(|event| { + matches!( + event, + ClientEvent::EchoLoss { seq: 0, .. } + | ClientEvent::LateReply { seq: 0, .. } + | ClientEvent::EchoReply { seq: 0, .. } + ) + }) + .collect::>(); + assert!(matches!( + classified.as_slice(), + [ + ClientEvent::EchoLoss { seq: 0, .. }, + ClientEvent::LateReply { seq: 0, .. } + ] + )); + + drop(handle.stop()); + let outcome = runtime.block_on(task); + let target = &outcome.recent_target_outcomes[0]; + assert!(matches!(target.end_reason, ManagedTargetEndReason::Stopped)); + assert!(target.cleanup_failure.is_none()); + server.finish(); +} + #[test] fn overdue_queued_reply_is_lost_then_late() { let reply_gate = Arc::new(PacketGate::default()); diff --git a/crates/irtt-client/src/probe.rs b/crates/irtt-client/src/probe.rs index 682bcfb..951fb1e 100644 --- a/crates/irtt-client/src/probe.rs +++ b/crates/irtt-client/src/probe.rs @@ -12,9 +12,24 @@ pub(crate) struct PendingProbe { pub timeout_at: Instant, } +#[derive(Debug)] +struct PendingEntry { + probe: PendingProbe, + previous: Option, + next: Option, +} + +#[derive(Debug)] +pub(crate) struct ExpiredBatch { + pub probes: Vec, + pub more_due: bool, +} + #[derive(Debug)] pub(crate) struct PendingMap { - map: HashMap, + map: HashMap, + first: Option, + last: Option, max_capacity: usize, } @@ -22,6 +37,8 @@ impl PendingMap { pub fn new(max_capacity: usize) -> Self { Self { map: HashMap::new(), + first: None, + last: None, max_capacity, } } @@ -42,43 +59,140 @@ impl PendingMap { } pub fn commit_insert(&mut self, probe: PendingProbe) { - let replaced = self.map.insert(probe.wire_seq, probe); + let wire_seq = probe.wire_seq; + let previous = self.last; + if let Some(previous) = previous { + let previous = self + .map + .get_mut(&previous) + .expect("pending list tail remains present"); + // probe_timeout is fixed for an open session and committed sends + // have nondecreasing monotonic timestamps, so deadlines append. + debug_assert!(previous.probe.timeout_at <= probe.timeout_at); + debug_assert!( + previous.next.is_none(), + "pending list tail has no successor" + ); + previous.next = Some(wire_seq); + } else { + debug_assert!(self.first.is_none(), "empty pending list has no head"); + self.first = Some(wire_seq); + } + + let replaced = self.map.insert( + wire_seq, + PendingEntry { + probe, + previous, + next: None, + }, + ); debug_assert!(replaced.is_none(), "preflight rejected pending collision"); + self.last = Some(wire_seq); + self.assert_links_consistent(); } pub fn remove(&mut self, wire_seq: u32) -> Option { - self.map.remove(&wire_seq) + let PendingEntry { + probe, + previous, + next, + } = self.map.remove(&wire_seq)?; + + if let Some(previous) = previous { + let previous = self + .map + .get_mut(&previous) + .expect("pending list predecessor remains present"); + debug_assert_eq!(previous.next, Some(wire_seq)); + previous.next = next; + } else { + debug_assert_eq!(self.first, Some(wire_seq)); + self.first = next; + } + if let Some(next) = next { + let next = self + .map + .get_mut(&next) + .expect("pending list successor remains present"); + debug_assert_eq!(next.previous, Some(wire_seq)); + next.previous = previous; + } else { + debug_assert_eq!(self.last, Some(wire_seq)); + self.last = previous; + } + + self.assert_links_consistent(); + Some(probe) } - pub fn drain_expired(&mut self, now: Instant) -> Vec { - let expired_keys: Vec = self - .map - .iter() - .filter(|(_, probe)| probe.timeout_at <= now) - .map(|(key, _)| *key) - .collect(); - let mut expired = Vec::with_capacity(expired_keys.len()); - for key in expired_keys { - if let Some(probe) = self.map.remove(&key) { - expired.push(probe); + pub(crate) fn drain_expired_bounded(&mut self, now: Instant, limit: usize) -> ExpiredBatch { + let mut probes = Vec::with_capacity(limit.min(self.map.len())); + while probes.len() < limit { + let Some(wire_seq) = self.first else { + break; + }; + let timeout_at = self + .map + .get(&wire_seq) + .expect("pending list head remains present") + .probe + .timeout_at; + if timeout_at > now { + break; } + + let probe = self + .remove(wire_seq) + .expect("pending list head remains present"); + probes.push(probe); } - expired.sort_by_key(|p| p.sent_at.mono); - expired + let more_due = self + .first + .and_then(|wire_seq| self.map.get(&wire_seq)) + .is_some_and(|entry| entry.probe.timeout_at <= now); + ExpiredBatch { probes, more_due } } pub fn len(&self) -> usize { self.map.len() } - #[cfg(feature = "tokio")] + #[cfg(any(feature = "tokio", test))] pub fn next_timeout_deadline(&self) -> Option { - self.map.values().map(|probe| probe.timeout_at).min() + self.first + .and_then(|wire_seq| self.map.get(&wire_seq)) + .map(|entry| entry.probe.timeout_at) } - #[cfg(feature = "tokio")] + #[cfg(any(feature = "tokio", test))] pub fn latest_timeout_deadline(&self) -> Option { - self.map.values().map(|probe| probe.timeout_at).max() + self.last + .and_then(|wire_seq| self.map.get(&wire_seq)) + .map(|entry| entry.probe.timeout_at) + } + + fn assert_links_consistent(&self) { + debug_assert_eq!(self.first.is_none(), self.map.is_empty()); + debug_assert_eq!(self.last.is_none(), self.map.is_empty()); + if let Some(first) = self.first { + debug_assert_eq!( + self.map + .get(&first) + .expect("pending list head remains present") + .previous, + None + ); + } + if let Some(last) = self.last { + debug_assert_eq!( + self.map + .get(&last) + .expect("pending list tail remains present") + .next, + None + ); + } } #[cfg(test)] @@ -90,6 +204,18 @@ impl PendingMap { pub fn capacity(&self) -> usize { self.map.capacity() } + + #[cfg(test)] + pub fn linkage(&self) -> (Option, Option) { + (self.first, self.last) + } + + #[cfg(test)] + pub fn entry_links(&self, wire_seq: u32) -> Option<(Option, Option)> { + self.map + .get(&wire_seq) + .map(|entry| (entry.previous, entry.next)) + } } fn pending_allocation_failed(source: TryReserveError) -> ClientError { @@ -244,6 +370,11 @@ mod tests { } } + fn insert(map: &mut PendingMap, probe: PendingProbe) { + map.preflight_insert(probe.wire_seq).unwrap(); + map.commit_insert(probe); + } + #[test] fn pending_map_rejects_over_capacity() { let mut map = PendingMap::new(2); @@ -303,6 +434,163 @@ mod tests { )); } + #[test] + fn pending_map_unlinks_head_middle_and_tail_and_updates_extrema() { + let mut map = PendingMap::new(4); + let now = Instant::now(); + insert(&mut map, pending(1, now + Duration::from_secs(1))); + insert(&mut map, pending(2, now + Duration::from_secs(2))); + insert(&mut map, pending(3, now + Duration::from_secs(3))); + + assert_eq!( + map.next_timeout_deadline(), + Some(now + Duration::from_secs(1)) + ); + assert_eq!( + map.latest_timeout_deadline(), + Some(now + Duration::from_secs(3)) + ); + assert_eq!(map.linkage(), (Some(1), Some(3))); + assert_eq!(map.entry_links(1), Some((None, Some(2)))); + assert_eq!(map.entry_links(2), Some((Some(1), Some(3)))); + assert_eq!(map.entry_links(3), Some((Some(2), None))); + + assert!(map.remove(2).is_some()); + assert_eq!( + map.next_timeout_deadline(), + Some(now + Duration::from_secs(1)) + ); + assert_eq!( + map.latest_timeout_deadline(), + Some(now + Duration::from_secs(3)) + ); + assert_eq!(map.linkage(), (Some(1), Some(3))); + assert_eq!(map.entry_links(1), Some((None, Some(3)))); + assert_eq!(map.entry_links(3), Some((Some(1), None))); + + assert!(map.remove(1).is_some()); + assert_eq!( + map.next_timeout_deadline(), + Some(now + Duration::from_secs(3)) + ); + assert_eq!( + map.latest_timeout_deadline(), + Some(now + Duration::from_secs(3)) + ); + assert_eq!(map.linkage(), (Some(3), Some(3))); + assert_eq!(map.entry_links(3), Some((None, None))); + + assert!(map.remove(3).is_some()); + assert_eq!(map.next_timeout_deadline(), None); + assert_eq!(map.latest_timeout_deadline(), None); + assert_eq!(map.linkage(), (None, None)); + } + + #[test] + fn pending_map_bounded_expiration_matches_exhaustive_order() { + let now = Instant::now(); + let probes = [ + pending(1, now + Duration::from_secs(1)), + pending(2, now + Duration::from_secs(2)), + pending(3, now + Duration::from_secs(3)), + pending(4, now + Duration::from_secs(4)), + ]; + let mut bounded = PendingMap::new(probes.len()); + let mut exhaustive = PendingMap::new(probes.len()); + for probe in probes.iter().cloned() { + insert(&mut bounded, probe.clone()); + insert(&mut exhaustive, probe); + } + + let empty = bounded.drain_expired_bounded(now + Duration::from_secs(4), 0); + assert!(empty.probes.is_empty()); + assert!(empty.more_due); + assert_eq!(bounded.linkage(), (Some(1), Some(4))); + + let mut actual = Vec::new(); + loop { + let batch = bounded.drain_expired_bounded(now + Duration::from_secs(4), 2); + actual.extend(batch.probes.into_iter().map(|probe| probe.wire_seq)); + if !batch.more_due { + break; + } + } + let expected = exhaustive + .drain_expired_bounded(now + Duration::from_secs(4), usize::MAX) + .probes + .into_iter() + .map(|probe| probe.wire_seq) + .collect::>(); + + assert_eq!(actual, expected); + assert_eq!(actual, vec![1, 2, 3, 4]); + assert_eq!(bounded.linkage(), (None, None)); + } + + #[test] + fn pending_map_sequence_reuse_leaves_no_stale_link() { + let mut map = PendingMap::new(1); + let now = Instant::now(); + insert(&mut map, pending(7, now)); + assert!(map.remove(7).is_some()); + insert(&mut map, pending(7, now + Duration::from_secs(1))); + + assert!(map.drain_expired_bounded(now, usize::MAX).probes.is_empty()); + assert_eq!(map.linkage(), (Some(7), Some(7))); + assert_eq!(map.entry_links(7), Some((None, None))); + assert_eq!( + map.drain_expired_bounded(now + Duration::from_secs(1), usize::MAX) + .probes + .into_iter() + .map(|probe| probe.wire_seq) + .collect::>(), + vec![7] + ); + assert_eq!(map.linkage(), (None, None)); + } + + #[test] + fn pending_map_commit_uses_only_preflighted_map_capacity() { + let mut map = PendingMap::new(2); + let now = Instant::now(); + map.preflight_insert(7).unwrap(); + let capacity = map.capacity(); + + map.commit_insert(pending(7, now)); + + assert_eq!(map.capacity(), capacity); + assert_eq!(map.len(), 1); + assert_eq!(map.linkage(), (Some(7), Some(7))); + } + + #[test] + fn pending_map_mixed_removals_preserve_links() { + let mut map = PendingMap::new(4); + let now = Instant::now(); + for seq in 1..=4 { + insert( + &mut map, + pending(seq, now + Duration::from_secs(u64::from(seq))), + ); + } + + assert!(map.remove(2).is_some()); + assert_eq!(map.entry_links(1), Some((None, Some(3)))); + assert_eq!(map.entry_links(3), Some((Some(1), Some(4)))); + let expired = map.drain_expired_bounded(now + Duration::from_secs(3), usize::MAX); + assert_eq!( + expired + .probes + .into_iter() + .map(|probe| probe.wire_seq) + .collect::>(), + vec![1, 3] + ); + assert_eq!(map.linkage(), (Some(4), Some(4))); + assert!(map.remove(4).is_some()); + assert_eq!(map.linkage(), (None, None)); + } + #[test] fn bounded_probe_tracking_evicts_oldest_entries() { let mut set = CompletedSet::new(3); diff --git a/crates/irtt-client/src/session/machine.rs b/crates/irtt-client/src/session/machine.rs index 7d3d8f9..7eade8b 100644 --- a/crates/irtt-client/src/session/machine.rs +++ b/crates/irtt-client/src/session/machine.rs @@ -157,6 +157,12 @@ pub(crate) struct ProbeCommit { next_packets_sent: u64, } +#[derive(Debug)] +pub(crate) struct TimeoutBatch { + pub events: Vec, + pub more_due: bool, +} + #[derive(Debug, Clone, Copy)] pub(crate) struct ProbeSent { pub(crate) seq: u32, @@ -408,11 +414,21 @@ impl SessionMachine { &mut self, now: Instant, ) -> Result, ClientError> { + let batch = self.poll_timeouts_bounded_at(now, usize::MAX)?; + debug_assert!(!batch.more_due); + Ok(batch.events) + } + + pub(crate) fn poll_timeouts_bounded_at( + &mut self, + now: Instant, + limit: usize, + ) -> Result { let session = self.open_session_mut()?; - let expired = session.pending.drain_expired(now); - let mut events = Vec::with_capacity(expired.len()); - for probe in expired { + let expired = session.pending.drain_expired_bounded(now, limit); + let mut events = Vec::with_capacity(expired.probes.len()); + for probe in expired.probes { events.push(ClientEvent::EchoLoss { seq: probe.wire_seq, sent_at: probe.sent_at, @@ -421,7 +437,10 @@ impl SessionMachine { session.timed_out.insert(probe); } - Ok(events) + Ok(TimeoutBatch { + events, + more_due: expired.more_due, + }) } pub(crate) fn prepare_close(&self) -> Result, ClientError> { @@ -1019,6 +1038,25 @@ impl SessionMachine { } } +#[cfg(all(test, feature = "tokio"))] +impl SessionMachine { + pub(crate) fn remove_pending_for_test(&mut self, wire_seq: u32) -> Option { + self.open_session_mut() + .expect("test pending probes require an open session") + .pending + .remove(wire_seq) + } + + pub(crate) fn replace_pending_for_test(&mut self, probe: PendingProbe) { + let session = self + .open_session_mut() + .expect("test pending probes require an open session"); + session.pending.remove(probe.wire_seq); + session.pending.preflight_insert(probe.wire_seq).unwrap(); + session.pending.commit_insert(probe); + } +} + #[cfg(test)] mod tests { use super::*; @@ -1803,6 +1841,34 @@ mod tests { assert_eq!(active(&machine).pending.len(), 0); } + #[test] + fn exhaustive_timeout_polling_returns_every_due_loss() { + let mut machine = open_machine(4, Duration::from_secs(1)); + let now = Instant::now(); + for seq in [2, 1, 0] { + let sent_at = timestamp(now - Duration::from_secs(u64::from(seq) + 1)); + let session = active_mut(&mut machine); + session.pending.preflight_insert(seq).unwrap(); + session.pending.commit_insert(PendingProbe { + wire_seq: seq, + sent_at, + timeout_at: sent_at.mono + Duration::from_secs(1), + }); + } + + let events = machine.poll_timeouts_at(now).unwrap(); + assert!(matches!( + events.as_slice(), + [ + ClientEvent::EchoLoss { seq: 2, .. }, + ClientEvent::EchoLoss { seq: 1, .. }, + ClientEvent::EchoLoss { seq: 0, .. }, + ] + )); + assert_eq!(active(&machine).pending.len(), 0); + assert_eq!(active(&machine).timed_out.len(), 3); + } + #[test] fn wrapping_sequence_from_max_to_zero_remains_valid() { let mut machine = open_machine(2, Duration::from_secs(1));