#![allow(clippy::items_after_statements)]
#![allow(clippy::too_many_lines)]
use std::net::SocketAddr;
use std::time::{Duration, Instant};
use super::testfix::*;
use super::timers::TimerKind;
use crate::constants::{
AEAD_TAG_LEN, AMPLIFICATION_FACTOR, DATA_HEADER_LEN, DEAD_TIMEOUT, FRAME_ACK, FRAME_PADDING,
};
use crate::error::ConnectionLost;
const PKT_OVERHEAD: u64 = (DATA_HEADER_LEN + AEAD_TAG_LEN) as u64;
const CHALLENGE_COST: u64 = 1 + 8;
const WARM_RTT: Duration = Duration::from_millis(20);
fn c_addr() -> SocketAddr {
v4(9, 41_000)
}
fn ack_frame(largest: u64) -> Vec<u8> {
let mut out = Vec::new();
put(&mut out, FRAME_ACK);
put(&mut out, largest);
put(&mut out, 0); put(&mut out, 0); put(&mut out, 0); out
}
fn room(s: &Solo) -> Option<u64> {
s.conn
.amplification_budget()
.map(|(sent, recv)| (AMPLIFICATION_FACTOR * recv).saturating_sub(sent))
}
fn warm(s: &mut Solo, start: Instant) -> (Duration, Instant) {
let r = s.conn.open(crate::core::Dir::Uni).expect("§16.4: open()");
s.conn.write(start, r, &[7u8; 200]).expect("write");
s.conn.flush(start);
let d = drain(&mut s.conn);
assert_eq!(
d.transmits().len(),
1,
"fixture: one packet, so the ACK below samples a known round trip"
);
let ack_at = start + WARM_RTT;
let _ = s.deliver_from(ack_at, a_addr(), &ack_frame(0));
assert_eq!(
s.conn.smoothed_rtt(),
WARM_RTT,
"fixture: the first sample *is* the smoothed RTT (§13.1)"
);
assert_eq!(
s.conn.timer(TimerKind::Pto),
None,
"§13.3's first precondition: the ACK emptied the sent map, so no \
`Pto` is armed — an idle connection must not self-sustain a probe train"
);
let anchor = ack_at + Duration::from_millis(1);
s.conn.write(anchor, r, &[7u8; 200]).expect("write");
s.conn.flush(anchor);
let _ = drain(&mut s.conn);
let base = s
.conn
.timer(TimerKind::Pto)
.expect("§13.3: a non-empty sent map on a validated address arms `Pto`")
- anchor;
assert!(
base > Duration::ZERO,
"fixture: the base interval must be positive"
);
(base, anchor)
}
fn fire_pto(s: &mut Solo, what: &str) -> Instant {
let at = s
.conn
.timer(TimerKind::Pto)
.unwrap_or_else(|| panic!("fixture: no `Pto` armed while {what}"));
s.conn.handle_timeout(at);
let d = drain(&mut s.conn);
assert_eq!(
d.transmits().len(),
1,
"§13.4: a firing PTO sends **one** ack-eliciting packet ({what})"
);
at
}
fn roam_and_starve(s: &mut Solo, at: Instant) {
roam_and_starve_at(s, at, at);
}
fn roam_and_starve_at(s: &mut Solo, recv_at: Instant, send_at: Instant) {
let before = s.conn.remote_address();
let d = s.deliver_from(recv_at, c_addr(), &[]);
assert_eq!(
s.conn.remote_address(),
Some(c_addr()),
"fixture: the peer moved (§7.3); it was at {before:?}"
);
assert!(
d.transmits().is_empty(),
"fixture: the roam itself emits nothing, so the whole 90-byte \
budget is this helper's to spend"
);
let space = room(s).expect("§13.6: a roam zeroes the budget — the address is unvalidated");
let payload = space - PKT_OVERHEAD - 1 - 1 - CHALLENGE_COST;
assert!(
payload < 64,
"fixture: the length varint above is one byte only below 64; got {payload}"
);
s.conn
.send_datagram(send_at, &ramp(0, payload as usize))
.expect("§11: a sub-maximum datagram is accepted");
let d = drain(&mut s.conn);
assert_eq!(
d.transmits().len(),
1,
"fixture: the shaping datagram is sized to fit, so it must leave whole"
);
assert_eq!(
d.transmits()[0].data.len() as u64,
space,
"fixture: the shaping packet must be *exactly* the room to be spent, \
or every assertion calibrated against a closed budget is calibrated \
against the wrong number"
);
assert_eq!(
room(s),
Some(0),
"fixture: the point of the helper — §7.3 now admits nothing at all"
);
}
#[derive(Debug)]
enum Drive {
Died(Instant, ConnectionLost),
Quiet { steps: usize, next: Option<Instant> },
}
fn drive(s: &mut Solo, from: Instant, horizon: Instant) -> Drive {
const MAX_STEPS: usize = 64;
let mut now = from;
for steps in 0..MAX_STEPS {
let d = drain(&mut s.conn);
if let Some(lost) = d.closed() {
return Drive::Died(now, lost);
}
match d.deadline {
None => return Drive::Quiet { steps, next: None },
Some(next) if next > horizon => {
return Drive::Quiet {
steps,
next: Some(next),
};
}
Some(next) => {
now = now.max(next);
s.conn.handle_timeout(now);
}
}
}
panic!(
"the driver loop did not leave {horizon:?} in {MAX_STEPS} steps, and \
is at {now:?}: the core is re-announcing a deadline the loop has \
already consumed. This is ruling 249's livelock — §16.3's one \
`!Send` actor, shared by every connection on the endpoint, spins \
at 100 % of a core here."
);
}
#[test]
fn a_starved_probe_train_announces_liveness_not_a_deadline_in_the_past() {
let start = t0();
let mut s = Solo::installed_at(start);
let (_base, _anchor) = warm(&mut s, start);
let mut last = start;
for i in 0..8 {
last = fire_pto(&mut s, &format!("saturating the train, firing {i}"));
}
let roam_at = last + Duration::from_millis(1);
roam_and_starve(&mut s, roam_at);
assert!(
s.conn.bytes_in_flight() > 0,
"fixture: §13.3's *first* precondition holds — the sent map is not \
empty, so a disarmed `Pto` here is the budget's doing and not the \
empty-map rule's"
);
assert_eq!(room(&s), Some(0), "fixture: the budget admits nothing");
assert_eq!(
s.conn.timer(TimerKind::Keepalive),
None,
"fixture: slice 7b's F1 gate already holds the keepalive at zero \
room (`tests_livelock.rs`), which is why `Liveness` is the next \
armed timer below rather than the beacon"
);
let t = roam_at + DEAD_TIMEOUT / 2;
s.conn.handle_timeout(t);
let d = drain(&mut s.conn);
assert!(
d.transmits().is_empty(),
"fixture: §7.3 admits nothing at room 0, so nothing left — if a \
probe had gone out, this test would prove nothing about the gate"
);
let announced = d.deadline.expect(
"§13.3: the `Timeout` falls to the next armed timer — `Liveness` at \
the latest. `None` here is the immortal-connection collapse",
);
assert!(
announced > t,
"ruling 249: the core announced {announced:?}, at or before the \
instant it was just handed ({t:?}). `sleep_until` completes \
immediately, `handle_timeout` re-fires, and §16.3's shared driver \
spins until `DEAD_TIMEOUT`"
);
assert_eq!(
s.conn.timer(TimerKind::Pto),
None,
"§13.3, amended: *the `Pto` timer is armed only while … §7.3's \
amplification budget admits a probe datagram*"
);
assert_eq!(
Some(announced),
s.conn.timer(TimerKind::Liveness),
"§13.3: the announced `Timeout` is the next armed timer, which here \
is §7.4's death clock"
);
assert_eq!(
announced,
roam_at + DEAD_TIMEOUT,
"§7.4: and the death clock is anchored on the last authenticated \
**receive**, which was the roam"
);
}
#[test]
fn the_starved_connection_still_dies_at_the_dead_timeout() {
let start = t0();
let mut s = Solo::installed_at(start);
let (_, _) = warm(&mut s, start);
let last = fire_pto(&mut s, "arming a backed-off train");
let roam_at = last + Duration::from_millis(1);
roam_and_starve_at(&mut s, roam_at, roam_at + Duration::from_millis(7));
assert!(
s.conn.bytes_in_flight() > 0,
"fixture: the sent map is not empty"
);
match drive(
&mut s,
roam_at,
roam_at + DEAD_TIMEOUT + Duration::from_secs(1),
) {
Drive::Died(at, lost) => {
assert!(
matches!(lost, ConnectionLost::TimedOut),
"§7.4: a blockaded connection dies of the death clock, not of \
anything else; got {lost:?}"
);
assert_eq!(
at,
roam_at + DEAD_TIMEOUT,
"§7.4: `DEAD_TIMEOUT` after the last authenticated **receive** \
— not after the last send, which was 7 ms later"
);
}
other => panic!(
"§13.4: *the session still dies at `DEAD_TIMEOUT` as §7.3 \
intends* — the loop instead reported {other:?}"
),
}
}
#[test]
fn the_refund_re_announces_the_probe_deadline_and_fires_a_probe() {
let start = t0();
let mut s = Solo::installed_at(start);
let (base, _) = warm(&mut s, start);
let last = fire_pto(&mut s, "arming a backed-off train");
let roam_at = last + Duration::from_millis(1);
roam_and_starve(&mut s, roam_at);
let frozen = roam_at + base * 2;
let t = roam_at + DEAD_TIMEOUT / 2;
s.conn.handle_timeout(t);
let _ = drain(&mut s.conn);
let in_flight = s.conn.bytes_in_flight();
let d = s.deliver_from(t, c_addr(), &[FRAME_PADDING as u8; 600]);
assert!(
d.closed().is_none(),
"fixture: the connection is alive at the refund"
);
let refunded = room(&s).expect("still unvalidated: no challenge has been answered");
assert!(
refunded > PKT_OVERHEAD + CHALLENGE_COST,
"fixture: the refund must admit a probe datagram; room is {refunded}"
);
assert_eq!(
s.conn.timer(TimerKind::Pto),
Some(frozen),
"§13.3: the deadline is announced again at the receive that refunds \
the budget — and **where it was left**, because the sent map, \
`pto_count` and the anchor were untouched throughout"
);
let fire_at = frozen.max(t);
s.conn.handle_timeout(fire_at);
let d = drain(&mut s.conn);
assert!(
!d.transmits().is_empty(),
"§13.4: the re-armed firing sends one ack-eliciting packet"
);
let probe = s
.packets(&d)
.into_iter()
.find(|frames| frames.iter().any(|f| matches!(f, Wire::PathChallenge(_))));
assert!(
probe.is_some(),
"§13.4/ruling 221: on an unvalidated address the probe carries \
`PATH_CHALLENGE` — *the only mechanism by which a lost challenge is \
asked again*. Got {:?}",
s.packets(&d)
);
assert!(
s.conn.bytes_in_flight() > in_flight,
"§13.5: and the probe is in the sent map — a probe that leaves \
untracked is not a probe"
);
}
#[test]
fn the_backoff_does_not_climb_through_a_closed_budget() {
let start = t0();
let mut s = Solo::installed_at(start);
let (base, _) = warm(&mut s, start);
let last = fire_pto(&mut s, "taking `pto_count` to exactly 1");
let roam_at = last + Duration::from_millis(1);
roam_and_starve(&mut s, roam_at);
let horizon = roam_at + DEAD_TIMEOUT / 2;
match drive(&mut s, roam_at, horizon) {
Drive::Quiet { steps, next } => assert_eq!(
next,
Some(roam_at + DEAD_TIMEOUT),
"§13.3: with the budget closed the only thing left to wait for is \
§7.4's death clock; the loop consumed {steps} deadline(s) first"
),
other => panic!("the connection must survive the blockade here: {other:?}"),
}
let _ = s.deliver_from(horizon, c_addr(), &[FRAME_PADDING as u8; 600]);
assert_eq!(
s.conn.timer(TimerKind::Pto),
Some(roam_at + base * 2),
"ruling 249: `pto_count` was 1 when the budget closed and is 1 when \
it opens — {:?} of blockade added no rungs to the ladder",
horizon - roam_at
);
}
#[test]
fn the_announced_probe_cadence_caps_at_eight_times_the_base() {
let start = t0();
let mut s = Solo::installed_at(start);
let (base, anchor) = warm(&mut s, start);
let mut prev = anchor;
let mut rungs = Vec::new();
for i in 0..8 {
let at = fire_pto(&mut s, &format!("walking the ladder, firing {i}"));
rungs.push(at - prev);
prev = at;
}
assert_eq!(
rungs,
vec![
base,
base * 2,
base * 4,
base * 8,
base * 8,
base * 8,
base * 8,
base * 8,
],
"§13.3: `2^pto_count` capped at `PTO_BACKOFF_CAP` = 2³ — the ladder \
doubles three times and then holds. Base is {base:?}"
);
assert!(
rungs.iter().any(|r| *r != rungs[0]),
"a build that never backs off announces the same interval forever"
);
assert_eq!(
*rungs.iter().max().expect("eight rungs"),
base * 8,
"§13.3's envelope: *the probe cadence never thins beyond 8 × PTO*"
);
assert_eq!(
rungs[rungs.len() - 1],
rungs[rungs.len() - 2],
"the cap is **reached**: the last two rungs are the same, so this is \
a ladder that arrived at its ceiling and not one still climbing"
);
}