subetha_cxc/
path_sensor.rs

1//! Path sensing from the peer's TTL / ECN observations.
2//!
3//! The receiver reads the IP TTL and ECN bits off every datagram (a passive
4//! cmsg, no protocol cost) and echoes them back in a [`PathFrame`]. The
5//! sender feeds that stream here to derive two feed-forward signals the
6//! adaptive controller fuses alongside loss and delay-trend:
7//!
8//!  - **Path shift**: a change in hop count means a router-level path change
9//!    (a re-route, a link failover). It often precedes a throughput change,
10//!    so it pre-arms protection before loss materializes. The signal spikes
11//!    to 1.0 on the change and decays.
12//!  - **ECN-CE rate**: an AQM router marks Congestion-Experienced *before*
13//!    it tail-drops. A rising CE rate is a direct "queue is building" signal
14//!    that, like a rising delay trend, calls for protection ahead of loss.
15//!
16//! The estimator holds no clock and does no I/O - the caller supplies each
17//! `(hop_count, ecn)` observation - so it is deterministic and exhaustively
18//! testable with synthetic traces.
19//!
20//! [`PathFrame`]: crate::control_frame::PathFrame
21
22/// The two-bit ECN codepoint marking Congestion Experienced (RFC 3168). An
23/// AQM router sets this on a packet it would otherwise have to drop.
24pub const ECN_CE: u8 = 0b11;
25
26/// Common initial IP TTL values, smallest first. Hosts start a packet at one
27/// of these and every router decrements by one, so the smallest of these that
28/// is at least the observed TTL is the likely origin, and the difference is
29/// the hop count.
30const INITIAL_TTLS: [u8; 3] = [64, 128, 255];
31
32/// Derive a hop count from an observed TTL: pick the smallest standard
33/// initial TTL not below the observed value, and subtract.
34pub fn hop_count_from_ttl(ttl: u8) -> u8 {
35    for &init in &INITIAL_TTLS {
36        if ttl <= init {
37            return init - ttl;
38        }
39    }
40    0
41}
42
43/// Rolling estimator over the peer's path observations.
44#[derive(Debug, Default)]
45pub struct PathSensor {
46    /// Last hop count seen, to detect a change.
47    last_hop_count: Option<u8>,
48    /// Decaying path-shift signal: 1.0 on a hop-count change, then fading.
49    shift: f32,
50    /// EWMA of the Congestion-Experienced marking rate, 0..=1.
51    ce_rate: f32,
52    /// Last raw `(ttl, ecn, hop_count)` echoed, for diagnostics.
53    last: Option<(u8, u8, u8)>,
54}
55
56impl PathSensor {
57    /// A fresh sensor with no observations.
58    pub fn new() -> Self {
59        Self::default()
60    }
61
62    /// Record one `(hop_count, ecn)` observation echoed by the peer.
63    pub fn observe(&mut self, ttl: u8, ecn: u8, hop_count: u8) {
64        let changed = self.last_hop_count.is_some_and(|p| p != hop_count);
65        self.last_hop_count = Some(hop_count);
66        // A hop-count change spikes the shift signal; an unchanged path lets
67        // it decay back toward zero.
68        if changed {
69            self.shift = 1.0;
70        } else {
71            self.shift *= 0.85;
72        }
73        let ce = if ecn & 0b11 == ECN_CE { 1.0 } else { 0.0 };
74        self.ce_rate += (ce - self.ce_rate) * 0.2;
75        self.last = Some((ttl, ecn, hop_count));
76    }
77
78    /// Path-shift signal, 0..=1: high just after a router-level path change.
79    pub fn path_shift(&self) -> f32 {
80        self.shift
81    }
82
83    /// Congestion-Experienced marking rate, 0..=1: a queue-building signal.
84    pub fn ecn_ce(&self) -> f32 {
85        self.ce_rate
86    }
87
88    /// Last observed `(ttl, ecn, hop_count)`, for diagnostics / telemetry.
89    pub fn last(&self) -> Option<(u8, u8, u8)> {
90        self.last
91    }
92}
93
94#[cfg(test)]
95mod tests {
96    use super::*;
97
98    #[test]
99    fn hop_count_inference_matches_known_initials() {
100        // Linux origin (64) eleven hops away.
101        assert_eq!(hop_count_from_ttl(53), 11);
102        // Windows origin (128) eleven hops away.
103        assert_eq!(hop_count_from_ttl(117), 11);
104        // Router origin (255), one hop.
105        assert_eq!(hop_count_from_ttl(254), 1);
106        // Direct (no decrement) from a 64-init host.
107        assert_eq!(hop_count_from_ttl(64), 0);
108    }
109
110    #[test]
111    fn steady_path_keeps_shift_low() {
112        let mut s = PathSensor::new();
113        for _ in 0..20 {
114            s.observe(53, 0, 11);
115        }
116        assert!(s.path_shift() < 0.05, "steady path -> shift ~0");
117    }
118
119    #[test]
120    fn hop_count_change_spikes_then_decays() {
121        let mut s = PathSensor::new();
122        for _ in 0..10 {
123            s.observe(53, 0, 11);
124        }
125        // A re-route: hop count jumps 11 -> 14.
126        s.observe(50, 0, 14);
127        assert!(s.path_shift() > 0.9, "path shift spikes on hop-count change");
128        // Settling back: the signal decays over subsequent steady samples.
129        for _ in 0..10 {
130            s.observe(50, 0, 14);
131        }
132        assert!(s.path_shift() < 0.2, "shift decays once the path is steady");
133    }
134
135    #[test]
136    fn ce_marking_rate_rises_with_congestion() {
137        let mut s = PathSensor::new();
138        for _ in 0..20 {
139            s.observe(53, 0, 11); // ECT, no congestion
140        }
141        assert!(s.ecn_ce() < 0.05, "no CE -> rate ~0");
142        for _ in 0..20 {
143            s.observe(53, ECN_CE, 11); // congestion experienced
144        }
145        assert!(s.ecn_ce() > 0.8, "sustained CE -> rate climbs toward 1");
146    }
147}
subetha_cxc/path_sensor.rs

subetha_cxc/
path_sensor.rs
