satd_events_client/error.rs
1//! Error type for the streaming client.
2
3use std::time::Duration;
4
5/// An error from the streaming client.
6///
7/// `Lagged` is deliberately **not** here — a slow-consumer lag notice is a
8/// normal [`Event`](crate::Event), recoverable and carrying a resume cursor.
9/// Variants here are conditions that stop forward progress on a call.
10///
11/// The variants that wrap a [`tonic::Status`] keep it boxed so the server's
12/// message and details survive classification; use [`is_retryable`](Self::is_retryable)
13/// to decide whether to back off and retry versus give up.
14#[derive(Debug, thiserror::Error)]
15#[non_exhaustive]
16pub enum StreamError {
17 /// An unclassified transport or RPC status from the server.
18 #[error("rpc error: {0}")]
19 Transport(#[source] Box<tonic::Status>),
20
21 /// Failed to establish the underlying transport channel.
22 #[error("connect error: {0}")]
23 Connect(#[source] tonic::transport::Error),
24
25 /// The endpoint string could not be parsed.
26 #[error("invalid endpoint: {0}")]
27 InvalidEndpoint(String),
28
29 /// The supplied bearer token is not a valid HTTP header value.
30 #[error("invalid bearer token: not a valid header value")]
31 InvalidToken,
32
33 /// A bearer token was configured against an endpoint that is not encrypted,
34 /// so the token — and, on a scan-key watch, the BIP 352 scan secrets that
35 /// share the stream — would cross the network in cleartext.
36 ///
37 /// Use an `https://` endpoint with [`tls`](crate::StreamClientBuilder::tls)
38 /// or [`tls_ca_pem`](crate::StreamClientBuilder::tls_ca_pem). For loopback
39 /// or a test harness, name the risk with
40 /// [`insecure_bearer_token`](crate::StreamClientBuilder::insecure_bearer_token).
41 #[error(
42 "refusing to send a bearer token over an unencrypted endpoint: {0}. \
43 Encrypt the connection (an https:// endpoint plus tls()/tls_ca_pem(), \
44 both of which need this crate's `tls` feature), or call \
45 insecure_bearer_token() instead of bearer_token() to accept the risk."
46 )]
47 InsecureCredential(String),
48
49 /// The bearer token was missing, malformed, or rejected (gRPC
50 /// `UNAUTHENTICATED`). Potentially fixable by presenting a valid token.
51 #[error("unauthenticated: {0}")]
52 Unauthenticated(#[source] Box<tonic::Status>),
53
54 /// The principal is authenticated but lacks the required capability
55 /// (`stream:subscribe` to open, `stream:watch` to add watches) — gRPC
56 /// `PERMISSION_DENIED`. A permanent configuration error, not retryable.
57 #[error("permission denied: {0}")]
58 PermissionDenied(#[source] Box<tonic::Status>),
59
60 /// The server's subscription cap or a per-principal rate limit was hit when
61 /// the stream was opened (gRPC `RESOURCE_EXHAUSTED`). Both are transient:
62 /// back off and retry. A full watch quota never surfaces here; a refused
63 /// add arrives as [`Event::WatchAddRejected`](crate::Event::WatchAddRejected).
64 #[error("resource exhausted: {0}")]
65 QuotaExhausted(#[source] Box<tonic::Status>),
66
67 /// Reserved for explicit rate-limit signaling in the resilience layer; not
68 /// produced yet. The server answers an over-rate `SetCursor` re-anchor
69 /// in-band, as [`Event::CursorRejected`](crate::Event::CursorRejected) with
70 /// [`CursorRejectReason::RateLimited`](crate::CursorRejectReason::RateLimited),
71 /// not with a status.
72 #[error("rate limited")]
73 RateLimited {
74 /// Suggested backoff before retrying, if known.
75 retry_after: Option<Duration>,
76 },
77
78 /// Reserved. `from_cursor` replay against a server with no block source is a
79 /// silent server-side fallback to forward-only (no status), so this is not
80 /// produced from the wire yet; the resilience layer detects the degraded
81 /// case from the event stream.
82 #[error("cursor replay unavailable: server has no block source")]
83 ReplayUnavailable,
84
85 /// A prefix watch was rejected because `bits` is outside the server's
86 /// configured `[streamprefixminbits, streamprefixmaxbits]` range.
87 #[error("prefix bits {got} out of server range [{min}, {max}]")]
88 PrefixBitsOutOfRange {
89 /// The `bits` value that was rejected.
90 got: u32,
91 /// Server minimum.
92 min: u32,
93 /// Server maximum.
94 max: u32,
95 },
96
97 /// A client-side argument was rejected before reaching the wire (e.g. a
98 /// prefix whose byte length does not match its `bits`).
99 #[error("invalid argument: {0}")]
100 InvalidArgument(String),
101
102 /// A received message could not be decoded into a typed value.
103 #[error("decode error: {0}")]
104 Decode(String),
105
106 /// The control channel was closed before the message could be sent (the
107 /// watch stream has been torn down).
108 #[error("control channel closed")]
109 ControlClosed,
110
111 /// A user-supplied watch-set loader (see
112 /// [`ResilientWatchConfig::watch_set_loader`](crate::ResilientWatchConfig::watch_set_loader))
113 /// returned an error while rebuilding the watch-set on (re)connect. Treated
114 /// by [`ResilientWatch`](crate::ResilientWatch) as a transient
115 /// reconnect-level condition: the loader is retried after a backoff on the
116 /// next connect rather than surfaced, so a momentary failure of the
117 /// integrator's source-of-truth does not crash the consumer.
118 #[error("watch-set loader failed: {0}")]
119 WatchSetLoader(#[source] Box<dyn std::error::Error + Send + Sync>),
120
121 /// The node is two or more minor versions (or a major version) older than
122 /// this SDK. Features the SDK was built against may be missing and request
123 /// fields the node does not recognise would be silently ignored, so the
124 /// stream is refused. Upgrade the node, or accept the risk with
125 /// [`allow_old_node`](crate::StreamClientBuilder::allow_old_node), which turns
126 /// this into a logged warning. Not retryable.
127 #[error(
128 "satd node {node} is too old for this SDK ({sdk}): two or more minor versions \
129 behind. Upgrade the node, or opt in with allow_old_node()"
130 )]
131 NodeTooOld {
132 /// The node's advertised version, or a note that it sent none.
133 node: String,
134 /// This SDK's version.
135 sdk: &'static str,
136 },
137
138 /// The node speaks a different event schema version than this SDK. Always
139 /// refused, whatever the versions; `allow_old_node` does not apply. Not
140 /// retryable.
141 #[error("satd node event schema {node} does not match this SDK's schema {sdk}")]
142 SchemaMismatch {
143 /// The node's schema version (`0` if it sent an unparseable value).
144 node: u32,
145 /// The schema version this SDK was built for.
146 sdk: u32,
147 },
148}
149
150impl StreamError {
151 /// Classify a tonic [`Status`](tonic::Status) into a typed variant, keeping
152 /// the original status (boxed) so its message and details are preserved.
153 pub(crate) fn from_status(status: tonic::Status) -> Self {
154 use tonic::Code;
155 match status.code() {
156 Code::Unauthenticated => StreamError::Unauthenticated(Box::new(status)),
157 Code::PermissionDenied => StreamError::PermissionDenied(Box::new(status)),
158 Code::ResourceExhausted => StreamError::QuotaExhausted(Box::new(status)),
159 _ => StreamError::Transport(Box::new(status)),
160 }
161 }
162
163 /// Whether retrying the operation (after a backoff) could plausibly succeed.
164 ///
165 /// `true` for transport failures the gRPC layer marks transient
166 /// (`UNAVAILABLE`, `DEADLINE_EXCEEDED`, `ABORTED`, `CANCELLED`,
167 /// `RESOURCE_EXHAUSTED`), for connection failures, and for the reserved
168 /// `RateLimited`. `false` for permanent conditions — bad endpoint/token,
169 /// `PERMISSION_DENIED`, client-side argument errors. `Unauthenticated` is
170 /// reported non-retryable: a blind retry with the same token will not help;
171 /// the caller should re-auth and reconnect deliberately. `NodeTooOld` and
172 /// `SchemaMismatch` are non-retryable: reconnecting to the same node gives
173 /// the same answer.
174 pub fn is_retryable(&self) -> bool {
175 use tonic::Code;
176 match self {
177 StreamError::Connect(_) => true,
178 StreamError::RateLimited { .. } => true,
179 StreamError::QuotaExhausted(_) => true,
180 StreamError::Transport(s) => matches!(
181 s.code(),
182 Code::Unavailable
183 | Code::DeadlineExceeded
184 | Code::Aborted
185 | Code::Cancelled
186 | Code::ResourceExhausted
187 ),
188 _ => false,
189 }
190 }
191}
192
193impl From<tonic::Status> for StreamError {
194 fn from(s: tonic::Status) -> Self {
195 StreamError::from_status(s)
196 }
197}
198
199impl From<tonic::transport::Error> for StreamError {
200 fn from(e: tonic::transport::Error) -> Self {
201 StreamError::Connect(e)
202 }
203}