use core::time::Duration;
#[derive(Debug, Clone, Copy)]
pub struct LivenessConfig {
pub min_pto: Duration,
pub path_down_ptos: u32,
pub idle_timeout: Duration,
pub keepalive_interval: Option<Duration>,
}
impl Default for LivenessConfig {
fn default() -> Self {
Self {
min_pto: Duration::from_millis(200),
path_down_ptos: 5,
idle_timeout: Duration::from_secs(30),
keepalive_interval: Some(Duration::from_secs(15)),
}
}
}
impl LivenessConfig {
pub fn for_test() -> Self {
Self {
min_pto: Duration::from_millis(10),
path_down_ptos: 3,
idle_timeout: Duration::from_millis(300),
keepalive_interval: Some(Duration::from_millis(40)),
}
}
fn pto(&self, min_rtt: Duration) -> Duration {
let scaled = min_rtt.saturating_mul(3);
if scaled > self.min_pto {
scaled
} else {
self.min_pto
}
}
fn path_down_after(&self, min_rtt: Duration) -> Duration {
self.pto(min_rtt).saturating_mul(self.path_down_ptos)
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum LivenessVerdict {
Unchanged,
PathDown,
Recovered,
Dead,
}
pub fn liveness_verdict(
silence: Duration,
inflight: u64,
min_rtt: Duration,
in_migrating: bool,
migrating_for: Duration,
cfg: &LivenessConfig,
) -> LivenessVerdict {
if in_migrating {
if silence <= cfg.pto(min_rtt) {
return LivenessVerdict::Recovered;
}
if migrating_for > cfg.idle_timeout {
return LivenessVerdict::Dead;
}
return LivenessVerdict::Unchanged;
}
if inflight > 0 && silence > cfg.path_down_after(min_rtt) {
LivenessVerdict::PathDown
} else {
LivenessVerdict::Unchanged
}
}
pub fn should_send_keepalive(
connected: bool,
inflight: u64,
inbound_silence: Duration,
since_last_keepalive: Duration,
cfg: &LivenessConfig,
) -> bool {
let Some(iv) = cfg.keepalive_interval else {
return false;
};
connected && inflight == 0 && inbound_silence >= iv && since_last_keepalive >= iv
}
#[cfg(test)]
mod tests {
use super::*;
fn cfg() -> LivenessConfig {
LivenessConfig {
min_pto: Duration::from_millis(100),
path_down_ptos: 4,
idle_timeout: Duration::from_secs(1),
keepalive_interval: None,
}
}
#[test]
fn no_outstanding_data_never_declares_path_down() {
let v = liveness_verdict(
Duration::from_secs(60),
0,
Duration::ZERO,
false,
Duration::ZERO,
&cfg(),
);
assert_eq!(v, LivenessVerdict::Unchanged);
}
#[test]
fn outstanding_data_just_below_threshold_is_unchanged() {
let v = liveness_verdict(
Duration::from_millis(399),
1200,
Duration::ZERO,
false,
Duration::ZERO,
&cfg(),
);
assert_eq!(v, LivenessVerdict::Unchanged);
}
#[test]
fn outstanding_data_past_threshold_is_path_down() {
let v = liveness_verdict(
Duration::from_millis(401),
1200,
Duration::ZERO,
false,
Duration::ZERO,
&cfg(),
);
assert_eq!(v, LivenessVerdict::PathDown);
}
#[test]
fn pto_scales_with_min_rtt() {
let below = liveness_verdict(
Duration::from_millis(2000),
800,
Duration::from_millis(200),
false,
Duration::ZERO,
&cfg(),
);
assert_eq!(
below,
LivenessVerdict::Unchanged,
"2000ms < 2400ms threshold"
);
let above = liveness_verdict(
Duration::from_millis(2500),
800,
Duration::from_millis(200),
false,
Duration::ZERO,
&cfg(),
);
assert_eq!(
above,
LivenessVerdict::PathDown,
"2500ms > 2400ms threshold"
);
}
#[test]
fn migrating_and_still_silent_is_unchanged_until_idle_timeout() {
let v = liveness_verdict(
Duration::from_secs(10),
800,
Duration::ZERO,
true,
Duration::from_millis(500),
&cfg(),
);
assert_eq!(v, LivenessVerdict::Unchanged);
}
#[test]
fn migrating_recovers_when_inbound_resumes() {
let v = liveness_verdict(
Duration::from_millis(50),
800,
Duration::ZERO,
true,
Duration::from_millis(500),
&cfg(),
);
assert_eq!(v, LivenessVerdict::Recovered);
}
#[test]
fn migrating_dies_after_idle_timeout() {
let v = liveness_verdict(
Duration::from_secs(10),
800,
Duration::ZERO,
true,
Duration::from_millis(1100),
&cfg(),
);
assert_eq!(v, LivenessVerdict::Dead);
}
#[test]
fn recovery_beats_death_at_the_idle_timeout_edge() {
let v = liveness_verdict(
Duration::from_millis(20),
800,
Duration::ZERO,
true,
Duration::from_secs(5),
&cfg(),
);
assert_eq!(v, LivenessVerdict::Recovered);
}
#[test]
fn for_test_config_is_faster_than_default() {
let t = LivenessConfig::for_test();
let d = LivenessConfig::default();
assert!(t.idle_timeout < d.idle_timeout);
assert!(t.min_pto <= d.min_pto);
assert!(t.path_down_ptos >= 1);
assert!(t.keepalive_interval.unwrap() < d.keepalive_interval.unwrap());
}
#[test]
fn default_config_is_sane() {
let d = LivenessConfig::default();
assert!(d.min_pto > Duration::ZERO);
assert!(d.path_down_ptos >= 1);
assert!(d.idle_timeout > d.min_pto);
let ka = d.keepalive_interval.expect("default keep-alive enabled");
assert!(
ka < d.idle_timeout,
"keep-alive must precede the idle-timeout"
);
assert!(ka > d.min_pto, "keep-alive must be rarer than one PTO");
}
fn ka_cfg() -> LivenessConfig {
LivenessConfig {
keepalive_interval: Some(Duration::from_millis(100)),
..cfg()
}
}
#[test]
fn keepalive_fires_on_an_idle_connected_path() {
assert!(should_send_keepalive(
true,
0,
Duration::from_millis(150),
Duration::from_millis(150),
&ka_cfg(),
));
}
#[test]
fn keepalive_suppressed_when_disabled() {
let cfg = LivenessConfig {
keepalive_interval: None,
..cfg()
};
assert!(!should_send_keepalive(
true,
0,
Duration::from_secs(60),
Duration::from_secs(60),
&cfg,
));
}
#[test]
fn keepalive_suppressed_with_data_in_flight() {
assert!(!should_send_keepalive(
true,
1200,
Duration::from_millis(150),
Duration::from_millis(150),
&ka_cfg(),
));
}
#[test]
fn keepalive_suppressed_when_not_connected() {
assert!(!should_send_keepalive(
false,
0,
Duration::from_millis(150),
Duration::from_millis(150),
&ka_cfg(),
));
}
#[test]
fn keepalive_suppressed_before_the_interval_and_throttled_after_a_recent_ping() {
assert!(!should_send_keepalive(
true,
0,
Duration::from_millis(50),
Duration::from_millis(150),
&ka_cfg(),
));
assert!(!should_send_keepalive(
true,
0,
Duration::from_millis(150),
Duration::from_millis(50),
&ka_cfg(),
));
}
}