df_lighter_core/live/session/
timing.rs

1//! The session's clock: its rules in milliseconds ([`Timing`]), what a tick asks of the socket's
2//! owner, and how long a closed socket waits before it connects again.
3
4use crate::live::session::{Link, Session};
5
6/// The clock's rules, in milliseconds.
7#[derive(Debug, Clone, Copy, PartialEq, Eq)]
8pub struct Timing {
9    /// Ping this long after the last ping, or the open. The venue closes a socket it has had no
10    /// frame from for two minutes, however much it sends itself; the pong also shows the socket
11    /// still answers.
12    pub ping_after_ms: u64,
13    /// A socket nothing has arrived on for this long, not even the pong to a ping, is dead. When
14    /// no close came, the connection is half open.
15    pub silent_ms: u64,
16    /// A socket open this long that heard something worked: the next failure starts the
17    /// reconnect delays over. One dropped sooner keeps backing off.
18    pub steady_ms: u64,
19    /// The first reconnect delay; each failure after doubles it.
20    pub first_retry_ms: u64,
21    /// The longest reconnect delay, before jitter adds up to a quarter.
22    pub last_retry_ms: u64,
23}
24
25impl Timing {
26    /// A 30 s ping, a quarter of the venue's idle limit; dead after 60 s of nothing; reconnects
27    /// 250 ms doubling to 20 s.
28    pub const DEFAULT: Timing = Timing {
29        ping_after_ms: 30_000,
30        silent_ms: 60_000,
31        steady_ms: 30_000,
32        first_retry_ms: 250,
33        last_retry_ms: 20_000,
34    };
35}
36
37impl Default for Timing {
38    fn default() -> Self {
39        Timing::DEFAULT
40    }
41}
42
43/// What a tick asks of the socket's owner.
44#[derive(Debug, Clone, Copy, PartialEq, Eq)]
45pub enum Tick {
46    Wait,
47    /// Send [`crate::live::PING`].
48    Ping,
49    /// Nothing has arrived for too long: close the socket, call [`Session::closed`], and
50    /// connect again after the delay it gives.
51    Reconnect,
52}
53
54/// What a closed socket leaves behind.
55#[derive(Debug, Clone, PartialEq, Eq)]
56pub struct Retry {
57    /// Wait this long, then connect again.
58    pub after_ms: u64,
59    /// Transactions sent and never answered, by id. Unknown, not failed: any may have landed, so
60    /// none is sent again; `account_tx` or a read of the transaction says.
61    pub unanswered: Vec<u64>,
62}
63
64impl Session {
65    /// What the socket needs at `now`: a ping every [`Timing::ping_after_ms`], busy or not, or a
66    /// reconnect after [`Timing::silent_ms`] of nothing at all.
67    pub fn tick(&mut self, now: u64) -> Tick {
68        let Link::Open { heard, pinged, .. } = &mut self.link else { return Tick::Wait };
69        if now.saturating_sub(*heard) >= self.timing.silent_ms {
70            return Tick::Reconnect;
71        }
72        if now.saturating_sub(*pinged) >= self.timing.ping_after_ms {
73            *pinged = now;
74            return Tick::Ping;
75        }
76        Tick::Wait
77    }
78
79    /// When a tick next has something to do; `None` while closed.
80    #[must_use]
81    pub fn due(&self) -> Option<u64> {
82        let Link::Open { heard, pinged, .. } = self.link else { return None };
83        let ping = pinged.saturating_add(self.timing.ping_after_ms);
84        let silent = heard.saturating_add(self.timing.silent_ms);
85        Some(ping.min(silent))
86    }
87
88    /// The socket closed at `now`, or a connect failed. Gives how long to wait before the next
89    /// connect, and the transactions left unanswered. The wait doubles from
90    /// [`Timing::first_retry_ms`] to [`Timing::last_retry_ms`] and `random` adds up to a quarter
91    /// more, so many clients dropped at once do not return at once. It starts over after a
92    /// socket that held ([`Timing::steady_ms`]). There is no last attempt.
93    pub fn closed(&mut self, now: u64, random: u64) -> Retry {
94        if let Link::Open { since, talked: true, .. } = self.link
95            && now.saturating_sub(since) >= self.timing.steady_ms
96        {
97            self.failures = 0;
98        }
99        self.link = Link::Closed;
100        self.chains.clear();
101        self.owed.clear();
102        let doubled = 1u64.checked_shl(self.failures).unwrap_or(u64::MAX);
103        let step = self.timing.first_retry_ms.saturating_mul(doubled).min(self.timing.last_retry_ms);
104        let jitter = random % (step / 4).saturating_add(1);
105        self.failures = self.failures.saturating_add(1);
106        Retry { after_ms: step.saturating_add(jitter), unanswered: std::mem::take(&mut self.waiting) }
107    }
108}
109
110#[cfg(test)]
111#[path = "tests/timing.rs"]
112mod tests;
df_lighter_core/live/session/timing.rs

df_lighter_core/live/session/
timing.rs
