feat(client): make reconnect exponential backoff configurable
Rust CI / cargo fmt (push) Successful in 17s
Rust CI / cargo clippy (push) Successful in 32s
Rust CI / cargo check (x86_64-unknown-linux-gnu) (push) Successful in 24s
Rust CI / cargo check (x86_64-unknown-linux-musl) (push) Successful in 32s
Rust CI / cargo test (x86_64-unknown-linux-gnu) (push) Successful in 1m6s
Rust CI / cargo doc (push) Successful in 26s
Rust CI / cargo test (x86_64-unknown-linux-musl) (push) Successful in 1m18s
Fuzz Smoke / cargo fuzz smoke (push) Successful in 1m39s
Netem Interop / netem interop (push) Successful in 1m22s
Supply chain / cargo-deny advisories (push) Successful in 19s
Supply chain / actionlint (push) Successful in 4s
Supply chain / cargo-deny bans licenses sources (push) Successful in 15s
Build binaries / Build (arm-musleabi) (push) Successful in 4m52s
Build binaries / Build (aarch64-musl) (push) Successful in 4m59s
Build binaries / Build (arm-musleabihf) (push) Successful in 4m17s
Build binaries / Build (armv7-musleabi) (push) Successful in 4m20s
Build binaries / Build (armv7-musleabihf) (push) Successful in 4m30s
Build binaries / Build (i686-musl) (push) Successful in 4m27s
Build binaries / Build (mips-musl) (push) Successful in 4m14s
Build binaries / Build (mips64-muslabi64) (push) Successful in 4m15s
Build binaries / Build (mips64el-muslabi64) (push) Successful in 4m15s
Build binaries / Build (mipsel-musl) (push) Successful in 4m14s
Build binaries / Build (riscv64gc-musl) (push) Successful in 4m39s
Build binaries / Build (x86_64-musl) (push) Successful in 4m40s
Rust CI / cargo fmt (push) Successful in 17s
Rust CI / cargo clippy (push) Successful in 32s
Rust CI / cargo check (x86_64-unknown-linux-gnu) (push) Successful in 24s
Rust CI / cargo check (x86_64-unknown-linux-musl) (push) Successful in 32s
Rust CI / cargo test (x86_64-unknown-linux-gnu) (push) Successful in 1m6s
Rust CI / cargo doc (push) Successful in 26s
Rust CI / cargo test (x86_64-unknown-linux-musl) (push) Successful in 1m18s
Fuzz Smoke / cargo fuzz smoke (push) Successful in 1m39s
Netem Interop / netem interop (push) Successful in 1m22s
Supply chain / cargo-deny advisories (push) Successful in 19s
Supply chain / actionlint (push) Successful in 4s
Supply chain / cargo-deny bans licenses sources (push) Successful in 15s
Build binaries / Build (arm-musleabi) (push) Successful in 4m52s
Build binaries / Build (aarch64-musl) (push) Successful in 4m59s
Build binaries / Build (arm-musleabihf) (push) Successful in 4m17s
Build binaries / Build (armv7-musleabi) (push) Successful in 4m20s
Build binaries / Build (armv7-musleabihf) (push) Successful in 4m30s
Build binaries / Build (i686-musl) (push) Successful in 4m27s
Build binaries / Build (mips-musl) (push) Successful in 4m14s
Build binaries / Build (mips64-muslabi64) (push) Successful in 4m15s
Build binaries / Build (mips64el-muslabi64) (push) Successful in 4m15s
Build binaries / Build (mipsel-musl) (push) Successful in 4m14s
Build binaries / Build (riscv64gc-musl) (push) Successful in 4m39s
Build binaries / Build (x86_64-musl) (push) Successful in 4m40s
This commit is contained in:
7 files changed
+73
-10
No files matched your search
@@ -72,7 +72,7 @@ pub struct ClientArgs {
|
|||||||
#[arg(
|
#[arg(
|
||||||
long,
|
long,
|
||||||
conflicts_with = "no_test",
|
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."
|
long_help = "Retry failed initial opens and closed, failed, or silent sessions independently per target. Requires --continuous or --duration 0. Unsuccessful connections use exponential backoff (defaults: 250ms initial, 10s maximum), configurable with --reconnect-initial-delay and --reconnect-max-delay. Use --reconnect-timeout to configure silence detection and --no-stats when only measurement events are needed."
|
||||||
)]
|
)]
|
||||||
pub reconnect: bool,
|
pub reconnect: bool,
|
||||||
|
|
||||||
@@ -80,6 +80,14 @@ pub struct ClientArgs {
|
|||||||
#[arg(long, default_value = "10s", value_parser = crate::shared::client::parse_duration, requires = "reconnect")]
|
#[arg(long, default_value = "10s", value_parser = crate::shared::client::parse_duration, requires = "reconnect")]
|
||||||
pub reconnect_timeout: Duration,
|
pub reconnect_timeout: Duration,
|
||||||
|
|
||||||
|
/// Initial retry delay, restored after a valid reply.
|
||||||
|
#[arg(long, default_value = "250ms", value_parser = crate::shared::client::parse_duration, requires = "reconnect")]
|
||||||
|
pub reconnect_initial_delay: Duration,
|
||||||
|
|
||||||
|
/// Maximum retry delay (must be at least twice the initial delay).
|
||||||
|
#[arg(long, default_value = "10s", value_parser = crate::shared::client::parse_duration, requires = "reconnect")]
|
||||||
|
pub reconnect_max_delay: Duration,
|
||||||
|
|
||||||
#[command(flatten)]
|
#[command(flatten)]
|
||||||
pub common: CommonClientArgs,
|
pub common: CommonClientArgs,
|
||||||
|
|
||||||
@@ -132,6 +140,15 @@ impl ClientArgs {
|
|||||||
if self.reconnect && !self.is_continuous() {
|
if self.reconnect && !self.is_continuous() {
|
||||||
return Err("--reconnect requires --continuous or --duration 0".to_owned());
|
return Err("--reconnect requires --continuous or --duration 0".to_owned());
|
||||||
}
|
}
|
||||||
|
if self
|
||||||
|
.reconnect_initial_delay
|
||||||
|
.checked_mul(2)
|
||||||
|
.is_none_or(|minimum| self.reconnect_max_delay < minimum)
|
||||||
|
{
|
||||||
|
return Err(
|
||||||
|
"--reconnect-max-delay must be at least twice --reconnect-initial-delay".to_owned(),
|
||||||
|
);
|
||||||
|
}
|
||||||
if self.targets_stdin && !self.is_continuous() {
|
if self.targets_stdin && !self.is_continuous() {
|
||||||
return Err("--targets-stdin requires --continuous or --duration 0".to_owned());
|
return Err("--targets-stdin requires --continuous or --duration 0".to_owned());
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -78,6 +78,8 @@ pub async fn run_stream(
|
|||||||
{
|
{
|
||||||
let mut config = setup.managed_config();
|
let mut config = setup.managed_config();
|
||||||
config.reconnect_timeout = args.reconnect.then_some(args.reconnect_timeout);
|
config.reconnect_timeout = args.reconnect.then_some(args.reconnect_timeout);
|
||||||
|
config.reconnect_initial_delay = args.reconnect_initial_delay;
|
||||||
|
config.reconnect_max_delay = args.reconnect_max_delay;
|
||||||
config
|
config
|
||||||
},
|
},
|
||||||
setup.managed_targets(),
|
setup.managed_targets(),
|
||||||
@@ -482,6 +484,8 @@ async fn run_stdin_stream(
|
|||||||
{
|
{
|
||||||
let mut config = setup.managed_config();
|
let mut config = setup.managed_config();
|
||||||
config.reconnect_timeout = args.reconnect.then_some(args.reconnect_timeout);
|
config.reconnect_timeout = args.reconnect.then_some(args.reconnect_timeout);
|
||||||
|
config.reconnect_initial_delay = args.reconnect_initial_delay;
|
||||||
|
config.reconnect_max_delay = args.reconnect_max_delay;
|
||||||
config
|
config
|
||||||
},
|
},
|
||||||
setup.managed_targets(),
|
setup.managed_targets(),
|
||||||
|
|||||||
@@ -104,6 +104,12 @@ pub struct ManagedClientConfig {
|
|||||||
/// Retry failed/closed sessions and detect silence while sending. Only valid
|
/// Retry failed/closed sessions and detect silence while sending. Only valid
|
||||||
/// for continuous measurement runs. None (the default) disables reconnection.
|
/// for continuous measurement runs. None (the default) disables reconnection.
|
||||||
pub reconnect_timeout: Option<Duration>,
|
pub reconnect_timeout: Option<Duration>,
|
||||||
|
/// First retry delay, also restored after a valid reply (default: 250 ms).
|
||||||
|
/// Must be positive and at most half of `reconnect_max_delay`.
|
||||||
|
pub reconnect_initial_delay: Duration,
|
||||||
|
/// Cap for per-target exponential backoff with a fixed multiplier of two
|
||||||
|
/// (default: 10 seconds). Must be schedulable from the current instant.
|
||||||
|
pub reconnect_max_delay: Duration,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl Default for ManagedClientConfig {
|
impl Default for ManagedClientConfig {
|
||||||
@@ -118,6 +124,8 @@ impl Default for ManagedClientConfig {
|
|||||||
max_live_target_generations: DEFAULT_MANAGED_MAX_LIVE_TARGET_GENERATIONS,
|
max_live_target_generations: DEFAULT_MANAGED_MAX_LIVE_TARGET_GENERATIONS,
|
||||||
final_drain: DEFAULT_MANAGED_FINAL_DRAIN,
|
final_drain: DEFAULT_MANAGED_FINAL_DRAIN,
|
||||||
reconnect_timeout: None,
|
reconnect_timeout: None,
|
||||||
|
reconnect_initial_delay: Duration::from_millis(250),
|
||||||
|
reconnect_max_delay: Duration::from_secs(10),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -370,6 +378,11 @@ pub struct ManagedTargetFailure {
|
|||||||
pub enum ManagedConfigError {
|
pub enum ManagedConfigError {
|
||||||
#[error("reconnection requires continuous measurement mode and a positive, schedulable silence timeout")]
|
#[error("reconnection requires continuous measurement mode and a positive, schedulable silence timeout")]
|
||||||
InvalidReconnectPolicy,
|
InvalidReconnectPolicy,
|
||||||
|
#[error("reconnection delays must be positive and schedulable, with maximum delay at least twice the initial delay (initial: {initial:?}, maximum: {maximum:?})")]
|
||||||
|
InvalidReconnectDelays {
|
||||||
|
initial: Duration,
|
||||||
|
maximum: Duration,
|
||||||
|
},
|
||||||
#[error("FinishWhenQuiescent requires at least one initial target")]
|
#[error("FinishWhenQuiescent requires at least one initial target")]
|
||||||
EmptyInitialTargets,
|
EmptyInitialTargets,
|
||||||
#[error("duplicate managed target id {id}")]
|
#[error("duplicate managed target id {id}")]
|
||||||
|
|||||||
@@ -367,6 +367,26 @@ fn build_task(
|
|||||||
maximum: tokio::sync::Semaphore::MAX_PERMITS,
|
maximum: tokio::sync::Semaphore::MAX_PERMITS,
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
if config.reconnect_initial_delay.is_zero()
|
||||||
|
|| config.reconnect_max_delay.is_zero()
|
||||||
|
|| config
|
||||||
|
.reconnect_initial_delay
|
||||||
|
.checked_mul(2)
|
||||||
|
.is_none_or(|minimum| config.reconnect_max_delay < minimum)
|
||||||
|
// Tokio represents timer deadlines in u64 milliseconds, reserving the
|
||||||
|
// top two values for state. Leave room for its millisecond rounding.
|
||||||
|
|| config.reconnect_max_delay.as_millis() >= u128::from(u64::MAX - 2)
|
||||||
|
|| config
|
||||||
|
.reconnect_max_delay
|
||||||
|
.checked_add(std::time::Duration::from_millis(1))
|
||||||
|
.and_then(|delay| Instant::now().checked_add(delay))
|
||||||
|
.is_none()
|
||||||
|
{
|
||||||
|
return Err(ManagedConfigError::InvalidReconnectDelays {
|
||||||
|
initial: config.reconnect_initial_delay,
|
||||||
|
maximum: config.reconnect_max_delay,
|
||||||
|
});
|
||||||
|
}
|
||||||
if let Some(timeout) = config.reconnect_timeout {
|
if let Some(timeout) = config.reconnect_timeout {
|
||||||
if timeout.is_zero()
|
if timeout.is_zero()
|
||||||
|| Instant::now().checked_add(timeout).is_none()
|
|| Instant::now().checked_add(timeout).is_none()
|
||||||
@@ -427,7 +447,7 @@ fn build_task(
|
|||||||
remote: None,
|
remote: None,
|
||||||
counters: TargetCounters::default(),
|
counters: TargetCounters::default(),
|
||||||
sent_before_session: 0,
|
sent_before_session: 0,
|
||||||
retry_delay: std::time::Duration::from_millis(250),
|
retry_delay: config.reconnect_initial_delay,
|
||||||
silence_since: None,
|
silence_since: None,
|
||||||
last_reply_at: None,
|
last_reply_at: None,
|
||||||
phase: Some(TargetPhase::Pending { client_config }),
|
phase: Some(TargetPhase::Pending { client_config }),
|
||||||
|
|||||||
@@ -329,7 +329,7 @@ impl ManagedClientTask {
|
|||||||
// A healthy sender with a long interval can remain idle.
|
// A healthy sender with a long interval can remain idle.
|
||||||
self.targets[index].silence_since = None;
|
self.targets[index].silence_since = None;
|
||||||
self.targets[index].last_reply_at = Some(received_at.mono);
|
self.targets[index].last_reply_at = Some(received_at.mono);
|
||||||
self.targets[index].retry_delay = Duration::from_millis(250);
|
self.targets[index].retry_delay = self.config.reconnect_initial_delay;
|
||||||
}
|
}
|
||||||
_ => {}
|
_ => {}
|
||||||
}
|
}
|
||||||
@@ -344,7 +344,7 @@ impl ManagedClientTask {
|
|||||||
fn finish_target(
|
fn finish_target(
|
||||||
&mut self,
|
&mut self,
|
||||||
index: usize,
|
index: usize,
|
||||||
end_reason: ManagedTargetEndReason,
|
mut end_reason: ManagedTargetEndReason,
|
||||||
cleanup_failure: Option<ManagedTargetFailure>,
|
cleanup_failure: Option<ManagedTargetFailure>,
|
||||||
) {
|
) {
|
||||||
if self.can_reconnect(index)
|
if self.can_reconnect(index)
|
||||||
@@ -359,10 +359,17 @@ impl ManagedClientTask {
|
|||||||
target.sent_before_session = target.counters.packets_sent;
|
target.sent_before_session = target.counters.packets_sent;
|
||||||
target.silence_since = None;
|
target.silence_since = None;
|
||||||
target.last_reply_at = None;
|
target.last_reply_at = None;
|
||||||
let deadline = Instant::now() + target.retry_delay;
|
if let Some(deadline) = Instant::now().checked_add(target.retry_delay) {
|
||||||
target.retry_delay = (target.retry_delay * 2).min(Duration::from_secs(10));
|
target.retry_delay = target
|
||||||
self.install_target_phase(index, TargetPhase::Retrying { deadline });
|
.retry_delay
|
||||||
return;
|
.checked_mul(2)
|
||||||
|
.unwrap_or(self.config.reconnect_max_delay)
|
||||||
|
.min(self.config.reconnect_max_delay);
|
||||||
|
self.install_target_phase(index, TargetPhase::Retrying { deadline });
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
// A delay validated at construction may no longer fit after a long run.
|
||||||
|
end_reason = ManagedTargetEndReason::Failed(duration_overflow_failure());
|
||||||
}
|
}
|
||||||
let outcome = Arc::new(self.targets[index].outcome(end_reason, cleanup_failure));
|
let outcome = Arc::new(self.targets[index].outcome(end_reason, cleanup_failure));
|
||||||
// Finalize history before the phase publisher installs Terminal, updates
|
// Finalize history before the phase publisher installs Terminal, updates
|
||||||
|
|||||||
@@ -142,7 +142,7 @@ impl ManagedClientTask {
|
|||||||
remote: None,
|
remote: None,
|
||||||
counters: TargetCounters::default(),
|
counters: TargetCounters::default(),
|
||||||
sent_before_session: 0,
|
sent_before_session: 0,
|
||||||
retry_delay: std::time::Duration::from_millis(250),
|
retry_delay: self.config.reconnect_initial_delay,
|
||||||
silence_since: None,
|
silence_since: None,
|
||||||
last_reply_at: None,
|
last_reply_at: None,
|
||||||
phase: Some(TargetPhase::Pending {
|
phase: Some(TargetPhase::Pending {
|
||||||
|
|||||||
+3
-1
@@ -237,7 +237,9 @@ warnings are disabled, and IPDV columns retain their names with absent values
|
|||||||
(`-` in tables, empty CSV/TSV fields, and JSON `null`). Measurement events and
|
(`-` in tables, empty CSV/TSV fields, and JSON `null`). Measurement events and
|
||||||
protocol-level received statistics remain available.
|
protocol-level received statistics remain available.
|
||||||
|
|
||||||
`--reconnect` enables automatic recovery for individual targets in continuous mode (`--continuous` or `--duration 0`). Failed connections, closed sessions, and unresponsive servers are retried without interrupting other targets. Retries use exponential backoff from 250 ms up to 10 seconds.
|
`--reconnect` enables automatic recovery for individual targets in continuous mode (`--continuous` or `--duration 0`). Failed connections, closed sessions, and unresponsive servers are retried without interrupting other targets. Retries double the delay after each failed attempt, from 250 ms up to 10 seconds by default. A valid reply resets the delay to the initial value; each target backs off independently.
|
||||||
|
|
||||||
|
`--reconnect-initial-delay DURATION` (default: `250ms`) and `--reconnect-max-delay DURATION` (default: `10s`) configure these delays. Both require `--reconnect`, must be positive and schedulable, and the maximum must be at least twice the initial delay. For example, `--reconnect-initial-delay 1s --reconnect-max-delay 5s` retries after `1s`, `2s`, `4s`, `5s`, then `5s` thereafter.
|
||||||
|
|
||||||
`--reconnect-timeout DURATION` controls how long a session can go without valid replies before reconnection is attempted (default: `10s`). The timeout applies only while actively sending probes and should account for the probe interval and expected packet loss.
|
`--reconnect-timeout DURATION` controls how long a session can go without valid replies before reconnection is attempted (default: `10s`). The timeout applies only while actively sending probes and should account for the probe interval and expected packet loss.
|
||||||
|
|
||||||
|
|||||||
Reference in new issue
Block a user