polyoxide_binance/spot/ws/
error.rs

1//! Errors of the spot socket tiers, and their classes.
2
3use std::time::Duration;
4
5use polyoxide_venue::{class_for_close_code, Class, Classify};
6use polyoxide_ws::{Kind, WsError};
7use thiserror::Error;
8
9use crate::spot::ws::StreamPath;
10
11polyoxide_venue::impl_ws_classification!();
12
13/// Error type for the spot socket tiers.
14#[derive(Error, Debug)]
15#[non_exhaustive]
16pub enum SpotWsError {
17    /// The connection could not be opened, or failed mid-stream: the kit's
18    /// one socket error, read through its [`Kind`]. A
19    /// server close and a connect timeout have variants of their own.
20    #[error("WebSocket transport error: {0}")]
21    Connect(#[source] WsError),
22
23    /// The connection did not open within the connect timeout.
24    #[error("no connection within {0:?}")]
25    ConnectTimeout(Duration),
26
27    /// The server closed the connection, with its close code and reason when
28    /// it sent them: `1008 "Too many requests"` after more than 5 incoming
29    /// messages in a second.
30    #[error("the server closed the connection ({code:?}: {reason})")]
31    Closed {
32        /// The close code.
33        code: Option<u16>,
34        /// The close reason.
35        reason: String,
36    },
37
38    /// The server answered a request with `{"error": {"code", "msg"}}`.
39    #[error("the server refused the request: {code} {msg}")]
40    Refused {
41        /// Binance's code.
42        code: i64,
43        /// Binance's message.
44        msg: String,
45    },
46
47    /// No answer to a request, or no pong to a ping (id 0), within the answer
48    /// timeout.
49    #[error("no answer to request {id} within {timeout:?}")]
50    NoAnswer {
51        /// The request id.
52        id: u64,
53        /// How long the client waited.
54        timeout: Duration,
55    },
56
57    /// An answer that was not the documented shape.
58    #[error("malformed answer to request {id}: {raw}")]
59    Response {
60        /// The request id.
61        id: u64,
62        /// The answer as sent.
63        raw: String,
64    },
65
66    /// A subscribe that would take a connection past Binance's cap. Nothing
67    /// was sent: a 1025th stream is not one the client probes, since the
68    /// server's answer to it closes the connection and loses all 1024.
69    #[error("a {path} connection carries at most {limit} streams")]
70    TooManyStreams {
71        /// The path whose connection is full.
72        path: StreamPath,
73        /// The cap.
74        limit: usize,
75    },
76
77    /// A text frame that did not decode as an update.
78    #[error("a frame on {stream:?} did not decode: {reason}")]
79    Frame {
80        /// The envelope's stream name, or empty when there was none.
81        stream: String,
82        /// The frame as sent.
83        raw: String,
84        /// Why it did not decode.
85        reason: String,
86    },
87
88    /// The supervised tier has stopped; its handles can no longer be used.
89    ///
90    /// `cause` is the error a path stopped on, as [`Classify`] read it, so a
91    /// stream stopped by a reconnect refused with 451 is classed and tagged as
92    /// that refusal is. `None` when the client stopped for another reason,
93    /// such as a task that panicked, or before the stop's cause was known.
94    #[error("the supervised connection has stopped{}", describe_stop(cause.as_ref()))]
95    Stopped {
96        /// The error a path stopped on.
97        cause: Option<StopCause>,
98    },
99}
100
101/// The error a supervised path stopped on, kept for
102/// [`SpotWsError::Stopped`] after the error itself was yielded: its class,
103/// whether it was a fault, the wait it asked for, and its text.
104#[derive(Debug, Clone, PartialEq, Eq)]
105#[non_exhaustive]
106pub struct StopCause {
107    /// The error's class.
108    pub class: Class,
109    /// Whether the error was a fault.
110    pub is_fault: bool,
111    /// The wait the error asked for.
112    pub retry_after: Option<Duration>,
113    /// The error's text.
114    pub message: String,
115}
116
117impl StopCause {
118    /// The cause of a stop on `err`.
119    pub fn of(err: &SpotWsError) -> Self {
120        Self {
121            class: err.class(),
122            is_fault: err.is_fault(),
123            retry_after: err.retry_after(),
124            message: err.to_string(),
125        }
126    }
127}
128
129/// The tail of a stop's message: its cause's text, when it has one.
130fn describe_stop(cause: Option<&StopCause>) -> String {
131    cause.map_or_else(String::new, |cause| format!(": {}", cause.message))
132}
133
134/// A server close is [`SpotWsError::Closed`], a connect timeout
135/// [`SpotWsError::ConnectTimeout`], and every other socket failure
136/// [`SpotWsError::Connect`].
137impl From<WsError> for SpotWsError {
138    fn from(err: WsError) -> Self {
139        match err.kind() {
140            Kind::Closed { code, reason } => Self::Closed {
141                code,
142                reason: reason.to_owned(),
143            },
144            Kind::ConnectTimeout { after } => Self::ConnectTimeout(after),
145            _ => Self::Connect(err),
146        }
147    }
148}
149
150/// The transport by the one socket table
151/// (`polyoxide_venue::impl_ws_classification!`), a server close by its close
152/// code, a refused request a `VenueRefusal` with Binance's code in decimal,
153/// an answer or frame that did not parse a `Decode`, and the client's own
154/// refusals an `InvalidRequest`.
155///
156/// One row is Binance's own, ahead of the table (D14): a handshake refused
157/// with `403` is `Restricted`, as on REST. Binance's firewall answers it, and
158/// it is not a credential failure, since the streams need none.
159///
160/// Every error is a fault except a handshake refused with `451`: Binance does
161/// not serve the caller's region, a defined outcome, as
162/// [`SpotError::RegionBlocked`](crate::spot::SpotError::RegionBlocked) is on
163/// REST. A `418` ban and a `403` from the firewall stay faults. A refused
164/// handshake's `Retry-After`, of any status, is the error's wait.
165///
166/// [`Stopped`](SpotWsError::Stopped) follows its cause, so a stop on a 451
167/// refusal is `Restricted` and not a fault; one with no cause is an
168/// `InvalidRequest` and a fault.
169impl Classify for SpotWsError {
170    fn class(&self) -> Class {
171        match self {
172            Self::Connect(err) => match err.kind() {
173                Kind::Refused { status: 403, .. } => Class::Restricted,
174                _ => ws_class(err),
175            },
176            Self::ConnectTimeout(_) | Self::NoAnswer { .. } => Class::Network,
177            Self::Closed { code, .. } => class_for_close_code(*code),
178            Self::Refused { code, .. } => Class::VenueRefusal {
179                code: Some(code.to_string().into()),
180            },
181            Self::Response { .. } | Self::Frame { .. } => Class::Decode,
182            Self::Stopped { cause: Some(cause) } => cause.class.clone(),
183            Self::TooManyStreams { .. } | Self::Stopped { cause: None } => Class::InvalidRequest,
184        }
185    }
186
187    fn is_fault(&self) -> bool {
188        match self {
189            Self::Connect(err) => ws_is_fault(err),
190            Self::Stopped { cause: Some(cause) } => cause.is_fault,
191            _ => true,
192        }
193    }
194
195    fn retry_after(&self) -> Option<Duration> {
196        match self {
197            Self::Connect(err) => ws_retry_after(err),
198            Self::Stopped { cause: Some(cause) } => cause.retry_after,
199            _ => self.class().retry_after(),
200        }
201    }
202}
203
204#[cfg(test)]
205mod tests {
206    use super::*;
207    use polyoxide_ws::tungstenite::{self, http};
208
209    fn http_error(status: u16) -> SpotWsError {
210        transport(tungstenite::Error::Http(
211            http::Response::builder().status(status).body(None).unwrap(),
212        ))
213    }
214
215    /// A transport failure as the connection reports it, through the kit.
216    fn transport(err: tungstenite::Error) -> SpotWsError {
217        SpotWsError::from(WsError::from(err))
218    }
219
220    #[test]
221    fn a_server_close_and_a_connect_timeout_keep_their_own_variants() {
222        let closed = SpotWsError::from(WsError::closed(Some(1008), "Too many requests"));
223        assert!(
224            matches!(&closed, SpotWsError::Closed { code: Some(1008), reason } if reason == "Too many requests"),
225            "{closed:?}"
226        );
227        let after = Duration::from_secs(10);
228        assert!(matches!(
229            SpotWsError::from(WsError::connect_timeout(after)),
230            SpotWsError::ConnectTimeout(timeout) if timeout == after
231        ));
232        // The bare tier maps an unanswered request itself, with its id.
233        assert!(matches!(
234            SpotWsError::from(WsError::no_answer(after)),
235            SpotWsError::Connect(_)
236        ));
237        assert!(matches!(
238            transport(tungstenite::Error::ConnectionClosed),
239            SpotWsError::Connect(_)
240        ));
241    }
242
243    #[test]
244    fn a_refused_429_surfaces_its_retry_after() {
245        let err = transport(tungstenite::Error::Http(
246            http::Response::builder()
247                .status(429)
248                .header("Retry-After", "7")
249                .body(None)
250                .unwrap(),
251        ));
252        assert_eq!(
253            err.class(),
254            Class::RateLimited {
255                retry_after: Some(Duration::from_secs(7))
256            }
257        );
258        assert_eq!(err.retry_after(), Some(Duration::from_secs(7)));
259        assert!(err.is_retriable());
260    }
261
262    #[test]
263    fn a_refused_503_surfaces_its_retry_after() {
264        // Every refusal's `Retry-After` is the error's wait; only a 429's class
265        // carries it.
266        let err = transport(tungstenite::Error::Http(
267            http::Response::builder()
268                .status(503)
269                .header("Retry-After", "30")
270                .body(None)
271                .unwrap(),
272        ));
273        assert_eq!(err.class(), Class::Unavailable { code: None });
274        assert_eq!(err.retry_after(), Some(Duration::from_secs(30)));
275    }
276
277    #[test]
278    fn a_stop_follows_its_cause() {
279        // A stop on a 451 refusal is that refusal's class and not a fault, a
280        // stop on a 429 keeps its wait, and one with no cause is the client's
281        // own `InvalidRequest`.
282        let region = SpotWsError::Stopped {
283            cause: Some(StopCause::of(&http_error(451))),
284        };
285        assert_eq!(region.class(), Class::Restricted);
286        assert!(!region.is_fault());
287        assert_eq!(
288            region.to_string(),
289            "the supervised connection has stopped: WebSocket transport error: HTTP error: 451 \
290             Unavailable For Legal Reasons"
291        );
292        let throttled = SpotWsError::Stopped {
293            cause: Some(StopCause::of(&transport(tungstenite::Error::Http(
294                http::Response::builder()
295                    .status(429)
296                    .header("Retry-After", "30")
297                    .body(None)
298                    .unwrap(),
299            )))),
300        };
301        assert_eq!(throttled.retry_after(), Some(Duration::from_secs(30)));
302        let bare = SpotWsError::Stopped { cause: None };
303        assert_eq!(bare.class(), Class::InvalidRequest);
304        assert!(bare.is_fault());
305        assert_eq!(bare.to_string(), "the supervised connection has stopped");
306    }
307
308    #[test]
309    fn spot_ws_error_is_send_sync() {
310        fn assert_send_sync<T: Send + Sync>() {}
311        assert_send_sync::<SpotWsError>();
312    }
313
314    #[test]
315    fn the_transport_follows_the_socket_table() {
316        let bad_header = http::header::HeaderName::from_bytes(b"in valid").unwrap_err();
317        let rows = [
318            (tungstenite::Error::ConnectionClosed, Class::Network),
319            (
320                tungstenite::Error::Io(std::io::ErrorKind::ConnectionReset.into()),
321                Class::Network,
322            ),
323            (
324                tungstenite::Error::Protocol(
325                    tungstenite::error::ProtocolError::ResetWithoutClosingHandshake,
326                ),
327                Class::Network,
328            ),
329            (
330                tungstenite::Error::WriteBufferFull(tungstenite::Message::Close(None)),
331                Class::Network,
332            ),
333            (tungstenite::Error::Utf8, Class::Network),
334            (tungstenite::Error::AlreadyClosed, Class::InvalidRequest),
335            (
336                tungstenite::Error::Url(tungstenite::error::UrlError::NoHostName),
337                Class::InvalidRequest,
338            ),
339            (
340                tungstenite::Error::HttpFormat(bad_header.into()),
341                Class::InvalidRequest,
342            ),
343            (
344                tungstenite::Error::Tls(tungstenite::error::TlsError::InvalidDnsName),
345                Class::InvalidRequest,
346            ),
347            (tungstenite::Error::AttackAttempt, Class::InvalidRequest),
348            (
349                tungstenite::Error::Capacity(tungstenite::error::CapacityError::TooManyHeaders),
350                Class::InvalidRequest,
351            ),
352        ];
353        for (err, class) in rows {
354            let err = transport(err);
355            assert_eq!(err.class(), class, "{err:?}");
356            assert!(err.is_fault(), "{err:?}");
357            assert_eq!(err.retry_after(), None, "{err:?}");
358        }
359        // (status, class, is_fault)
360        for (status, class, fault) in [
361            (401, Class::Unauthorized, true),
362            // Binance's firewall, as on REST.
363            (403, Class::Restricted, true),
364            (404, Class::VenueRefusal { code: None }, true),
365            // A ban the client earned by overspending.
366            (418, Class::Restricted, true),
367            (429, Class::RateLimited { retry_after: None }, true),
368            // A region block, as `SpotError::RegionBlocked` is on REST.
369            (451, Class::Restricted, false),
370            (503, Class::Unavailable { code: None }, true),
371            (200, Class::Decode, true),
372        ] {
373            assert_eq!(http_error(status).class(), class, "{status}");
374            assert_eq!(http_error(status).is_fault(), fault, "{status}");
375        }
376    }
377
378    #[test]
379    fn every_variant_classifies() {
380        let rows = [
381            (http_error(503), Class::Unavailable { code: None }),
382            (
383                SpotWsError::ConnectTimeout(Duration::from_secs(10)),
384                Class::Network,
385            ),
386            (
387                SpotWsError::Closed {
388                    code: Some(1008),
389                    reason: "Too many requests".into(),
390                },
391                Class::VenueRefusal {
392                    code: Some("1008".into()),
393                },
394            ),
395            (
396                SpotWsError::Closed {
397                    code: Some(1001),
398                    reason: String::new(),
399                },
400                Class::Network,
401            ),
402            (
403                SpotWsError::Closed {
404                    code: None,
405                    reason: String::new(),
406                },
407                Class::Network,
408            ),
409            (
410                SpotWsError::Refused {
411                    code: 2,
412                    msg: "Invalid request".into(),
413                },
414                Class::VenueRefusal {
415                    code: Some("2".into()),
416                },
417            ),
418            (
419                SpotWsError::NoAnswer {
420                    id: 3,
421                    timeout: Duration::from_secs(10),
422                },
423                Class::Network,
424            ),
425            (
426                SpotWsError::Response {
427                    id: 4,
428                    raw: "{}".into(),
429                },
430                Class::Decode,
431            ),
432            (
433                SpotWsError::TooManyStreams {
434                    path: StreamPath::Stream,
435                    limit: 1024,
436                },
437                Class::InvalidRequest,
438            ),
439            (
440                SpotWsError::Frame {
441                    stream: "btcusdt@aggTrade".into(),
442                    raw: "{}".into(),
443                    reason: "missing field".into(),
444                },
445                Class::Decode,
446            ),
447            (SpotWsError::Stopped { cause: None }, Class::InvalidRequest),
448        ];
449        for (err, class) in rows {
450            assert_eq!(err.class(), class, "{err:?}");
451            assert!(err.is_fault(), "{err:?}");
452            assert_eq!(err.retry_after(), None, "{err:?}");
453            assert_eq!(err.is_retriable(), class.is_retriable(), "{err:?}");
454        }
455    }
456}
polyoxide_binance/spot/ws/error.rs

polyoxide_binance/spot/ws/
error.rs
