df_hyperliquid_core/client/
api.rs

1//! One network's `/info` and `/exchange` behind one rate-limit gate. [`Api`]
2//! is a [`Transport`], so every read and send passes the gate and the meter;
3//! the [`Http`] underneath knows nothing of networks.
4
5use std::sync::{Mutex, MutexGuard, PoisonError};
6
7use crate::client::limit::{Gate, Meter, Spent, action_weight, base_weight, per_items, read_weight};
8use crate::client::network::Network;
9use crate::client::transport::{Http, INFO, Request, Transport, TransportError};
10use crate::utils::json::Json;
11
12/// How the meter names a post to `/exchange`.
13const EXCHANGE_KIND: &str = "exchange";
14
15/// Unix milliseconds. Injected: std has no clock in the browser or Workers,
16/// and a test pins the time.
17pub type Clock = fn() -> u64;
18
19#[derive(Debug, Default)]
20struct Limiter {
21    gate: Gate,
22    meter: Meter,
23}
24
25/// One network's endpoints, rate limited.
26pub struct Api<H> {
27    http: H,
28    network: Network,
29    clock: Clock,
30    limiter: Mutex<Limiter>,
31}
32
33impl<H: Http> Api<H> {
34    pub fn new(http: H, network: Network, clock: Clock) -> Self {
35        Api { http, network, clock, limiter: Mutex::new(Limiter::default()) }
36    }
37
38    #[must_use]
39    pub const fn network(&self) -> Network {
40        self.network
41    }
42
43    #[must_use]
44    pub const fn http(&self) -> &H {
45        &self.http
46    }
47
48    /// What went out over the last minute, and how long the gate stays closed.
49    pub fn spent(&self) -> Spent {
50        let now = (self.clock)();
51        let mut limiter = self.limiter();
52        let cooldown_ms = limiter.gate.cooldown_ms(now);
53        Spent { cooldown_ms, ..limiter.meter.spent(now) }
54    }
55
56    fn limiter(&self) -> MutexGuard<'_, Limiter> {
57        self.limiter.lock().unwrap_or_else(PoisonError::into_inner)
58    }
59}
60
61impl<H: Http> Transport for Api<H> {
62    /// Past the gate, then to the network's endpoint. A read near the limit
63    /// is held back as [`TransportError::RateLimited`] without going out; an
64    /// action never is, but a closed gate holds both. Nothing is retried.
65    async fn post(&self, path: &str, body: String) -> Result<String, TransportError> {
66        let is_read = path == INFO;
67        let kind = match is_read {
68            true => Json::parse(&body)
69                .ok()
70                .and_then(|request| request.get("type").and_then(Json::as_str).map(ToOwned::to_owned))
71                .unwrap_or_default(),
72            false => EXCHANGE_KIND.to_owned(),
73        };
74        let batched = match is_read {
75            true => 0,
76            false => batched(&body),
77        };
78        let ticket = {
79            let now = (self.clock)();
80            let mut limiter = self.limiter();
81            let ticket = limiter.gate.open(now).map_err(|wait| rate_limited(Some(wait)))?;
82            if is_read && let Some(wait) = limiter.meter.wait_for(now, base_weight(&kind)) {
83                return Err(rate_limited(Some(wait)));
84            }
85            ticket
86        };
87        let request = Request::post_json(format!("{}{path}", self.network.api_url()), body);
88        let answer = self.http.send(request).await;
89        let now = (self.clock)();
90        let mut limiter = self.limiter();
91        let response = answer?;
92        if response.status == 429 {
93            limiter.meter.refuse(now);
94            let retry_after_ms = response.header("retry-after").and_then(|seconds| seconds.trim().parse::<u64>().ok());
95            let wait = limiter.gate.struck(ticket, now, retry_after_ms.map(|seconds| seconds.saturating_mul(1_000)));
96            return Err(rate_limited(Some(wait)));
97        }
98        let text = response.text();
99        let weight = match (is_read, response.is_success()) {
100            (true, true) => read_weight(&kind, per_items(&kind).map_or(0, |_| items(&text))),
101            (true, false) => base_weight(&kind),
102            (false, _) => action_weight(batched),
103        };
104        limiter.meter.record(now, &kind, weight);
105        match response.is_success() {
106            true => {
107                limiter.gate.passed(ticket);
108                Ok(text)
109            }
110            false => Err(TransportError::Status { status: response.status, body: text }),
111        }
112    }
113}
114
115const fn rate_limited(retry_after_ms: Option<u64>) -> TransportError {
116    TransportError::RateLimited { retry_after_ms }
117}
118
119/// The orders, cancels or modifies an `/exchange` body batches.
120fn batched(body: &str) -> usize {
121    let Ok(body) = Json::parse(body) else { return 0 };
122    let Some(action) = body.get("action") else { return 0 };
123    ["orders", "cancels", "modifies"]
124        .iter()
125        .find_map(|key| action.get(key).and_then(Json::as_array))
126        .map_or(0, <[Json]>::len)
127}
128
129/// How many items an answer lists; 0 for anything that is not a list.
130fn items(text: &str) -> usize {
131    serde_json::from_str::<Vec<serde::de::IgnoredAny>>(text).map_or(0, |list| list.len())
132}
133
134#[cfg(test)]
135#[path = "tests/api.rs"]
136mod tests;
df_hyperliquid_core/client/api.rs

df_hyperliquid_core/client/
api.rs
