polyoxide_binance/usdm/ws/
error.rs

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

polyoxide_binance/usdm/ws/
error.rs
