feat(client): implement automatic session reconnection and disable statistics
Rust CI / cargo fmt (push) Failing after 16s
Rust CI / cargo check (x86_64-unknown-linux-musl) (push) Failing after 7s
Rust CI / cargo test (x86_64-unknown-linux-musl) (push) Failing after 10s
Rust CI / cargo check (x86_64-unknown-linux-gnu) (push) Successful in 36s
Rust CI / cargo clippy (push) Successful in 37s
Netem Interop / netem interop (push) Failing after 14s
Supply chain / cargo-deny advisories (push) Failing after 4s
Rust CI / cargo doc (push) Successful in 26s
Supply chain / cargo-deny bans licenses sources (push) Failing after 4s
Supply chain / actionlint (push) Successful in 3s
Fuzz Smoke / cargo fuzz smoke (push) Failing after 31s
Rust CI / cargo test (x86_64-unknown-linux-gnu) (push) Successful in 1m19s
Rust CI / cargo fmt (push) Failing after 16s
Rust CI / cargo check (x86_64-unknown-linux-musl) (push) Failing after 7s
Rust CI / cargo test (x86_64-unknown-linux-musl) (push) Failing after 10s
Rust CI / cargo check (x86_64-unknown-linux-gnu) (push) Successful in 36s
Rust CI / cargo clippy (push) Successful in 37s
Netem Interop / netem interop (push) Failing after 14s
Supply chain / cargo-deny advisories (push) Failing after 4s
Rust CI / cargo doc (push) Successful in 26s
Supply chain / cargo-deny bans licenses sources (push) Failing after 4s
Supply chain / actionlint (push) Successful in 3s
Fuzz Smoke / cargo fuzz smoke (push) Failing after 31s
Rust CI / cargo test (x86_64-unknown-linux-gnu) (push) Successful in 1m19s
This commit is contained in:
9 files changed
+282
-35
No files matched your search
@@ -55,7 +55,7 @@ pub struct ClientArgs {
|
||||
default_value = "10s",
|
||||
value_parser = parse_test_duration,
|
||||
help = "Test duration (e.g. 2.5s, 1m30s, 1h); use --continuous for continuous mode (0 is also accepted)",
|
||||
long_help = "Test duration (e.g. 2.5s, 1m30s, 500us, 100ns, 1h); use --continuous for continuous mode (0 is also accepted).\n\nFinite runs retain exact statistics for final summaries. Continuous mode uses bounded-memory running statistics and prints a final summary when interrupted; --targets-stdin also prints the final retained summary on EOF."
|
||||
long_help = "Test duration (e.g. 2.5s, 1m30s, 500us, 100ns, 1h); use --continuous for continuous mode (0 is also accepted).\n\nUnless --no-stats is selected, finite runs retain exact statistics for final summaries. Continuous mode uses bounded-memory running statistics and prints a final summary when interrupted; --targets-stdin also prints the final retained summary on EOF."
|
||||
)]
|
||||
pub duration: Duration,
|
||||
|
||||
@@ -63,6 +63,22 @@ pub struct ClientArgs {
|
||||
#[arg(long, conflicts_with = "duration")]
|
||||
pub continuous: bool,
|
||||
|
||||
/// Disable statistics collection and final summaries; retain measurement event output.
|
||||
#[arg(long)]
|
||||
pub no_stats: bool,
|
||||
|
||||
/// Retry failed, closed, or silent sessions (requires continuous mode).
|
||||
#[arg(
|
||||
long,
|
||||
conflicts_with = "no_test",
|
||||
long_help = "Retry failed initial opens and closed, failed, or silent sessions independently per target. Requires --continuous or --duration 0. Unsuccessful connections back off from 250ms to 10s. Use --reconnect-timeout to configure silence detection and --no-stats when only measurement events are needed."
|
||||
)]
|
||||
pub reconnect: bool,
|
||||
|
||||
/// Reconnect after this long without valid replies while sending probes.
|
||||
#[arg(long, default_value = "10s", value_parser = crate::shared::client::parse_duration, requires = "reconnect")]
|
||||
pub reconnect_timeout: Duration,
|
||||
|
||||
#[command(flatten)]
|
||||
pub common: CommonClientArgs,
|
||||
|
||||
@@ -112,6 +128,9 @@ impl ClientArgs {
|
||||
/// Targets are only required here. `--list-columns` returns before this is
|
||||
/// called, so listing columns still needs no target.
|
||||
pub async fn prepare(&self) -> Result<ManagedRunSetup, String> {
|
||||
if self.reconnect && !self.is_continuous() {
|
||||
return Err("--reconnect requires --continuous or --duration 0".to_owned());
|
||||
}
|
||||
if self.targets_stdin && !self.is_continuous() {
|
||||
return Err("--targets-stdin requires --continuous or --duration 0".to_owned());
|
||||
}
|
||||
|
||||
@@ -58,8 +58,11 @@ pub async fn run_stream(
|
||||
args.verbose,
|
||||
)
|
||||
.map_err(|err| io::Error::new(io::ErrorKind::InvalidInput, err))?;
|
||||
if let Some(warning) = reconnect_stats_warning(&args) {
|
||||
eprintln!("{warning}");
|
||||
}
|
||||
if setup.stdin_controlled {
|
||||
return run_stdin_stream(setup, output_config, shutdown).await;
|
||||
return run_stdin_stream(setup, output_config, shutdown, &args).await;
|
||||
}
|
||||
let continuous = args.is_continuous();
|
||||
let target_count = setup.target_count();
|
||||
@@ -70,12 +73,21 @@ pub async fn run_stream(
|
||||
if *shutdown.borrow() {
|
||||
return Ok(());
|
||||
}
|
||||
let (task, handle) = ManagedClient::task(setup.managed_config(), setup.managed_targets())?;
|
||||
let (task, handle) = ManagedClient::task(
|
||||
{
|
||||
let mut config = setup.managed_config();
|
||||
config.reconnect_timeout = args.reconnect.then_some(args.reconnect_timeout);
|
||||
config
|
||||
},
|
||||
setup.managed_targets(),
|
||||
)?;
|
||||
let mut events = handle.subscribe()?;
|
||||
let mut status = handle.subscribe_status();
|
||||
let owner = ManagedWorker::start(task, handle.clone())?;
|
||||
let mut stdout = io::LineWriter::new(io::stdout().lock());
|
||||
let mut stream_output = StreamOutput {
|
||||
continuous: args.is_continuous(),
|
||||
collect_stats: !args.no_stats,
|
||||
config: output_config,
|
||||
header_printed: false,
|
||||
print_final_summary: false,
|
||||
@@ -86,6 +98,7 @@ pub async fn run_stream(
|
||||
let mut stats = setup
|
||||
.targets
|
||||
.iter()
|
||||
.filter(|_| !args.no_stats)
|
||||
.map(|target| {
|
||||
(
|
||||
target.label.clone(),
|
||||
@@ -168,13 +181,13 @@ pub async fn run_stream(
|
||||
}),
|
||||
};
|
||||
stream_output.print_final_summary =
|
||||
!args.no_test && should_print_final_summary(continuous, interrupted);
|
||||
!args.no_stats && !args.no_test && should_print_final_summary(continuous, interrupted);
|
||||
stream_output.show_running_only_summary_note =
|
||||
continuous && interrupted && stream_output.print_final_summary;
|
||||
for target in &setup.targets {
|
||||
let target_stats = stats.get(&target.label).expect(
|
||||
"every prepared target must have a stats collector inserted before the run begins",
|
||||
);
|
||||
let Some(target_stats) = stats.get(&target.label) else {
|
||||
continue;
|
||||
};
|
||||
if multi_target
|
||||
&& stream_output.print_final_summary
|
||||
&& stream_output.config.prints_summary()
|
||||
@@ -361,17 +374,22 @@ fn process_stdin_event<W: Write>(
|
||||
terminal_targets: &mut BoundedTargetSet,
|
||||
) -> io::Result<()> {
|
||||
match event {
|
||||
ManagedEvent::TargetStateChanged { target, .. } => {
|
||||
ManagedEvent::TargetStateChanged { target, .. } if stream_output.collect_stats => {
|
||||
stats
|
||||
.entry(target)
|
||||
.or_insert_with(|| StatsCollector::new(stats_config(true)));
|
||||
}
|
||||
ManagedEvent::Client { target, event } => {
|
||||
let collector = stats
|
||||
.entry(target.clone())
|
||||
.or_insert_with(|| StatsCollector::new(stats_config(true)));
|
||||
let update = collector.process(&event);
|
||||
stream_output.print_event(&event, Some(target.id.as_str()), &update)?;
|
||||
let update = stream_output.collect_stats.then(|| {
|
||||
let collector = stats
|
||||
.entry(target.clone())
|
||||
.or_insert_with(|| StatsCollector::new(stats_config(true)));
|
||||
if matches!(event, ClientEvent::SessionStarted(_)) {
|
||||
*collector = StatsCollector::new(stats_config(true));
|
||||
}
|
||||
collector.process(&event)
|
||||
});
|
||||
stream_output.print_event(&event, Some(target.id.as_str()), update.as_ref())?;
|
||||
}
|
||||
ManagedEvent::TargetFinished { outcome }
|
||||
if terminal_targets.insert(outcome.target.clone()) =>
|
||||
@@ -402,7 +420,11 @@ fn reconcile_stdin_stats(
|
||||
fn snapshot_stdin_summary_targets(
|
||||
stats: &mut BTreeMap<TargetInstance, StatsCollector>,
|
||||
status: &ManagedStatus,
|
||||
enabled: bool,
|
||||
) -> BTreeSet<TargetInstance> {
|
||||
if !enabled {
|
||||
return BTreeSet::new();
|
||||
}
|
||||
let targets = status
|
||||
.targets
|
||||
.iter()
|
||||
@@ -427,11 +449,19 @@ async fn run_stdin_stream(
|
||||
setup: ManagedRunSetup,
|
||||
output_config: OutputConfig,
|
||||
mut shutdown: tokio::sync::watch::Receiver<bool>,
|
||||
args: &ClientArgs,
|
||||
) -> Result<(), Box<dyn std::error::Error>> {
|
||||
if *shutdown.borrow() {
|
||||
return Ok(());
|
||||
}
|
||||
let (task, handle) = ManagedClient::task(setup.managed_config(), setup.managed_targets())?;
|
||||
let (task, handle) = ManagedClient::task(
|
||||
{
|
||||
let mut config = setup.managed_config();
|
||||
config.reconnect_timeout = args.reconnect.then_some(args.reconnect_timeout);
|
||||
config
|
||||
},
|
||||
setup.managed_targets(),
|
||||
)?;
|
||||
let mut events = handle.subscribe()?;
|
||||
let mut status = handle.subscribe_status();
|
||||
let owner = ManagedWorker::start(task, handle.clone())?;
|
||||
@@ -440,6 +470,8 @@ async fn run_stdin_stream(
|
||||
|
||||
let mut stdout = io::LineWriter::new(io::stdout().lock());
|
||||
let mut stream_output = StreamOutput {
|
||||
continuous: args.is_continuous(),
|
||||
collect_stats: !args.no_stats,
|
||||
config: output_config,
|
||||
header_printed: false,
|
||||
print_final_summary: false,
|
||||
@@ -510,7 +542,11 @@ async fn run_stdin_stream(
|
||||
}
|
||||
if stop.is_some() {
|
||||
preparing = None;
|
||||
summary_targets = Some(snapshot_stdin_summary_targets(&mut stats, &snapshot));
|
||||
summary_targets = Some(snapshot_stdin_summary_targets(
|
||||
&mut stats,
|
||||
&snapshot,
|
||||
!args.no_stats,
|
||||
));
|
||||
if request_managed_stop_once(&mut stop_requested) {
|
||||
drop(handle.stop());
|
||||
}
|
||||
@@ -537,7 +573,7 @@ async fn run_stdin_stream(
|
||||
}
|
||||
Err(error) => {
|
||||
stop = Some(StdinStop::Fatal(format!("invalid --targets-stdin target set: {error}")));
|
||||
summary_targets = Some(snapshot_stdin_summary_targets(&mut stats, &status.borrow()));
|
||||
summary_targets = Some(snapshot_stdin_summary_targets(&mut stats, &status.borrow(), !args.no_stats));
|
||||
if request_managed_stop_once(&mut stop_requested) { drop(handle.stop()); }
|
||||
}
|
||||
}
|
||||
@@ -558,7 +594,7 @@ async fn run_stdin_stream(
|
||||
}
|
||||
Err(_) => {
|
||||
stop = Some(StdinStop::Fatal("--targets-stdin update was rejected".to_owned()));
|
||||
summary_targets = Some(snapshot_stdin_summary_targets(&mut stats, &status.borrow()));
|
||||
summary_targets = Some(snapshot_stdin_summary_targets(&mut stats, &status.borrow(), !args.no_stats));
|
||||
if request_managed_stop_once(&mut stop_requested) { drop(handle.stop()); }
|
||||
}
|
||||
}
|
||||
@@ -595,7 +631,8 @@ async fn run_stdin_stream(
|
||||
);
|
||||
}
|
||||
|
||||
let print_summary = matches!(stop, Some(StdinStop::Interrupted | StdinStop::Eof));
|
||||
let print_summary =
|
||||
!args.no_stats && matches!(stop, Some(StdinStop::Interrupted | StdinStop::Eof));
|
||||
stream_output.print_final_summary = print_summary;
|
||||
stream_output.show_running_only_summary_note = print_summary;
|
||||
if let Some(summary_targets) = summary_targets {
|
||||
@@ -636,11 +673,17 @@ fn process_event<W: Write>(
|
||||
) -> io::Result<()> {
|
||||
match event {
|
||||
ManagedEvent::Client { target, event } => {
|
||||
let collector = stats
|
||||
.entry(target.id.as_str().to_owned())
|
||||
.or_insert_with(|| StatsCollector::new(stats_config(false)));
|
||||
let update = collector.process(&event);
|
||||
stream_output.print_event(&event, Some(target.id.as_str()), &update)?;
|
||||
let update = stream_output.collect_stats.then(|| {
|
||||
let collector = stats
|
||||
.get_mut(target.id.as_str())
|
||||
.expect("prepared target has a statistics collector");
|
||||
if matches!(event, ClientEvent::SessionStarted(_)) {
|
||||
let config = stats_config(stream_output.continuous);
|
||||
*collector = StatsCollector::new(config);
|
||||
}
|
||||
collector.process(&event)
|
||||
});
|
||||
stream_output.print_event(&event, Some(target.id.as_str()), update.as_ref())?;
|
||||
}
|
||||
ManagedEvent::TargetFinished { outcome } => {
|
||||
terminal_targets.insert(outcome.target.clone());
|
||||
@@ -678,6 +721,8 @@ fn dropped_event_warning(dropped_events: u64) -> Option<String> {
|
||||
}
|
||||
|
||||
struct StreamOutput<'a, W: Write> {
|
||||
continuous: bool,
|
||||
collect_stats: bool,
|
||||
config: OutputConfig,
|
||||
header_printed: bool,
|
||||
print_final_summary: bool,
|
||||
@@ -689,10 +734,10 @@ impl<W: Write> StreamOutput<'_, W> {
|
||||
&mut self,
|
||||
event: &ClientEvent,
|
||||
target: Option<&str>,
|
||||
update: &EventStatsUpdate,
|
||||
update: Option<&EventStatsUpdate>,
|
||||
) -> io::Result<()> {
|
||||
self.print_header()?;
|
||||
if let Some(line) = self.config.render_event(event, target, Some(update)) {
|
||||
if let Some(line) = self.config.render_event(event, target, update) {
|
||||
writeln!(self.out, "{line}")?;
|
||||
}
|
||||
Ok(())
|
||||
@@ -734,7 +779,7 @@ fn stats_config(continuous: bool) -> StatsConfig {
|
||||
}
|
||||
}
|
||||
fn finite_stats_memory_warning(args: &ClientArgs, target_count: usize) -> Option<String> {
|
||||
if args.no_test || args.is_continuous() || args.duration.is_zero() {
|
||||
if args.no_stats || args.no_test || args.is_continuous() || args.duration.is_zero() {
|
||||
return None;
|
||||
}
|
||||
let target_count = u64::try_from(target_count).unwrap_or(u64::MAX);
|
||||
@@ -761,3 +806,8 @@ fn finite_stats_memory_warning(args: &ClientArgs, target_count: usize) -> Option
|
||||
};
|
||||
Some(format!("irtt-rs: warning: finite exact statistics may retain about {formatted} for this run; {guidance}"))
|
||||
}
|
||||
|
||||
fn reconnect_stats_warning(args: &ClientArgs) -> Option<&'static str> {
|
||||
(args.reconnect && !args.no_stats).then_some("irtt-rs: warning: statistics spanning multiple sessions may not accurately represent one uninterrupted measurement run; use --no-stats for measurement events only. Session statistics reset on reconnect.")
|
||||
}
|
||||
|
||||
@@ -101,6 +101,9 @@ pub struct ManagedClientConfig {
|
||||
pub outcome_history_limit: usize,
|
||||
pub max_live_target_generations: usize,
|
||||
pub final_drain: Duration,
|
||||
/// Retry failed/closed sessions and detect silence while sending. Only valid
|
||||
/// for continuous measurement runs. None (the default) disables reconnection.
|
||||
pub reconnect_timeout: Option<Duration>,
|
||||
}
|
||||
|
||||
impl Default for ManagedClientConfig {
|
||||
@@ -114,6 +117,7 @@ impl Default for ManagedClientConfig {
|
||||
outcome_history_limit: DEFAULT_MANAGED_OUTCOME_HISTORY_LIMIT,
|
||||
max_live_target_generations: DEFAULT_MANAGED_MAX_LIVE_TARGET_GENERATIONS,
|
||||
final_drain: DEFAULT_MANAGED_FINAL_DRAIN,
|
||||
reconnect_timeout: None,
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -364,6 +368,8 @@ pub struct ManagedTargetFailure {
|
||||
/// Configuration failure detected before any runtime work begins.
|
||||
#[derive(Debug, Error)]
|
||||
pub enum ManagedConfigError {
|
||||
#[error("reconnection requires continuous measurement mode and a positive, schedulable silence timeout")]
|
||||
InvalidReconnectPolicy,
|
||||
#[error("FinishWhenQuiescent requires at least one initial target")]
|
||||
EmptyInitialTargets,
|
||||
#[error("duplicate managed target id {id}")]
|
||||
|
||||
@@ -327,7 +327,14 @@ impl ManagedClientTask {
|
||||
let mut non_send_deadline = None;
|
||||
let mut send_deadline = None;
|
||||
for target in &self.targets {
|
||||
non_send_deadline = non_send_deadline
|
||||
.into_iter()
|
||||
.chain(target.silence_deadline(self.config.reconnect_timeout))
|
||||
.min();
|
||||
match target.phase() {
|
||||
TargetPhase::Retrying { deadline } => {
|
||||
non_send_deadline = non_send_deadline.into_iter().chain(Some(*deadline)).min();
|
||||
}
|
||||
TargetPhase::Active {
|
||||
client,
|
||||
schedule,
|
||||
|
||||
@@ -367,6 +367,15 @@ fn build_task(
|
||||
maximum: tokio::sync::Semaphore::MAX_PERMITS,
|
||||
});
|
||||
}
|
||||
if let Some(timeout) = config.reconnect_timeout {
|
||||
if timeout.is_zero()
|
||||
|| Instant::now().checked_add(timeout).is_none()
|
||||
|| config.client.request.duration.is_some()
|
||||
|| config.client.request.run_mode != crate::RunMode::Normal
|
||||
{
|
||||
return Err(ManagedConfigError::InvalidReconnectPolicy);
|
||||
}
|
||||
}
|
||||
if config.max_live_target_generations == 0 {
|
||||
return Err(ManagedConfigError::ZeroLiveGenerationLimit);
|
||||
}
|
||||
@@ -417,6 +426,10 @@ fn build_task(
|
||||
server_addr: Arc::from(target.server_addr),
|
||||
remote: None,
|
||||
counters: TargetCounters::default(),
|
||||
sent_before_session: 0,
|
||||
retry_delay: std::time::Duration::from_millis(250),
|
||||
silence_since: None,
|
||||
last_reply_at: None,
|
||||
phase: Some(TargetPhase::Pending { client_config }),
|
||||
});
|
||||
}
|
||||
|
||||
@@ -206,6 +206,12 @@ impl ManagedClientTask {
|
||||
};
|
||||
self.targets[index].sync_packets_sent(&client);
|
||||
send_waiting = result.is_pending();
|
||||
if send_waiting {
|
||||
// A blocked sender is not evidence of an unresponsive peer. Arm
|
||||
// a fresh silence period only after transmission resumes.
|
||||
self.targets[index].silence_since = None;
|
||||
self.targets[index].last_reply_at = None;
|
||||
}
|
||||
self.targets[index].restore_phase(TargetPhase::Active {
|
||||
client,
|
||||
schedule,
|
||||
|
||||
@@ -5,7 +5,7 @@ use std::{
|
||||
pin::Pin,
|
||||
sync::Arc,
|
||||
task::{Context, Poll},
|
||||
time::Instant,
|
||||
time::{Duration, Instant},
|
||||
};
|
||||
|
||||
use crate::{
|
||||
@@ -27,6 +27,9 @@ type ConnectFuture =
|
||||
Pin<Box<dyn Future<Output = Result<AsyncClient, ClientError>> + Send + 'static>>;
|
||||
|
||||
pub(super) enum TargetPhase {
|
||||
Retrying {
|
||||
deadline: Instant,
|
||||
},
|
||||
Pending {
|
||||
client_config: ClientConfig,
|
||||
},
|
||||
@@ -65,6 +68,7 @@ pub(super) enum TargetPhase {
|
||||
impl TargetPhase {
|
||||
pub(super) fn lifecycle(&self) -> ManagedTargetLifecycle {
|
||||
match self {
|
||||
Self::Retrying { .. } => ManagedTargetLifecycle::Pending,
|
||||
Self::Pending { .. } => ManagedTargetLifecycle::Pending,
|
||||
Self::Connecting { .. } => ManagedTargetLifecycle::Connecting,
|
||||
Self::Opening { .. } => ManagedTargetLifecycle::Opening,
|
||||
@@ -182,11 +186,30 @@ pub(super) struct TargetRuntime {
|
||||
pub(super) server_addr: Arc<str>,
|
||||
pub(super) remote: Option<std::net::SocketAddr>,
|
||||
pub(super) counters: TargetCounters,
|
||||
pub(super) sent_before_session: u64,
|
||||
pub(super) retry_delay: Duration,
|
||||
pub(super) silence_since: Option<Instant>,
|
||||
pub(super) last_reply_at: Option<Instant>,
|
||||
// None only while a synchronous operation owns the phase locally.
|
||||
pub(super) phase: Option<TargetPhase>,
|
||||
}
|
||||
|
||||
impl TargetRuntime {
|
||||
pub(super) fn silence_deadline(&self, timeout: Option<Duration>) -> Option<Instant> {
|
||||
let TargetPhase::Active {
|
||||
schedule,
|
||||
send_waiting,
|
||||
..
|
||||
} = self.phase()
|
||||
else {
|
||||
return None;
|
||||
};
|
||||
if *send_waiting || schedule.is_finished() {
|
||||
return None;
|
||||
}
|
||||
self.silence_since?.checked_add(timeout?)
|
||||
}
|
||||
|
||||
pub(super) fn phase(&self) -> &TargetPhase {
|
||||
self.phase
|
||||
.as_ref()
|
||||
@@ -218,16 +241,18 @@ impl TargetRuntime {
|
||||
self.membership.is_desired() && matches!(phase, TargetPhase::Active { .. })
|
||||
}
|
||||
|
||||
/// 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.
|
||||
/// Combine the current client's authoritative sent count with retired
|
||||
/// sessions, so one target generation keeps durable accounting on retries.
|
||||
pub(super) fn sync_packets_sent(&mut self, client: &AsyncClient) {
|
||||
self.counters.packets_sent = client.packets_sent();
|
||||
self.counters.packets_sent = self
|
||||
.sent_before_session
|
||||
.saturating_add(client.packets_sent());
|
||||
}
|
||||
|
||||
pub(super) fn sync_live_packets_sent(&mut self) {
|
||||
self.counters.packets_sent = self.phase().client().packets_sent();
|
||||
self.counters.packets_sent = self
|
||||
.sent_before_session
|
||||
.saturating_add(self.phase().client().packets_sent());
|
||||
}
|
||||
|
||||
/// Build this target's durable outcome.
|
||||
@@ -284,6 +309,30 @@ impl ManagedClientTask {
|
||||
|
||||
pub(super) fn publish_client_events(&mut self, index: usize, events: Vec<ClientEvent>) {
|
||||
for event in events {
|
||||
match &event {
|
||||
ClientEvent::EchoSent { sent_at, .. } => {
|
||||
let target = &mut self.targets[index];
|
||||
let anchor = target
|
||||
.last_reply_at
|
||||
.filter(|reply| {
|
||||
self.config.reconnect_timeout.is_some_and(|timeout| {
|
||||
sent_at.mono.duration_since(*reply) < timeout
|
||||
})
|
||||
})
|
||||
.unwrap_or(sent_at.mono);
|
||||
target.silence_since.get_or_insert(anchor);
|
||||
}
|
||||
ClientEvent::EchoReply { received_at, .. }
|
||||
| ClientEvent::LateReply { received_at, .. }
|
||||
| ClientEvent::DuplicateReply { received_at, .. } => {
|
||||
// A reply disarms silence until another probe is committed.
|
||||
// A healthy sender with a long interval can remain idle.
|
||||
self.targets[index].silence_since = None;
|
||||
self.targets[index].last_reply_at = Some(received_at.mono);
|
||||
self.targets[index].retry_delay = Duration::from_millis(250);
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
self.targets[index].counters.observe(&event);
|
||||
self.publish_event(ManagedEvent::Client {
|
||||
target: self.targets[index].instance.clone(),
|
||||
@@ -298,6 +347,23 @@ impl ManagedClientTask {
|
||||
end_reason: ManagedTargetEndReason,
|
||||
cleanup_failure: Option<ManagedTargetFailure>,
|
||||
) {
|
||||
if self.can_reconnect(index)
|
||||
&& matches!(
|
||||
end_reason,
|
||||
ManagedTargetEndReason::Failed(_)
|
||||
| ManagedTargetEndReason::PeerClosed
|
||||
| ManagedTargetEndReason::TestComplete
|
||||
)
|
||||
{
|
||||
let target = &mut self.targets[index];
|
||||
target.sent_before_session = target.counters.packets_sent;
|
||||
target.silence_since = None;
|
||||
target.last_reply_at = None;
|
||||
let deadline = Instant::now() + target.retry_delay;
|
||||
target.retry_delay = (target.retry_delay * 2).min(Duration::from_secs(10));
|
||||
self.install_target_phase(index, TargetPhase::Retrying { deadline });
|
||||
return;
|
||||
}
|
||||
let outcome = Arc::new(self.targets[index].outcome(end_reason, cleanup_failure));
|
||||
// Finalize history before the phase publisher installs Terminal, updates
|
||||
// pacing and publishes one complete status followed by the lifecycle event.
|
||||
@@ -311,6 +377,12 @@ impl ManagedClientTask {
|
||||
self.publish_event(ManagedEvent::TargetFinished { outcome });
|
||||
}
|
||||
|
||||
fn can_reconnect(&self, index: usize) -> bool {
|
||||
self.config.reconnect_timeout.is_some()
|
||||
&& !self.phase.is_stopping()
|
||||
&& self.targets[index].membership.is_desired()
|
||||
}
|
||||
|
||||
fn fail_target(&mut self, index: usize, phase: ManagedTargetFailurePhase, error: ClientError) {
|
||||
let failure = classify_client_error(phase, &error);
|
||||
self.finish_target(index, ManagedTargetEndReason::Failed(failure), None);
|
||||
@@ -332,6 +404,9 @@ impl ManagedClientTask {
|
||||
};
|
||||
client.discard_prepared_probe();
|
||||
self.targets[index].sync_live_packets_sent();
|
||||
if self.can_reconnect(index) {
|
||||
return self.begin_close(index, primary_end, None, now);
|
||||
}
|
||||
match cleanup {
|
||||
OpenSessionFailureCleanup::Drain => self.begin_drain(index, primary_end, now),
|
||||
OpenSessionFailureCleanup::Close => self.begin_close(index, primary_end, None, now),
|
||||
@@ -359,6 +434,27 @@ impl ManagedClientTask {
|
||||
let retirement = self.effective_retirement(index);
|
||||
let stopping = self.phase.is_stopping() || retirement.is_some();
|
||||
match self.targets[index].phase_mut() {
|
||||
TargetPhase::Retrying { deadline } => {
|
||||
if stopping {
|
||||
self.finish_target(
|
||||
index,
|
||||
retirement.unwrap_or(ManagedTargetEndReason::Stopped),
|
||||
None,
|
||||
);
|
||||
false
|
||||
} else if now >= *deadline {
|
||||
let target = &self.targets[index].config;
|
||||
let mut client_config = self.config.client.clone();
|
||||
client_config.address_family = target
|
||||
.address_family
|
||||
.unwrap_or(client_config.address_family);
|
||||
client_config.auth = target.auth.resolve(&client_config.auth);
|
||||
self.install_target_phase(index, TargetPhase::Pending { client_config });
|
||||
true
|
||||
} else {
|
||||
false
|
||||
}
|
||||
}
|
||||
TargetPhase::Pending { .. } => {
|
||||
if stopping {
|
||||
self.finish_target(
|
||||
@@ -465,6 +561,16 @@ impl ManagedClientTask {
|
||||
if stopping {
|
||||
client.discard_prepared_probe();
|
||||
self.targets[index].sync_live_packets_sent();
|
||||
if self.config.reconnect_timeout.is_some() {
|
||||
return self.begin_close(
|
||||
index,
|
||||
retirement
|
||||
.clone()
|
||||
.unwrap_or(ManagedTargetEndReason::Stopped),
|
||||
None,
|
||||
now,
|
||||
);
|
||||
}
|
||||
return self.begin_drain(
|
||||
index,
|
||||
retirement
|
||||
@@ -500,6 +606,22 @@ impl ManagedClientTask {
|
||||
self.finish_target(index, ManagedTargetEndReason::PeerClosed, None);
|
||||
return false;
|
||||
}
|
||||
if self.targets[index]
|
||||
.silence_deadline(self.config.reconnect_timeout)
|
||||
.is_some_and(|deadline| deadline <= now)
|
||||
{
|
||||
self.targets[index].sync_live_packets_sent();
|
||||
return self.begin_close(
|
||||
index,
|
||||
ManagedTargetEndReason::Failed(ManagedTargetFailure {
|
||||
phase: ManagedTargetFailurePhase::Receiving,
|
||||
kind: ManagedTargetFailureKind::Timeout,
|
||||
message: Arc::from("session silent while sending probes"),
|
||||
}),
|
||||
None,
|
||||
now,
|
||||
);
|
||||
}
|
||||
let TargetPhase::Active {
|
||||
client, schedule, ..
|
||||
} = self.targets[index].phase_mut()
|
||||
|
||||
@@ -45,7 +45,8 @@ struct UpdatePlan {
|
||||
|
||||
fn synchronously_retireable(phase: &TargetPhase) -> bool {
|
||||
match phase {
|
||||
TargetPhase::Pending { .. }
|
||||
TargetPhase::Retrying { .. }
|
||||
| TargetPhase::Pending { .. }
|
||||
| TargetPhase::Connecting { .. }
|
||||
| TargetPhase::Terminal { .. } => true,
|
||||
TargetPhase::Opening { open, .. } => !open.has_in_flight_work(),
|
||||
@@ -140,6 +141,10 @@ impl ManagedClientTask {
|
||||
membership: TargetMembership::Desired,
|
||||
remote: None,
|
||||
counters: TargetCounters::default(),
|
||||
sent_before_session: 0,
|
||||
retry_delay: std::time::Duration::from_millis(250),
|
||||
silence_since: None,
|
||||
last_reply_at: None,
|
||||
phase: Some(TargetPhase::Pending {
|
||||
client_config: planned.client_config,
|
||||
}),
|
||||
|
||||
Reference in new issue
Block a user