rustrade_data/
error.rs

1#[cfg(feature = "databento")]
2use crate::exchange::databento::DatabentoErrorKind;
3use crate::subscription::{SubKind, candle::CandleInterval};
4use rustrade_instrument::{exchange::ExchangeId, index::error::IndexError};
5use rustrade_integration::{error::SocketError, subscription::SubscriptionId};
6use serde::{Deserialize, Serialize};
7use thiserror::Error;
8
9/// All errors generated in `rustrade-data`.
10#[derive(Debug, Clone, Eq, PartialEq, Ord, PartialOrd, Hash, Deserialize, Serialize, Error)]
11pub enum DataError {
12    #[error("failed to index market data Subscriptions: {0}")]
13    Index(#[from] IndexError),
14
15    #[error("failed to initialise reconnecting MarketStream due to empty subscriptions")]
16    SubscriptionsEmpty,
17
18    #[error("unsupported DynamicStreams Subscription SubKind: {0}")]
19    UnsupportedSubKind(SubKind),
20
21    #[error("initial snapshot missing for: {0}")]
22    InitialSnapshotMissing(SubscriptionId),
23
24    #[error("initial snapshot invalid: {0}")]
25    InitialSnapshotInvalid(String),
26
27    #[error("SocketError: {0}")]
28    Socket(String),
29
30    /// Databento-specific error with categorized kind for programmatic handling.
31    #[cfg(feature = "databento")]
32    #[error("Databento {kind} error ({context}): {message}")]
33    Databento {
34        kind: DatabentoErrorKind,
35        context: String,
36        message: String,
37    },
38
39    #[error("unsupported dynamic Subscription for exchange: {exchange}, kind: {sub_kind}")]
40    Unsupported {
41        exchange: ExchangeId,
42        sub_kind: SubKind,
43    },
44
45    #[error("exchange {exchange} does not support candle interval: {interval}")]
46    UnsupportedInterval {
47        exchange: ExchangeId,
48        interval: CandleInterval,
49    },
50
51    #[error(
52        "\
53        InvalidSequence: first_update_id {first_update_id} does not follow on from the \
54        prev_last_update_id {prev_last_update_id} \
55    "
56    )]
57    InvalidSequence {
58        prev_last_update_id: u64,
59        first_update_id: u64,
60    },
61}
62
63impl DataError {
64    /// Determine if an error requires a [`MarketStream`](super::MarketStream) to re-initialise.
65    // Explicit `match` (not `matches!`) is kept so additional terminal variants can be classified
66    // arm-by-arm as they are added; the lint would otherwise push this to a single `matches!`.
67    #[allow(clippy::match_like_matches_macro)]
68    pub fn is_terminal(&self) -> bool {
69        match self {
70            DataError::InvalidSequence { .. } => true,
71            _ => false,
72        }
73    }
74}
75
76impl From<SocketError> for DataError {
77    fn from(value: SocketError) -> Self {
78        Self::Socket(value.to_string())
79    }
80}
81
82#[cfg(test)]
83mod tests {
84    use super::*;
85
86    #[test]
87    fn test_data_error_is_terminal() {
88        struct TestCase {
89            input: DataError,
90            expected: bool,
91        }
92
93        let tests = vec![
94            TestCase {
95                // TC0: is terminal w/ DataError::InvalidSequence
96                input: DataError::InvalidSequence {
97                    prev_last_update_id: 0,
98                    first_update_id: 0,
99                },
100                expected: true,
101            },
102            TestCase {
103                // TC1: is not terminal w/ DataError::Socket
104                input: DataError::from(SocketError::Sink),
105                expected: false,
106            },
107            TestCase {
108                // TC2: not terminal w/ DataError::UnsupportedInterval — a caller
109                // configuration error, not a stream condition warranting re-init.
110                input: DataError::UnsupportedInterval {
111                    exchange: ExchangeId::HyperliquidPerp,
112                    interval: CandleInterval::Sec1,
113                },
114                expected: false,
115            },
116        ];
117
118        for (index, test) in tests.into_iter().enumerate() {
119            let actual = test.input.is_terminal();
120            assert_eq!(actual, test.expected, "TC{} failed", index);
121        }
122    }
123}
rustrade_data/error.rs

rustrade_data/
error.rs
