obby_client/
timer.rs

1//! Deadline scheduling, with no clock of its own.
2//!
3//! Every instant here is a monotonic millisecond count the host measured itself. Nothing in this
4//! file reads a clock, so a test drives the whole thing by feeding it numbers.
5
6use alloc::collections::BTreeMap;
7use alloc::string::String;
8use alloc::vec::Vec;
9
10/// A moment in time, as the host's two clocks see it.
11///
12/// Monotonic time drives every deadline, because it only ever moves forward; wall clock can jump
13/// when a user resets their system clock or NTP steps it. Wall clock exists only to stamp a message
14/// the server did not stamp itself with `server-time`, so this module never reads `unix_ms` at all.
15#[derive(Debug, Clone, Copy, PartialEq, Eq)]
16#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
17pub struct Now {
18    /// Milliseconds on a clock that never goes backward.
19    pub monotonic_ms: u64,
20    /// Milliseconds since the Unix epoch.
21    pub unix_ms: u64,
22}
23
24/// The reference client's `PING` keepalive interval.
25pub(crate) const PING_KEEPALIVE_MS: u64 = 30_000;
26
27/// The reference client's `PONG` timeout: no reply this long after a `PING` means the link is dead.
28pub(crate) const DEAD_LINK_MS: u64 = 10_000;
29
30/// How long a typing indicator stands before it goes stale.
31///
32/// The `done` that would clear it can be lost, and a client that waits for one shows someone typing
33/// forever, so the indicator expires on its own.
34pub(crate) const TYPING_EXPIRY_MS: u64 = 6_000;
35
36/// One named deadline a host can arm and later collect once it falls due.
37#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
38#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
39pub(crate) enum Deadline {
40    /// Time to send the next `PING` to prove the link is still alive.
41    PingKeepalive,
42    /// No `PONG` answered in time: the link is dead and should be torn down.
43    DeadLink,
44    /// Time to attempt the next reconnect, per [`ReconnectBackoff`].
45    Reconnect,
46    /// A target's typing indicator expires, keyed by whatever the host uses to identify it (for
47    /// instance a channel and nick pair).
48    /// Someone composing a message, keyed by where and who.
49    ///
50    /// Two fields rather than one joined string: any separator we picked could appear in a nick or
51    /// a channel, and two different pairs would then share one deadline.
52    Typing(String, String),
53}
54
55/// A set of named deadlines, armed and drained by monotonic time alone.
56///
57/// A host calls [`Timers::set`] with an absolute instant computed from its own [`Now`], and later
58/// calls [`Timers::expire`] with a fresh `Now` to collect whatever fell due since the last drain.
59#[derive(Debug, Clone, Default)]
60pub(crate) struct Timers {
61    deadlines: BTreeMap<Deadline, u64>,
62}
63
64impl Timers {
65    /// A scheduler with nothing armed.
66    pub(crate) fn new() -> Self {
67        Self::default()
68    }
69
70    /// Arm a deadline for this monotonic instant, replacing any previous one under the same key.
71    pub(crate) fn set(&mut self, deadline: Deadline, at_monotonic_ms: u64) {
72        self.deadlines.insert(deadline, at_monotonic_ms);
73    }
74
75    /// Disarm a deadline, if one was set.
76    pub(crate) fn clear(&mut self, deadline: &Deadline) {
77        self.deadlines.remove(deadline);
78    }
79
80    /// Every deadline that has fallen due by `now`, earliest first, removed from the schedule so a
81    /// second call at the same instant will not return them again.
82    pub(crate) fn expire(&mut self, now: Now) -> Vec<Deadline> {
83        drain_due(&mut self.deadlines, now.monotonic_ms, |at| *at)
84            .into_iter()
85            .map(|(deadline, _)| deadline)
86            .collect()
87    }
88
89    /// The earliest pending deadline, so a host can sleep exactly that long instead of polling.
90    pub(crate) fn next(&self) -> Option<u64> {
91        self.deadlines.values().min().copied()
92    }
93}
94
95/// Exponential reconnect backoff: `base * 2^attempt`, capped, giving up after a bounded number of
96/// attempts.
97///
98/// The core never adds randomness: the same sequence of calls always produces the same delays, which
99/// is what makes it testable by feeding in attempt counts. A host that wants to avoid a reconnect
100/// storm across many clients adds jitter on top of the delay [`ReconnectBackoff::next_delay_ms`]
101/// returns, before arming [`Deadline::Reconnect`] with the result.
102#[derive(Debug, Clone, Copy, PartialEq, Eq)]
103pub(crate) struct ReconnectBackoff {
104    base_ms: u64,
105    cap_ms: u64,
106    max_attempts: u32,
107    attempts: u32,
108}
109
110impl ReconnectBackoff {
111    /// The reference client's starting delay, before any doubling.
112    pub(crate) const DEFAULT_BASE_MS: u64 = 2_000;
113    /// The reference client's ceiling: no delay grows past this.
114    pub(crate) const DEFAULT_CAP_MS: u64 = 300_000;
115    /// The reference client's give-up point.
116    pub(crate) const DEFAULT_MAX_ATTEMPTS: u32 = 100;
117
118    /// A backoff with its own base delay, cap and attempt limit.
119    pub(crate) fn new(base_ms: u64, cap_ms: u64, max_attempts: u32) -> Self {
120        Self {
121            base_ms,
122            cap_ms,
123            max_attempts,
124            attempts: 0,
125        }
126    }
127
128    /// The delay before the next attempt, or `None` once `max_attempts` is exhausted.
129    ///
130    /// Each call advances the attempt counter, so the first call returns `base_ms`, the second
131    /// `base_ms * 2`, and so on up to the cap, then `None` forever until [`Self::reset`].
132    pub(crate) fn next_delay_ms(&mut self) -> Option<u64> {
133        if self.attempts >= self.max_attempts {
134            return None;
135        }
136        // `checked_shl` rather than `<<` because a bounded attempt count can still exceed 63 with a
137        // generous `max_attempts`, and shifting that far is what would otherwise panic
138        let factor = 1u64.checked_shl(self.attempts).unwrap_or(u64::MAX);
139        let delay = self.base_ms.saturating_mul(factor).min(self.cap_ms);
140        self.attempts += 1;
141        Some(delay)
142    }
143
144    /// Start the sequence over, for after a connection succeeds.
145    pub(crate) fn reset(&mut self) {
146        self.attempts = 0;
147    }
148}
149
150impl Default for ReconnectBackoff {
151    /// The reference client's own values: a 2 second base doubling to a 5 minute cap, giving up
152    /// after 100 attempts.
153    fn default() -> Self {
154        Self::new(
155            Self::DEFAULT_BASE_MS,
156            Self::DEFAULT_CAP_MS,
157            Self::DEFAULT_MAX_ATTEMPTS,
158        )
159    }
160}
161
162/// Remove every entry whose deadline has passed, earliest first.
163///
164/// Deadlines are the value in the map rather than the key, so the map's own ordering is by whatever
165/// names the entry and says nothing about when it falls due. Both the timer wheel and the labelled
166/// commands need the same drain, and having it once is what keeps "due" meaning one thing.
167pub(crate) fn drain_due<K, V>(
168    map: &mut alloc::collections::BTreeMap<K, V>,
169    now_ms: u64,
170    deadline_of: impl Fn(&V) -> u64,
171) -> Vec<(K, V)>
172where
173    K: Ord + Clone,
174{
175    let mut due: Vec<(u64, K)> = map
176        .iter()
177        .filter(|(_, value)| deadline_of(value) <= now_ms)
178        .map(|(key, value)| (deadline_of(value), key.clone()))
179        .collect();
180    due.sort_by_key(|(at, _)| *at);
181    due.into_iter()
182        .filter_map(|(_, key)| map.remove(&key).map(|value| (key, value)))
183        .collect()
184}
185
186#[cfg(test)]
187mod tests {
188    use super::*;
189    use alloc::string::ToString;
190    use alloc::vec;
191
192    fn now(monotonic_ms: u64) -> Now {
193        Now {
194            monotonic_ms,
195            unix_ms: 0,
196        }
197    }
198
199    #[test]
200    fn a_deadline_fires_exactly_at_its_instant_and_not_before() {
201        let mut timers = Timers::new();
202        timers.set(Deadline::PingKeepalive, 1_000);
203        assert!(timers.expire(now(999)).is_empty());
204        assert_eq!(timers.expire(now(1_000)), vec![Deadline::PingKeepalive]);
205    }
206
207    #[test]
208    fn several_deadlines_expire_in_one_call_in_order() {
209        let mut timers = Timers::new();
210        timers.set(Deadline::DeadLink, 300);
211        timers.set(Deadline::PingKeepalive, 100);
212        timers.set(Deadline::Reconnect, 200);
213        let due = timers.expire(now(1_000));
214        assert_eq!(
215            due,
216            vec![
217                Deadline::PingKeepalive,
218                Deadline::Reconnect,
219                Deadline::DeadLink
220            ]
221        );
222    }
223
224    #[test]
225    fn expiring_drains_a_deadline_so_it_does_not_fire_twice() {
226        let mut timers = Timers::new();
227        timers.set(Deadline::DeadLink, 100);
228        assert_eq!(timers.expire(now(200)), vec![Deadline::DeadLink]);
229        assert!(timers.expire(now(200)).is_empty());
230    }
231
232    #[test]
233    fn next_returns_the_earliest_pending_deadline() {
234        let mut timers = Timers::new();
235        assert_eq!(timers.next(), None);
236        timers.set(Deadline::DeadLink, 500);
237        timers.set(Deadline::PingKeepalive, 200);
238        assert_eq!(timers.next(), Some(200));
239    }
240
241    #[test]
242    fn clearing_a_deadline_disarms_it() {
243        let mut timers = Timers::new();
244        timers.set(Deadline::PingKeepalive, 100);
245        timers.clear(&Deadline::PingKeepalive);
246        assert!(timers.expire(now(1_000)).is_empty());
247        assert_eq!(timers.next(), None);
248    }
249
250    #[test]
251    fn per_key_typing_deadlines_are_independent() {
252        let mut timers = Timers::new();
253        timers.set(Deadline::Typing("#a".to_string(), "alice".to_string()), 100);
254        timers.set(Deadline::Typing("#a".to_string(), "bob".to_string()), 200);
255        assert_eq!(
256            timers.expire(now(100)),
257            vec![Deadline::Typing("#a".to_string(), "alice".to_string())]
258        );
259        assert_eq!(timers.next(), Some(200));
260    }
261
262    #[test]
263    fn reconnect_backoff_doubles_then_caps_then_gives_up() {
264        let mut backoff = ReconnectBackoff::new(1_000, 5_000, 4);
265        assert_eq!(backoff.next_delay_ms(), Some(1_000));
266        assert_eq!(backoff.next_delay_ms(), Some(2_000));
267        assert_eq!(backoff.next_delay_ms(), Some(4_000));
268        assert_eq!(
269            backoff.next_delay_ms(),
270            Some(5_000),
271            "8000 would exceed the 5000 cap"
272        );
273        assert_eq!(
274            backoff.next_delay_ms(),
275            None,
276            "the fifth attempt exceeds max_attempts"
277        );
278    }
279
280    #[test]
281    fn reconnect_backoff_resets_back_to_the_base_delay() {
282        let mut backoff = ReconnectBackoff::new(1_000, 5_000, 4);
283        assert_eq!(backoff.next_delay_ms(), Some(1_000));
284        assert_eq!(backoff.next_delay_ms(), Some(2_000));
285        backoff.reset();
286        assert_eq!(
287            backoff.next_delay_ms(),
288            Some(1_000),
289            "a connection that succeeded means the next failure starts over"
290        );
291    }
292
293    #[test]
294    fn default_reconnect_backoff_matches_the_reference_client() {
295        let mut backoff = ReconnectBackoff::default();
296        assert_eq!(backoff.next_delay_ms(), Some(2_000));
297        assert_eq!(backoff.next_delay_ms(), Some(4_000));
298    }
299}
obby_client/timer.rs

obby_client/
timer.rs
