From 5506220f67af4385a9cdab50d5c720cff610d11e Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Sat, 15 Aug 2026 01:36:04 +0700 Subject: [PATCH] refactor(client): centralize managed target accounting Target-local accounting was spread across the managed driver: the event counter increments were inline in publish_client_events, the authoritative packets_sent was copied from the client at fifteen separate call sites, and ManagedTargetOutcome was constructed twice from the same six fields. TargetCounters::observe now owns the per-event accounting, TargetRuntime::sync_packets_sent/record_packets_sent own the authoritative sent count, and TargetRuntime::outcome owns outcome construction, replacing the free target_outcome function and the inline copy in finish_target. No counter is touched outside TargetRuntime any more. The replies_received semantics are unchanged and now documented on both the private counter and the public outcome field: on-time unique echo replies only, with late unique replies counted by late. observe matches ClientEvent exhaustively, so a new event variant must be classified deliberately. --- crates/irtt-client/src/managed/model.rs | 12 ++ crates/irtt-client/src/managed/task.rs | 165 ++++++++++++++---------- crates/irtt-client/src/managed/tests.rs | 103 +++++++++++++++ 3 files changed, 209 insertions(+), 71 deletions(-) diff --git a/crates/irtt-client/src/managed/model.rs b/crates/irtt-client/src/managed/model.rs index fac057f..5083e3d 100644 --- a/crates/irtt-client/src/managed/model.rs +++ b/crates/irtt-client/src/managed/model.rs @@ -198,16 +198,28 @@ pub enum ManagedEndReason { } /// Durable outcome for one target generation. +/// +/// These counters are durable: unlike [`ManagedEvent`], which is a lossy +/// presentation stream, they account for everything the target observed. #[derive(Clone, Debug, Eq, PartialEq)] pub struct ManagedTargetOutcome { pub target: TargetInstance, pub server_addr: Arc, pub remote: Option, pub end_reason: ManagedTargetEndReason, + /// Probes the underlying client reported as sent. pub packets_sent: u64, + /// On-time unique echo replies. + /// + /// A unique reply that arrived after its probe timed out is counted by + /// `late` instead, so this is the count of replies that arrived in time. pub replies_received: u64, + /// Replies for a sequence that had already been answered. pub duplicates: u64, + /// Replies that arrived after their probe timed out, whether or not the + /// client could still match them to retained send state. pub late: u64, + /// Warning events observed for this target. pub warning_events: u64, pub cleanup_failure: Option, } diff --git a/crates/irtt-client/src/managed/task.rs b/crates/irtt-client/src/managed/task.rs index f5571a0..b4496f0 100644 --- a/crates/irtt-client/src/managed/task.rs +++ b/crates/irtt-client/src/managed/task.rs @@ -394,15 +394,51 @@ impl TargetState { } } -#[derive(Default)] +#[derive(Debug, Default, Eq, PartialEq)] struct TargetCounters { packets_sent: u64, + /// On-time unique echo replies. + /// + /// A late unique reply is counted by `late` and deliberately not here, so + /// `replies_received` stays the count of replies that arrived within their + /// probe timeout. These counters are durable, unlike the lossy managed + /// presentation events. replies_received: u64, duplicates: u64, late: u64, warning_events: u64, } +impl TargetCounters { + /// Account for one client event. + /// + /// Events that carry no durable target-local count are ignored. Probe + /// sends are not counted here: `packets_sent` is authoritative in the + /// underlying client and is synchronized through + /// [`TargetRuntime::sync_packets_sent`]. + fn observe(&mut self, event: &ClientEvent) { + match event { + ClientEvent::EchoReply { .. } => { + self.replies_received = self.replies_received.saturating_add(1); + } + ClientEvent::DuplicateReply { .. } => { + self.duplicates = self.duplicates.saturating_add(1); + } + ClientEvent::LateReply { .. } => { + self.late = self.late.saturating_add(1); + } + ClientEvent::Warning { .. } => { + self.warning_events = self.warning_events.saturating_add(1); + } + ClientEvent::EchoSent { .. } + | ClientEvent::EchoLoss { .. } + | ClientEvent::SessionStarted { .. } + | ClientEvent::NoTestCompleted { .. } + | ClientEvent::SessionClosed { .. } => {} + } + } +} + struct TargetRuntime { instance: TargetInstance, config: ManagedTargetConfig, @@ -416,6 +452,46 @@ struct TargetRuntime { state: TargetState, } +impl TargetRuntime { + /// Adopt the underlying client's authoritative sent count. + /// + /// The client owns `packets_sent`; this runtime only mirrors it so a + /// terminal outcome can report it after the client is gone. + fn sync_packets_sent(&mut self, client: &AsyncClient) { + self.record_packets_sent(client.packets_sent()); + } + + /// Adopt an already-read authoritative sent count. + fn record_packets_sent(&mut self, packets_sent: u64) { + self.counters.packets_sent = packets_sent; + } + + /// Account for one client event against this target's durable counters. + fn observe(&mut self, event: &ClientEvent) { + self.counters.observe(event); + } + + /// Build this target's durable outcome. + fn outcome( + &self, + end_reason: ManagedTargetEndReason, + cleanup_failure: Option, + ) -> ManagedTargetOutcome { + ManagedTargetOutcome { + target: self.instance.clone(), + server_addr: Arc::clone(&self.server_addr), + remote: self.remote, + end_reason, + packets_sent: self.counters.packets_sent, + replies_received: self.counters.replies_received, + duplicates: self.counters.duplicates, + late: self.counters.late, + warning_events: self.counters.warning_events, + cleanup_failure, + } + } +} + #[derive(Clone, Copy, Debug, Eq, PartialEq)] struct TimeoutBacklogEntry { index: usize, @@ -454,25 +530,6 @@ fn synchronously_retireable(state: &TargetState) -> bool { } } -fn target_outcome( - target: &TargetRuntime, - end_reason: ManagedTargetEndReason, - cleanup_failure: Option, -) -> ManagedTargetOutcome { - ManagedTargetOutcome { - target: target.instance.clone(), - server_addr: Arc::clone(&target.server_addr), - remote: target.remote, - end_reason, - packets_sent: target.counters.packets_sent, - replies_received: target.counters.replies_received, - duplicates: target.counters.duplicates, - late: target.counters.late, - warning_events: target.counters.warning_events, - cleanup_failure, - } -} - #[derive(Default)] struct OutcomeHistory { limit: usize, @@ -709,29 +766,7 @@ impl ManagedClientTask { fn publish_client_events(&mut self, index: usize, events: Vec) { for event in events { - match event { - ClientEvent::EchoReply { .. } => { - self.targets[index].counters.replies_received = self.targets[index] - .counters - .replies_received - .saturating_add(1); - } - ClientEvent::DuplicateReply { .. } => { - self.targets[index].counters.duplicates = - self.targets[index].counters.duplicates.saturating_add(1); - } - ClientEvent::LateReply { .. } => { - self.targets[index].counters.late = - self.targets[index].counters.late.saturating_add(1); - } - ClientEvent::Warning { .. } => { - self.targets[index].counters.warning_events = self.targets[index] - .counters - .warning_events - .saturating_add(1); - } - _ => {} - } + self.targets[index].observe(&event); self.publish_event(ManagedEvent::Client { target: self.targets[index].instance.clone(), event, @@ -746,19 +781,7 @@ impl ManagedClientTask { cleanup_failure: Option, ) { self.install_target_state(index, TargetState::Terminal); - let target = &self.targets[index]; - let outcome = ManagedTargetOutcome { - target: target.instance.clone(), - server_addr: Arc::clone(&target.server_addr), - remote: target.remote, - end_reason, - packets_sent: target.counters.packets_sent, - replies_received: target.counters.replies_received, - duplicates: target.counters.duplicates, - late: target.counters.late, - warning_events: target.counters.warning_events, - cleanup_failure, - }; + let outcome = self.targets[index].outcome(end_reason, cleanup_failure); self.history.record(outcome.clone()); self.replace_status(); self.publish_event(ManagedEvent::TargetFinished { @@ -782,7 +805,7 @@ impl ManagedClientTask { ) -> bool { let primary_end = ManagedTargetEndReason::Failed(classify_client_error(phase, &error)); client.discard_prepared_probe(); - self.targets[index].counters.packets_sent = client.packets_sent(); + self.targets[index].sync_packets_sent(&client); match cleanup { OpenSessionFailureCleanup::Drain => self.begin_drain(index, client, primary_end, now), OpenSessionFailureCleanup::Close => { @@ -866,7 +889,7 @@ impl ManagedClientTask { if retirement.synchronous { synchronous.insert(retirement.index); if !matches!(target.state, TargetState::Terminal) { - let outcome = target_outcome(target, retirement.reason.clone(), None); + let outcome = target.outcome(retirement.reason.clone(), None); self.history.record(outcome.clone()); finished.push(ManagedEvent::TargetFinished { outcome: Arc::new(outcome), @@ -1137,7 +1160,7 @@ impl ManagedClientTask { .effective_retirement(index) .unwrap_or(ManagedTargetEndReason::Stopped); if !open.has_in_flight_work() { - self.targets[index].counters.packets_sent = client.packets_sent(); + self.targets[index].sync_packets_sent(&client); self.finish_target(index, retirement, None); return false; } @@ -1153,7 +1176,7 @@ impl ManagedClientTask { if self.state == DriverState::Stopping || self.effective_retirement(index).is_some() { - self.targets[index].counters.packets_sent = client.packets_sent(); + self.targets[index].sync_packets_sent(&client); self.begin_drain( index, client, @@ -1207,7 +1230,7 @@ impl ManagedClientTask { if self.state == DriverState::Stopping || self.effective_retirement(index).is_some() { client.discard_prepared_probe(); - self.targets[index].counters.packets_sent = client.packets_sent(); + self.targets[index].sync_packets_sent(&client); return self.begin_drain( index, client, @@ -1241,12 +1264,12 @@ impl ManagedClientTask { } }; if client.is_peer_closed() { - self.targets[index].counters.packets_sent = client.packets_sent(); + self.targets[index].sync_packets_sent(&client); self.finish_target(index, ManagedTargetEndReason::PeerClosed, None); return false; } if client.is_run_complete() { - self.targets[index].counters.packets_sent = client.packets_sent(); + self.targets[index].sync_packets_sent(&client); return self.begin_drain( index, client, @@ -1309,7 +1332,7 @@ impl ManagedClientTask { true } Poll::Ready(Err(error)) => { - self.targets[index].counters.packets_sent = client.packets_sent(); + self.targets[index].sync_packets_sent(&client); cleanup_failure.get_or_insert_with(|| { classify_client_error(ManagedTargetFailurePhase::Receiving, &error) }); @@ -1317,14 +1340,14 @@ impl ManagedClientTask { } }; if let Some(error) = injected_receive_failure { - self.targets[index].counters.packets_sent = client.packets_sent(); + self.targets[index].sync_packets_sent(&client); cleanup_failure.get_or_insert_with(|| { classify_client_error(ManagedTargetFailurePhase::Receiving, &error) }); return self.begin_close(index, client, primary_end, cleanup_failure, now); } if client.is_peer_closed() { - self.targets[index].counters.packets_sent = client.packets_sent(); + self.targets[index].sync_packets_sent(&client); self.finish_target(index, ManagedTargetEndReason::PeerClosed, cleanup_failure); return false; } @@ -1380,7 +1403,7 @@ impl ManagedClientTask { } => match client.poll_close(cx) { Poll::Pending => { if now >= deadline { - self.targets[index].counters.packets_sent = client.packets_sent(); + self.targets[index].sync_packets_sent(&client); self.finish_target( index, primary_end, @@ -1398,7 +1421,7 @@ impl ManagedClientTask { } } Poll::Ready(Ok(events)) => { - self.targets[index].counters.packets_sent = client.packets_sent(); + self.targets[index].sync_packets_sent(&client); self.publish_client_events(index, events); #[cfg(test)] let injected_close_failure = self.take_drain_close_failure().map(|error| { @@ -1414,7 +1437,7 @@ impl ManagedClientTask { false } Poll::Ready(Err(error)) => { - self.targets[index].counters.packets_sent = client.packets_sent(); + self.targets[index].sync_packets_sent(&client); let cleanup = classify_client_error(ManagedTargetFailurePhase::Closing, &error); self.finish_target(index, primary_end, cleanup_failure.or(Some(cleanup))); false @@ -1620,7 +1643,7 @@ impl ManagedClientTask { .then(|| client.skip_missed_probe_slots_at(Instant::now())) .transpose() .err(); - self.targets[index].counters.packets_sent = after; + self.targets[index].record_packets_sent(after); match result { Poll::Pending => { self.targets[index].send_waiting = true; @@ -1875,7 +1898,7 @@ impl ManagedClientTask { } } Err(error) => { - self.targets[index].counters.packets_sent = client.packets_sent(); + self.targets[index].sync_packets_sent(&client); cleanup_failure.get_or_insert_with(|| { classify_client_error(ManagedTargetFailurePhase::Timing, &error) }); diff --git a/crates/irtt-client/src/managed/tests.rs b/crates/irtt-client/src/managed/tests.rs index 284d0c2..e7c9e56 100644 --- a/crates/irtt-client/src/managed/tests.rs +++ b/crates/irtt-client/src/managed/tests.rs @@ -2954,3 +2954,106 @@ fn terminal_subscription_is_closed() { Err(ManagedSubscribeError::Closed) )); } + +/// Build the client events target accounting distinguishes. +fn accounting_events() -> Vec { + let remote: std::net::SocketAddr = "127.0.0.1:2112".parse().unwrap(); + let at = ClientTimestamp { + mono: Instant::now(), + wall: SystemTime::UNIX_EPOCH, + }; + let rtt = crate::RttSample { + raw: Duration::from_millis(1), + adjusted: None, + effective: crate::SignedDuration::from_nanos(1_000_000), + }; + vec![ + ClientEvent::EchoSent { + seq: 0, + remote, + scheduled_at: at.mono, + sent_at: at, + bytes: 64, + send_call: Duration::ZERO, + timer_error: Duration::ZERO, + }, + ClientEvent::EchoReply { + seq: 0, + remote, + sent_at: at, + received_at: at, + rtt, + server_timing: None, + one_way: None, + received_stats: None, + bytes: 64, + packet_meta: crate::PacketMeta::default(), + }, + ClientEvent::DuplicateReply { + seq: 0, + remote, + received_at: at, + bytes: 64, + }, + ClientEvent::LateReply { + seq: 1, + highest_seen: 2, + remote, + sent_at: Some(at), + received_at: at, + rtt: Some(rtt), + server_timing: None, + one_way: None, + received_stats: None, + bytes: 64, + packet_meta: crate::PacketMeta::default(), + }, + ClientEvent::EchoLoss { + seq: 2, + sent_at: at, + timeout_at: at.mono, + }, + ClientEvent::Warning { + kind: crate::WarningKind::WrongToken, + message: "wrong token".to_owned(), + at, + }, + ] +} + +#[test] +fn target_counters_account_for_each_observed_event_once() { + let mut counters = TargetCounters::default(); + for event in accounting_events() { + counters.observe(&event); + } + + // A late unique reply is counted as late, not as a received reply, and a + // probe send is not accounted here because the client owns packets_sent. + assert_eq!(counters.packets_sent, 0); + assert_eq!(counters.replies_received, 1); + assert_eq!(counters.duplicates, 1); + assert_eq!(counters.late, 1); + assert_eq!(counters.warning_events, 1); +} + +#[test] +fn target_counters_ignore_events_without_a_durable_count() { + let ignored: Vec = accounting_events() + .into_iter() + .filter(|event| { + matches!( + event, + ClientEvent::EchoSent { .. } | ClientEvent::EchoLoss { .. } + ) + }) + .collect(); + assert_eq!(ignored.len(), 2); + + let mut counters = TargetCounters::default(); + for event in ignored { + counters.observe(&event); + } + + assert_eq!(counters, TargetCounters::default()); +}