helius_stream/
health.rs

1use std::time::{Duration, Instant};
2
3/// Tracks slot continuity, gaps, and freshness of incoming updates.
4#[derive(Debug)]
5pub struct StreamHealth {
6    last_slot: u64,
7    last_update: Instant,
8    max_observed_gap: u64,
9    consecutive_clean: u64,
10    total_updates: u64,
11    total_gaps: u64,
12}
13
14impl StreamHealth {
15    pub fn new() -> Self {
16        Self {
17            last_slot: 0,
18            last_update: Instant::now(),
19            max_observed_gap: 0,
20            consecutive_clean: 0,
21            total_updates: 0,
22            total_gaps: 0,
23        }
24    }
25
26    /// Returns true if this update revealed a slot gap.
27    pub fn record_update(&mut self, slot: u64) -> bool {
28        let gap = if self.last_slot > 0 && slot > self.last_slot + 1 {
29            let g = slot - self.last_slot - 1;
30            self.max_observed_gap = self.max_observed_gap.max(g);
31            self.total_gaps += 1;
32            self.consecutive_clean = 0;
33            true
34        } else {
35            self.consecutive_clean += 1;
36            false
37        };
38        self.last_slot = self.last_slot.max(slot);
39        self.last_update = Instant::now();
40        self.total_updates += 1;
41        gap
42    }
43
44    pub fn is_stale(&self, max_age: Duration) -> bool {
45        self.last_update.elapsed() > max_age
46    }
47
48    pub fn last_slot(&self) -> u64 { self.last_slot }
49    pub fn total_gaps(&self) -> u64 { self.total_gaps }
50    pub fn total_updates(&self) -> u64 { self.total_updates }
51    pub fn max_observed_gap(&self) -> u64 { self.max_observed_gap }
52    pub fn consecutive_clean(&self) -> u64 { self.consecutive_clean }
53
54    pub fn gap_rate(&self) -> f64 {
55        if self.total_updates == 0 { return 0.0; }
56        self.total_gaps as f64 / self.total_updates as f64
57    }
58
59    pub fn last_update(&self) -> Instant { self.last_update }
60}
61
62impl Default for StreamHealth {
63    fn default() -> Self { Self::new() }
64}
65
66/// Exponential backoff with cap, for reconnect attempts.
67#[derive(Debug, Clone)]
68pub struct ReconnectPolicy {
69    attempt: u32,
70    base_ms: u64,
71    max_ms: u64,
72}
73
74impl ReconnectPolicy {
75    pub fn new() -> Self {
76        Self { attempt: 0, base_ms: 100, max_ms: 10_000 }
77    }
78
79    pub fn with_bounds(base_ms: u64, max_ms: u64) -> Self {
80        Self { attempt: 0, base_ms, max_ms }
81    }
82
83    pub fn next_delay_ms(&mut self) -> u64 {
84        let cap = self.max_ms.min(self.base_ms * (1u64 << self.attempt.min(10)));
85        self.attempt += 1;
86        cap / 2
87    }
88
89    pub fn reset(&mut self) { self.attempt = 0; }
90    pub fn attempt(&self) -> u32 { self.attempt }
91}
92
93impl Default for ReconnectPolicy {
94    fn default() -> Self { Self::new() }
95}
96
97#[cfg(test)]
98mod tests {
99    use super::*;
100
101    #[test]
102    fn clean_stream_no_gap() {
103        let mut h = StreamHealth::new();
104        for s in 100..110u64 { assert!(!h.record_update(s)); }
105        assert_eq!(h.total_gaps(), 0);
106    }
107
108    #[test]
109    fn gap_detected_on_slot_skip() {
110        let mut h = StreamHealth::new();
111        h.record_update(100);
112        assert!(h.record_update(105));
113        assert_eq!(h.max_observed_gap, 4);
114    }
115
116    #[test]
117    fn stale_with_zero_threshold() {
118        let h = StreamHealth::new();
119        assert!(h.is_stale(Duration::from_millis(0)));
120    }
121
122    #[test]
123    fn backoff_grows_and_caps() {
124        let mut r = ReconnectPolicy::new();
125        let d0 = r.next_delay_ms();
126        let d1 = r.next_delay_ms();
127        assert!(d1 >= d0);
128        for _ in 0..30 { r.next_delay_ms(); }
129        assert!(r.next_delay_ms() <= r.max_ms);
130    }
131
132    #[test]
133    fn backoff_resets() {
134        let mut r = ReconnectPolicy::new();
135        for _ in 0..6 { r.next_delay_ms(); }
136        let high = r.next_delay_ms();
137        r.reset();
138        assert!(r.next_delay_ms() < high);
139    }
140
141    #[test]
142    fn gap_rate_bounded() {
143        let mut h = StreamHealth::new();
144        for i in 0..10u64 { h.record_update(i * 2); }
145        assert!(h.gap_rate() > 0.0 && h.gap_rate() < 1.0);
146    }
147}
helius_stream/health.rs

helius_stream/
health.rs
