polyoxide_binance/usdm/ws/
error.rs

1//! Errors of the socket tiers and how each one recovers.
2
3use std::time::Duration;
4
5use thiserror::Error;
6
7use crate::usdm::ws::{stream::StreamName, StreamPath};
8
9/// What to do about an error.
10#[derive(Debug, Clone, Copy, PartialEq, Eq)]
11#[non_exhaustive]
12pub enum Recovery {
13    /// Replace the connection and replay its streams.
14    Reconnect,
15    /// Skip this frame and keep reading.
16    SkipFrame,
17    /// Give up: retrying replays the same failure.
18    Fatal,
19}
20
21/// Error type for the socket tiers.
22#[derive(Error, Debug)]
23#[non_exhaustive]
24pub enum UsdmWsError {
25    /// The connection could not be opened, or failed mid-stream.
26    #[error("WebSocket transport error: {0}")]
27    Connect(#[from] Box<tokio_tungstenite::tungstenite::Error>),
28
29    /// The connection did not open within the connect timeout.
30    #[error("no connection within {0:?}")]
31    ConnectTimeout(Duration),
32
33    /// The server closed the connection, with its close code and reason when
34    /// it sent them: `1008 "Too many requests"` after a burst of requests,
35    /// `1008 "Invalid request"` after a refused 1025th stream.
36    #[error("the server closed the connection ({code:?}: {reason})")]
37    Closed {
38        /// The close code.
39        code: Option<u16>,
40        /// The close reason.
41        reason: String,
42    },
43
44    /// The server answered a request with `{"error": {"code", "msg"}}`.
45    #[error("the server refused the request: {code} {msg}")]
46    Refused {
47        /// Binance's code.
48        code: i64,
49        /// Binance's message.
50        msg: String,
51    },
52
53    /// No answer to a request, or no pong to a ping (id 0), within the answer
54    /// timeout.
55    #[error("no answer to request {id} within {timeout:?}")]
56    NoAnswer {
57        /// The request id.
58        id: u64,
59        /// How long the client waited.
60        timeout: Duration,
61    },
62
63    /// An answer that was not the documented shape.
64    #[error("malformed answer to request {id}: {raw}")]
65    Response {
66        /// The request id.
67        id: u64,
68        /// The answer as sent.
69        raw: String,
70    },
71
72    /// A subscribe that would take a connection past Binance's cap. Nothing
73    /// was sent: the server answers the 1025th stream with an error and then
74    /// closes the connection, losing all 1024.
75    #[error("a {path} connection carries at most {limit} streams")]
76    TooManyStreams {
77        /// The path whose connection is full.
78        path: StreamPath,
79        /// The cap.
80        limit: usize,
81    },
82
83    /// A stream for the other path, on the bare tier. Nothing was sent.
84    #[error("{stream} rides the {} path, not {path}", stream.path())]
85    WrongPath {
86        /// The stream.
87        stream: StreamName,
88        /// The connection's path.
89        path: StreamPath,
90    },
91
92    /// A text frame that did not decode as an update.
93    #[error("a frame on {stream:?} did not decode: {reason}")]
94    Frame {
95        /// The envelope's stream name, or empty when there was none.
96        stream: String,
97        /// The frame as sent.
98        raw: String,
99        /// Why it did not decode.
100        reason: String,
101    },
102
103    /// The supervised tier has stopped; its handles can no longer be used.
104    #[error("the supervised connection has stopped")]
105    Stopped,
106}
107
108impl From<tokio_tungstenite::tungstenite::Error> for UsdmWsError {
109    fn from(err: tokio_tungstenite::tungstenite::Error) -> Self {
110        Self::Connect(Box::new(err))
111    }
112}
113
114impl UsdmWsError {
115    /// How the supervised tier treats this error.
116    pub fn recovery(&self) -> Recovery {
117        use tokio_tungstenite::tungstenite::Error as Ws;
118        match self {
119            Self::Connect(err) => match &**err {
120                // A handshake refused with 408, 425, 429 or 5xx is the host
121                // being unwell or throttling, as core's `is_retriable` reads
122                // those statuses; any other status is our request.
123                Ws::Http(response) => {
124                    let status = response.status();
125                    if status.is_server_error() || matches!(status.as_u16(), 408 | 425 | 429) {
126                        Recovery::Reconnect
127                    } else {
128                        Recovery::Fatal
129                    }
130                }
131                Ws::Url(_) | Ws::Tls(_) | Ws::HttpFormat(_) | Ws::AttackAttempt => Recovery::Fatal,
132                // `tungstenite::Error` is non_exhaustive; an unknown transport
133                // error costs one retry cycle if it turns out permanent.
134                _ => Recovery::Reconnect,
135            },
136            Self::ConnectTimeout(_)
137            | Self::Closed { .. }
138            | Self::NoAnswer { .. }
139            | Self::Response { .. } => Recovery::Reconnect,
140            Self::Frame { .. } => Recovery::SkipFrame,
141            Self::Refused { .. }
142            | Self::TooManyStreams { .. }
143            | Self::WrongPath { .. }
144            | Self::Stopped => Recovery::Fatal,
145        }
146    }
147}
148
149#[cfg(test)]
150mod tests {
151    use super::*;
152    use tokio_tungstenite::tungstenite::{self, http};
153
154    fn http_error(status: u16) -> UsdmWsError {
155        UsdmWsError::from(tungstenite::Error::Http(
156            http::Response::builder().status(status).body(None).unwrap(),
157        ))
158    }
159
160    #[test]
161    fn a_handshake_refused_by_the_host_reconnects_and_one_refused_for_us_is_fatal() {
162        assert_eq!(http_error(503).recovery(), Recovery::Reconnect);
163        assert_eq!(http_error(429).recovery(), Recovery::Reconnect);
164        assert_eq!(http_error(408).recovery(), Recovery::Reconnect);
165        assert_eq!(http_error(425).recovery(), Recovery::Reconnect);
166        assert_eq!(http_error(451).recovery(), Recovery::Fatal);
167        assert_eq!(http_error(404).recovery(), Recovery::Fatal);
168        assert_eq!(http_error(418).recovery(), Recovery::Fatal);
169    }
170
171    #[test]
172    fn losing_the_connection_reconnects_and_a_bad_frame_is_skipped() {
173        for err in [
174            UsdmWsError::from(tungstenite::Error::ConnectionClosed),
175            UsdmWsError::ConnectTimeout(Duration::from_secs(10)),
176            UsdmWsError::Closed {
177                code: Some(1008),
178                reason: "Too many requests".into(),
179            },
180            UsdmWsError::NoAnswer {
181                id: 3,
182                timeout: Duration::from_secs(10),
183            },
184            UsdmWsError::Response {
185                id: 4,
186                raw: "{\"id\":4}".into(),
187            },
188        ] {
189            assert_eq!(err.recovery(), Recovery::Reconnect, "{err}");
190        }
191        let frame = UsdmWsError::Frame {
192            stream: "btcusdt@aggTrade".into(),
193            raw: "{}".into(),
194            reason: "missing field".into(),
195        };
196        assert_eq!(frame.recovery(), Recovery::SkipFrame);
197    }
198
199    #[test]
200    fn a_refusal_and_a_request_the_client_refused_to_send_are_fatal() {
201        let refused = UsdmWsError::Refused {
202            code: 2,
203            msg: "Invalid request".into(),
204        };
205        assert_eq!(refused.recovery(), Recovery::Fatal);
206        let full = UsdmWsError::TooManyStreams {
207            path: StreamPath::Market,
208            limit: 1024,
209        };
210        assert_eq!(full.recovery(), Recovery::Fatal);
211        assert_eq!(
212            full.to_string(),
213            "a market connection carries at most 1024 streams"
214        );
215        let wrong = UsdmWsError::WrongPath {
216            stream: StreamName::BookTicker(crate::usdm::types::Symbol::new("BTCUSDT").unwrap()),
217            path: StreamPath::Market,
218        };
219        assert_eq!(wrong.recovery(), Recovery::Fatal);
220        assert_eq!(
221            wrong.to_string(),
222            "btcusdt@bookTicker rides the public path, not market"
223        );
224        let url = UsdmWsError::from(tungstenite::Error::Url(
225            tungstenite::error::UrlError::NoHostName,
226        ));
227        assert_eq!(url.recovery(), Recovery::Fatal);
228        assert_eq!(UsdmWsError::Stopped.recovery(), Recovery::Fatal);
229    }
230
231    #[test]
232    fn usdm_ws_error_is_send_sync() {
233        fn assert_send_sync<T: Send + Sync>() {}
234        assert_send_sync::<UsdmWsError>();
235    }
236}
polyoxide_binance/usdm/ws/error.rs

polyoxide_binance/usdm/ws/
error.rs
