df_lighter_core/live/session/
mod.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. It keeps what is wanted and subscribes it on every
3//! socket, with the account's auth token where a channel takes it; checks each order book's nonce
4//! chain and, on a gap, owes the venue the resubscribe that brings a fresh snapshot; answers the
5//! venue's pings; and sends signed transactions, matching each answer to its id. Of the venue's
6//! limits, a connection's subscriptions and an IP's accounts are refused here; connections and
7//! message rates are the owner's to keep.
8//!
9//! - [`limits`]: the venue's limits and why a session refuses.
10//! - [`timing`]: the ping, the silence that ends a socket, the reconnect delays.
11
12pub mod limits;
13pub mod timing;
14
15pub use limits::{
16    Limits, MAX_ACCOUNTS, MAX_CONNECTIONS, MAX_INFLIGHT_MESSAGES, MAX_MESSAGES_PER_MINUTE, MAX_SUBSCRIPTIONS,
17    MAX_SUBSCRIPTIONS_PER_CONNECTION, Refused,
18};
19pub use timing::{Retry, Tick, Timing};
20
21use std::collections::BTreeMap;
22use std::fmt;
23
24use crate::client::Network;
25use crate::live::{
26    BadFrame, Data, Frame, Gap, PONG, Subscription, decode, request::batch_fits, resubscribe, send_tx, send_tx_batch,
27    subscribe, unsubscribe,
28};
29use crate::utils::MarketId;
30use crate::utils::sign::Signed;
31
32#[derive(Debug, Clone, Copy, PartialEq, Eq)]
33enum Link {
34    Closed,
35    Open {
36        /// When it opened.
37        since: u64,
38        /// When anything last arrived; `since` until something does.
39        heard: u64,
40        /// Whether anything has arrived at all.
41        talked: bool,
42        /// When the last ping went out; `since` until one does.
43        pinged: u64,
44    },
45}
46
47/// The account's auth token: shown to no one, in a log or otherwise.
48#[derive(Clone, PartialEq, Eq)]
49struct Token(String);
50
51impl fmt::Debug for Token {
52    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
53        f.write_str("Token(..)")
54    }
55}
56
57/// One connection to one network's websocket.
58#[derive(Debug, Clone)]
59pub struct Session {
60    network: Network,
61    readonly: bool,
62    limits: Limits,
63    timing: Timing,
64    token: Option<Token>,
65    wanted: Vec<Subscription>,
66    link: Link,
67    /// Reconnects since a socket last worked.
68    failures: u32,
69    next_id: u64,
70    waiting: Vec<u64>,
71    /// Each book's last nonce on this socket; `None` while a fresh snapshot is awaited.
72    chains: BTreeMap<MarketId, Option<u64>>,
73    /// Frames the venue is owed: a pong, a resubscribe after a gap.
74    owed: Vec<String>,
75}
76
77impl Session {
78    /// A closed session on `network`'s socket, with the venue's limits and the default timing.
79    #[must_use]
80    pub fn new(network: Network) -> Self {
81        Session {
82            network,
83            readonly: false,
84            limits: Limits::VENUE,
85            timing: Timing::DEFAULT,
86            token: None,
87            wanted: Vec::new(),
88            link: Link::Closed,
89            failures: 0,
90            next_id: 1,
91            waiting: Vec::new(),
92            chains: BTreeMap::new(),
93            owed: Vec::new(),
94        }
95    }
96
97    /// On the read-only socket (`?readonly=true`), which serves every channel, private ones with
98    /// their token included, from restricted regions too, and takes no transactions.
99    #[must_use]
100    pub const fn read_only(mut self) -> Self {
101        self.readonly = true;
102        self
103    }
104
105    /// A connection's share of the limits; sessions sharing an IP split its accounts.
106    #[must_use]
107    pub const fn with_limits(mut self, limits: Limits) -> Self {
108        self.limits = limits;
109        self
110    }
111
112    #[must_use]
113    pub const fn with_timing(mut self, timing: Timing) -> Self {
114        self.timing = timing;
115        self
116    }
117
118    /// With the account's auth token ([`crate::Client::auth_token`]) for the channels that take
119    /// it. A token reads one account and lapses (eight hours at most): set a fresh one before it
120    /// does with [`Session::set_auth`].
121    #[must_use]
122    pub fn with_auth(mut self, token: impl Into<String>) -> Self {
123        self.token = Some(Token(token.into()));
124        self
125    }
126
127    /// A fresh token, or none. The venue reads a token when a channel is subscribed: what is
128    /// already subscribed goes on, and the new one goes out with the next subscribe, resubscribe
129    /// or reconnect.
130    pub fn set_auth(&mut self, token: Option<String>) {
131        self.token = token.map(Token);
132    }
133
134    #[must_use]
135    pub const fn network(&self) -> Network {
136        self.network
137    }
138
139    #[must_use]
140    pub const fn is_read_only(&self) -> bool {
141        self.readonly
142    }
143
144    /// Where to connect: the network's socket, or its read-only one.
145    #[must_use]
146    pub const fn url(&self) -> &'static str {
147        match self.readonly {
148            true => self.network.ws_readonly_url(),
149            false => self.network.ws_url(),
150        }
151    }
152
153    #[must_use]
154    pub const fn is_open(&self) -> bool {
155        matches!(self.link, Link::Open { .. })
156    }
157
158    /// What this session subscribes to on every socket, in the order it was asked for.
159    #[must_use]
160    pub fn wanted(&self) -> &[Subscription] {
161        &self.wanted
162    }
163
164    /// The distinct accounts the wanted subscriptions follow.
165    #[must_use]
166    pub fn accounts(&self) -> usize {
167        let mut accounts: Vec<_> = self.wanted.iter().filter_map(Subscription::account).collect();
168        accounts.sort_unstable();
169        accounts.dedup();
170        accounts.len()
171    }
172
173    /// The ids of transactions sent on this socket and not yet answered.
174    #[must_use]
175    pub fn waiting(&self) -> &[u64] {
176        &self.waiting
177    }
178
179    /// Wants `subscription` from now on, on this socket and every later one. Gives the frame to
180    /// send when the socket is open and it is new; nothing when it was already wanted (the venue
181    /// refuses a second subscribe). Past the limits, without the token it needs, or on a network
182    /// that does not serve it, it is refused and not kept.
183    pub fn subscribe(&mut self, subscription: Subscription) -> Result<Option<String>, Refused> {
184        if self.wanted.contains(&subscription) {
185            return Ok(None);
186        }
187        subscription.check(self.network)?;
188        if subscription.needs_auth() && self.token.is_none() {
189            return Err(Refused::NoToken(subscription.channel()));
190        }
191        if self.wanted.len() >= self.limits.subscriptions {
192            return Err(Refused::Subscriptions { limit: self.limits.subscriptions });
193        }
194        if let Some(account) = subscription.account()
195            && !self.wanted.iter().any(|wanted| wanted.account() == Some(account))
196            && self.accounts() >= self.limits.accounts
197        {
198            return Err(Refused::Accounts { limit: self.limits.accounts });
199        }
200        let frame = self.is_open().then(|| subscribe(&subscription, self.auth()));
201        self.wanted.push(subscription);
202        Ok(frame)
203    }
204
205    /// Stops wanting `subscription`. Gives the frame to send when the socket is open and it was
206    /// wanted.
207    pub fn unsubscribe(&mut self, subscription: &Subscription) -> Option<String> {
208        let at = self.wanted.iter().position(|wanted| wanted == subscription)?;
209        let gone = self.wanted.remove(at);
210        if let Subscription::OrderBook { market } = gone {
211            self.chains.remove(&market);
212        }
213        self.is_open().then(|| unsubscribe(&gone))
214    }
215
216    /// An unsubscribe and a subscribe for a fresh snapshot of what is wanted, on an open socket;
217    /// nothing otherwise. A second subscribe alone the venue refuses.
218    pub fn resubscribe(&mut self, subscription: &Subscription) -> Vec<String> {
219        if !self.is_open() || !self.wanted.contains(subscription) {
220            return Vec::new();
221        }
222        if let Subscription::OrderBook { market } = subscription {
223            self.chains.insert(*market, None);
224        }
225        resubscribe(subscription, self.auth()).into()
226    }
227
228    /// A signed transaction's frame (`jsonapi/sendtx`) and the id its answer will name. Only on
229    /// an open socket that is not read-only, and only while fewer than the limit wait for an
230    /// answer. One with the owner's signature: [`Session::send_tx_info`] with
231    /// [`Signed::tx_info_with_l1`].
232    pub fn send_tx(&mut self, signed: &Signed) -> Result<(u64, String), Refused> {
233        self.send_tx_info(signed.tx_type().code(), &signed.tx_info())
234    }
235
236    /// [`Session::send_tx`] for a type and `tx_info` as signed.
237    pub fn send_tx_info(&mut self, tx_type: u8, tx_info: &str) -> Result<(u64, String), Refused> {
238        let id = self.take_id()?;
239        let frame = send_tx(&id.to_string(), tx_type, tx_info).map_err(Refused::Unwritable)?;
240        Ok((self.sent(id), frame))
241    }
242
243    /// Up to fifty signed transactions as one batch (`jsonapi/sendtxbatch`): one account and API
244    /// key, consecutive nonces. None may need the owner's signature.
245    pub fn send_tx_batch(&mut self, signed: &[Signed]) -> Result<(u64, String), Refused> {
246        let txs: Vec<(u8, String)> = signed.iter().map(|one| (one.tx_type().code(), one.tx_info())).collect();
247        self.send_tx_batch_infos(&txs)
248    }
249
250    /// [`Session::send_tx_batch`] for types and `tx_info`s as signed.
251    pub fn send_tx_batch_infos(&mut self, txs: &[(u8, String)]) -> Result<(u64, String), Refused> {
252        if !batch_fits(txs.len()) {
253            return Err(Refused::Batch(txs.len()));
254        }
255        let id = self.take_id()?;
256        let frame = send_tx_batch(&id.to_string(), txs);
257        Ok((self.sent(id), frame))
258    }
259
260    /// The socket opened at `now`. Gives the frames that subscribe to everything wanted, in
261    /// order. Each book's chain starts again from its snapshot.
262    pub fn opened(&mut self, now: u64) -> Vec<String> {
263        self.link = Link::Open { since: now, heard: now, talked: false, pinged: now };
264        self.chains.clear();
265        self.owed.clear();
266        self.wanted.iter().map(|subscription| subscribe(subscription, self.auth())).collect()
267    }
268
269    /// A text frame arrived at `now`. Anything that arrives shows the socket is alive, whether it
270    /// reads or not. A transaction's answer or refusal stops it waiting; a ping is owed a pong;
271    /// an order book update that does not follow its book comes back as [`Frame::Gap`], and the
272    /// resubscribe that brings a fresh snapshot is owed ([`Session::owed`]).
273    pub fn received(&mut self, now: u64, text: &str) -> Result<Frame, BadFrame> {
274        let frame = decode(text);
275        if let Link::Open { heard, talked, .. } = &mut self.link {
276            *heard = (*heard).max(now);
277            *talked = true;
278        }
279        let frame = frame?;
280        if let Some(id) = frame.answers().and_then(|id| id.parse::<u64>().ok()) {
281            self.waiting.retain(|waiting| *waiting != id);
282        }
283        Ok(match frame {
284            Frame::Ping => {
285                self.owed.push(PONG.to_owned());
286                Frame::Ping
287            }
288            Frame::Snapshot(Data::OrderBook(book)) => {
289                self.chains.insert(book.market, Some(book.nonce));
290                Frame::Snapshot(Data::OrderBook(book))
291            }
292            Frame::Update(Data::OrderBook(book)) => match self.chains.get(&book.market).copied() {
293                Some(Some(expected)) if book.begin_nonce != expected => {
294                    let gap = Gap { market: book.market, expected, got: book.begin_nonce };
295                    let again = self.resubscribe(&Subscription::OrderBook { market: book.market });
296                    self.owed.extend(again);
297                    self.chains.insert(book.market, None);
298                    Frame::Gap(gap)
299                }
300                Some(Some(_)) => {
301                    self.chains.insert(book.market, Some(book.nonce));
302                    Frame::Update(Data::OrderBook(book))
303                }
304                _ => Frame::Update(Data::OrderBook(book)),
305            },
306            other => other,
307        })
308    }
309
310    /// The frames the venue is owed, to send now and once: the pong to its ping, the
311    /// unsubscribe and subscribe that bring a book a fresh snapshot after a gap.
312    pub fn owed(&mut self) -> Vec<String> {
313        std::mem::take(&mut self.owed)
314    }
315
316    fn auth(&self) -> Option<&str> {
317        self.token.as_ref().map(|token| token.0.as_str())
318    }
319
320    /// The next id, when a transaction may go out at all.
321    fn take_id(&mut self) -> Result<u64, Refused> {
322        if self.readonly {
323            return Err(Refused::ReadOnly);
324        }
325        if !self.is_open() {
326            return Err(Refused::Closed);
327        }
328        if self.waiting.len() >= self.limits.transactions {
329            return Err(Refused::Transactions { limit: self.limits.transactions });
330        }
331        let id = self.next_id;
332        self.next_id = self.next_id.saturating_add(1);
333        Ok(id)
334    }
335
336    /// `id` went out: it waits for its answer.
337    fn sent(&mut self, id: u64) -> u64 {
338        self.waiting.push(id);
339        id
340    }
341}
342
343#[cfg(test)]
344#[path = "tests/session.rs"]
345mod tests;
df_lighter_core/live/session/mod.rs

df_lighter_core/live/session/
mod.rs
