df_hyperliquid_core/live/
session.rs

1//! One connection's state with no socket and no clock: the socket's owner hands in what happened
2//! and when, and sends what comes back. Of the venue's per-IP limits, subscriptions, users and
3//! posts in flight are refused here; connections and message rates are the owner's to keep.
4
5use crate::client::Network;
6use crate::live::frame::{BadFrame, Frame, decode};
7use crate::live::request::{Post, post, subscribe, unsubscribe};
8use crate::live::subscription::{BadSubscription, Subscription};
9
10/// Websocket connections an IP may hold.
11pub const MAX_CONNECTIONS: usize = 10;
12/// New websocket connections an IP may open a minute.
13pub const MAX_NEW_CONNECTIONS_PER_MINUTE: usize = 30;
14/// Subscriptions an IP may hold across its connections.
15pub const MAX_SUBSCRIPTIONS: usize = 1_000;
16/// Distinct users an IP's user subscriptions may name.
17pub const MAX_USERS: usize = 10;
18/// Messages an IP may send a minute across its connections.
19pub const MAX_MESSAGES_PER_MINUTE: usize = 2_000;
20/// Posts an IP may have waiting for an answer across its connections.
21pub const MAX_INFLIGHT_POSTS: usize = 100;
22
23/// What one session may hold. [`Limits::VENUE`] is the whole of an IP's allowance; a session
24/// sharing the IP with others takes a part of it.
25#[derive(Debug, Clone, Copy, PartialEq, Eq)]
26pub struct Limits {
27    pub subscriptions: usize,
28    pub users: usize,
29    pub posts: usize,
30}
31
32impl Limits {
33    pub const VENUE: Limits = Limits { subscriptions: MAX_SUBSCRIPTIONS, users: MAX_USERS, posts: MAX_INFLIGHT_POSTS };
34}
35
36impl Default for Limits {
37    fn default() -> Self {
38        Limits::VENUE
39    }
40}
41
42/// The clock's rules, in milliseconds.
43#[derive(Debug, Clone, Copy, PartialEq, Eq)]
44pub struct Timing {
45    /// Ping a socket nothing has arrived on for this long. The venue closes one it has sent
46    /// nothing to for 60 s; a ping makes it send a pong.
47    pub ping_after_ms: u64,
48    /// A socket nothing has arrived on for this long, pong included, is dead. The venue would
49    /// have closed it by now; when no close came, the connection is half open.
50    pub silent_ms: u64,
51    /// A socket open this long that heard something worked: the next failure starts the
52    /// reconnect delays over. One dropped sooner keeps backing off.
53    pub steady_ms: u64,
54    /// The first reconnect delay; each failure after doubles it.
55    pub first_retry_ms: u64,
56    /// The longest reconnect delay, before jitter adds up to a quarter.
57    pub last_retry_ms: u64,
58}
59
60impl Timing {
61    /// A 30 s ping, dead after 60 s of nothing (the venue's own idle limit), and reconnects
62    /// 250 ms doubling to 20 s.
63    pub const DEFAULT: Timing = Timing {
64        ping_after_ms: 30_000,
65        silent_ms: 60_000,
66        steady_ms: 30_000,
67        first_retry_ms: 250,
68        last_retry_ms: 20_000,
69    };
70}
71
72impl Default for Timing {
73    fn default() -> Self {
74        Timing::DEFAULT
75    }
76}
77
78/// Why nothing was sent.
79#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
80pub enum Refused {
81    #[error("{limit} subscriptions is the most this session holds; the venue allows an IP 1000")]
82    Subscriptions { limit: usize },
83    #[error("{limit} users is the most this session follows; the venue allows an IP 10")]
84    Users { limit: usize },
85    #[error("{limit} posts are already waiting for an answer; the venue allows an IP 100")]
86    Posts { limit: usize },
87    /// A post is never held to send later: an action sent late may land after the moment it
88    /// was meant for, and one sent again may land twice.
89    #[error("the socket is not open, and a post is never held to send later")]
90    Closed,
91    #[error(transparent)]
92    Bad(#[from] BadSubscription),
93}
94
95/// What a tick asks of the socket's owner.
96#[derive(Debug, Clone, Copy, PartialEq, Eq)]
97pub enum Tick {
98    Wait,
99    /// Send [`crate::live::request::PING`].
100    Ping,
101    /// Nothing has arrived for too long: close the socket, call [`Session::closed`], and
102    /// connect again after the delay it gives.
103    Reconnect,
104}
105
106/// What a closed socket leaves behind.
107#[derive(Debug, Clone, PartialEq, Eq)]
108pub struct Retry {
109    /// Wait this long, then connect again.
110    pub after_ms: u64,
111    /// Posts sent and never answered. Unknown, not failed: an action among them may have
112    /// landed, so none is sent again; read the account to find out.
113    pub unanswered: Vec<u64>,
114}
115
116#[derive(Debug, Clone, Copy, PartialEq, Eq)]
117enum Link {
118    Closed,
119    Open {
120        /// When it opened.
121        since: u64,
122        /// When anything last arrived; `since` until something does.
123        heard: u64,
124        /// Whether anything has arrived at all.
125        talked: bool,
126        /// When the last ping went out; `since` until one does.
127        pinged: u64,
128    },
129}
130
131/// One connection to one network's websocket.
132#[derive(Debug, Clone)]
133pub struct Session {
134    network: Network,
135    limits: Limits,
136    timing: Timing,
137    wanted: Vec<Subscription>,
138    link: Link,
139    /// Reconnects since a socket last worked.
140    failures: u32,
141    next_id: u64,
142    waiting: Vec<u64>,
143}
144
145impl Session {
146    /// A closed session on `network`, with the venue's limits and the default timing.
147    #[must_use]
148    pub fn new(network: Network) -> Self {
149        Session {
150            network,
151            limits: Limits::VENUE,
152            timing: Timing::DEFAULT,
153            wanted: Vec::new(),
154            link: Link::Closed,
155            failures: 0,
156            next_id: 1,
157            waiting: Vec::new(),
158        }
159    }
160
161    /// The limits are per IP: sessions sharing one split them.
162    #[must_use]
163    pub const fn with_limits(mut self, limits: Limits) -> Self {
164        self.limits = limits;
165        self
166    }
167
168    #[must_use]
169    pub const fn with_timing(mut self, timing: Timing) -> Self {
170        self.timing = timing;
171        self
172    }
173
174    #[must_use]
175    pub const fn network(&self) -> Network {
176        self.network
177    }
178
179    /// Where to connect: the network's public websocket.
180    #[must_use]
181    pub const fn url(&self) -> &'static str {
182        self.network.ws_url()
183    }
184
185    #[must_use]
186    pub const fn is_open(&self) -> bool {
187        matches!(self.link, Link::Open { .. })
188    }
189
190    /// What this session subscribes to on every socket, in the order it was asked for.
191    #[must_use]
192    pub fn wanted(&self) -> &[Subscription] {
193        &self.wanted
194    }
195
196    /// The distinct users the wanted subscriptions follow.
197    #[must_use]
198    pub fn users(&self) -> usize {
199        let mut users: Vec<_> = self.wanted.iter().filter_map(Subscription::user).collect();
200        users.sort_unstable();
201        users.dedup();
202        users.len()
203    }
204
205    /// The ids of posts sent on this socket and not yet answered.
206    #[must_use]
207    pub fn waiting(&self) -> &[u64] {
208        &self.waiting
209    }
210
211    /// Wants `subscription` from now on, on this socket and every later one. Gives the frame to
212    /// send when the socket is open and it is new; nothing when it was already wanted. Past the
213    /// limits, or when the venue would not take it, it is refused and not kept.
214    pub fn subscribe(&mut self, subscription: Subscription) -> Result<Option<String>, Refused> {
215        if self.wanted.contains(&subscription) {
216            return Ok(None);
217        }
218        subscription.check()?;
219        if self.wanted.len() >= self.limits.subscriptions {
220            return Err(Refused::Subscriptions { limit: self.limits.subscriptions });
221        }
222        if let Some(user) = subscription.user()
223            && !self.wanted.iter().any(|wanted| wanted.user() == Some(user))
224            && self.users() >= self.limits.users
225        {
226            return Err(Refused::Users { limit: self.limits.users });
227        }
228        let frame = self.is_open().then(|| subscribe(&subscription));
229        self.wanted.push(subscription);
230        Ok(frame)
231    }
232
233    /// Stops wanting `subscription`. Gives the frame to send when the socket is open and it was
234    /// wanted.
235    pub fn unsubscribe(&mut self, subscription: &Subscription) -> Option<String> {
236        let at = self.wanted.iter().position(|wanted| wanted == subscription)?;
237        let gone = self.wanted.remove(at);
238        self.is_open().then(|| unsubscribe(&gone))
239    }
240
241    /// Gives the next post's id and its frame. Only on an open socket, and only while fewer
242    /// than the limit are waiting for an answer.
243    pub fn post(&mut self, request: &Post) -> Result<(u64, String), Refused> {
244        if !self.is_open() {
245            return Err(Refused::Closed);
246        }
247        if self.waiting.len() >= self.limits.posts {
248            return Err(Refused::Posts { limit: self.limits.posts });
249        }
250        let id = self.next_id;
251        self.next_id = self.next_id.saturating_add(1);
252        self.waiting.push(id);
253        Ok((id, post(id, request)))
254    }
255
256    /// The socket opened at `now`. Gives the frames that subscribe to everything wanted, in
257    /// order.
258    pub fn opened(&mut self, now: u64) -> Vec<String> {
259        self.link = Link::Open { since: now, heard: now, talked: false, pinged: now };
260        self.wanted.iter().map(subscribe).collect()
261    }
262
263    /// A text frame arrived at `now`. Anything that arrives shows the socket is alive, whether it
264    /// reads or not. A post's answer or refusal stops it waiting.
265    pub fn received(&mut self, now: u64, text: &str) -> Result<Frame, BadFrame> {
266        if let Link::Open { heard, talked, .. } = &mut self.link {
267            *heard = (*heard).max(now);
268            *talked = true;
269        }
270        let frame = decode(text);
271        let answered = match &frame {
272            Ok(Frame::Post(reply)) => Some(reply.id),
273            Ok(Frame::Error(refusal)) => refusal.post_id(),
274            _ => None,
275        };
276        if let Some(id) = answered {
277            self.waiting.retain(|waiting| *waiting != id);
278        }
279        frame
280    }
281
282    /// What the socket needs at `now`: a ping after [`Timing::ping_after_ms`] of quiet (once,
283    /// until something arrives or as long again passes), or a reconnect after
284    /// [`Timing::silent_ms`] of nothing.
285    pub fn tick(&mut self, now: u64) -> Tick {
286        let Link::Open { heard, pinged, .. } = &mut self.link else { return Tick::Wait };
287        if now.saturating_sub(*heard) >= self.timing.silent_ms {
288            return Tick::Reconnect;
289        }
290        if now.saturating_sub((*heard).max(*pinged)) >= self.timing.ping_after_ms {
291            *pinged = now;
292            return Tick::Ping;
293        }
294        Tick::Wait
295    }
296
297    /// When a tick next has something to do; `None` while closed.
298    #[must_use]
299    pub fn due(&self) -> Option<u64> {
300        let Link::Open { heard, pinged, .. } = self.link else { return None };
301        let ping = heard.max(pinged).saturating_add(self.timing.ping_after_ms);
302        let silent = heard.saturating_add(self.timing.silent_ms);
303        Some(ping.min(silent))
304    }
305
306    /// The socket closed at `now`, or a connect failed. Gives how long to wait before the next
307    /// connect, and the posts left unanswered. The wait doubles from
308    /// [`Timing::first_retry_ms`] to [`Timing::last_retry_ms`] and `random` adds up to a quarter
309    /// more, so many clients dropped at once do not return at once. It starts over after a
310    /// socket that held ([`Timing::steady_ms`]). There is no last attempt.
311    pub fn closed(&mut self, now: u64, random: u64) -> Retry {
312        if let Link::Open { since, talked: true, .. } = self.link
313            && now.saturating_sub(since) >= self.timing.steady_ms
314        {
315            self.failures = 0;
316        }
317        self.link = Link::Closed;
318        let doubled = 1u64.checked_shl(self.failures).unwrap_or(u64::MAX);
319        let step = self.timing.first_retry_ms.saturating_mul(doubled).min(self.timing.last_retry_ms);
320        let jitter = random % (step / 4).saturating_add(1);
321        self.failures = self.failures.saturating_add(1);
322        Retry { after_ms: step.saturating_add(jitter), unanswered: std::mem::take(&mut self.waiting) }
323    }
324}
325
326#[cfg(test)]
327#[path = "tests/session.rs"]
328mod tests;
df_hyperliquid_core/live/session.rs

df_hyperliquid_core/live/
session.rs
