df_hyperliquid_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 a socket no pong or data has arrived on for this long. The venue closes one it has
10    /// sent nothing to for 60 s, and its acks do not count; a ping makes it send a pong.
11    pub ping_after_ms: u64,
12    /// A socket nothing has arrived on for this long, pong included, is dead. The venue would
13    /// have closed it by now; when no close came, the connection is half open.
14    pub silent_ms: u64,
15    /// A socket open this long that heard something worked: the next failure starts the
16    /// reconnect delays over. One dropped sooner keeps backing off.
17    pub steady_ms: u64,
18    /// The first reconnect delay; each failure after doubles it.
19    pub first_retry_ms: u64,
20    /// The longest reconnect delay, before jitter adds up to a quarter.
21    pub last_retry_ms: u64,
22}
23
24impl Timing {
25    /// A 30 s ping, dead after 60 s of nothing (the venue's own idle limit), and reconnects
26    /// 250 ms doubling to 20 s.
27    pub const DEFAULT: Timing = Timing {
28        ping_after_ms: 30_000,
29        silent_ms: 60_000,
30        steady_ms: 30_000,
31        first_retry_ms: 250,
32        last_retry_ms: 20_000,
33    };
34}
35
36impl Default for Timing {
37    fn default() -> Self {
38        Timing::DEFAULT
39    }
40}
41
42/// What a tick asks of the socket's owner.
43#[derive(Debug, Clone, Copy, PartialEq, Eq)]
44pub enum Tick {
45    Wait,
46    /// Send [`crate::live::request::PING`].
47    Ping,
48    /// Nothing has arrived for too long: close the socket, call [`Session::closed`], and
49    /// connect again after the delay it gives.
50    Reconnect,
51}
52
53/// What a closed socket leaves behind.
54#[derive(Debug, Clone, PartialEq, Eq)]
55pub struct Retry {
56    /// Wait this long, then connect again.
57    pub after_ms: u64,
58    /// Posts sent and never answered. Unknown, not failed: an action among them may have
59    /// landed, so none is sent again; read the account to find out.
60    pub unanswered: Vec<u64>,
61}
62
63impl Session {
64    /// What the socket needs at `now`: a ping after [`Timing::ping_after_ms`] with no pong or
65    /// data (once, until one arrives or as long again passes), or a reconnect after
66    /// [`Timing::silent_ms`] of nothing at all.
67    pub fn tick(&mut self, now: u64) -> Tick {
68        let Link::Open { heard, kept, 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((*kept).max(*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, kept, pinged, .. } = self.link else { return None };
83        let ping = kept.max(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 posts 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        let doubled = 1u64.checked_shl(self.failures).unwrap_or(u64::MAX);
101        let step = self.timing.first_retry_ms.saturating_mul(doubled).min(self.timing.last_retry_ms);
102        let jitter = random % (step / 4).saturating_add(1);
103        self.failures = self.failures.saturating_add(1);
104        Retry { after_ms: step.saturating_add(jitter), unanswered: std::mem::take(&mut self.waiting) }
105    }
106}
107
108#[cfg(test)]
109#[path = "tests/timing.rs"]
110mod tests;
df_hyperliquid_core/live/session/timing.rs

df_hyperliquid_core/live/session/
timing.rs
