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