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
use std::error::Error;
use std::fmt;
/// Stable stream terminal codes emitted through netbat `SUB_ERR` / `SUB_END`.
pub mod stream_code {
/// Subscription id is not registered.
pub const UNKNOWN_SUBSCRIPTION: &str = "unknown_subscription";
/// Resume cursor bytes failed decode or semantic checks.
pub const CURSOR_INVALID: &str = "cursor_invalid";
/// Resume cursor does not match subscription/category binding.
pub const CURSOR_MISMATCH: &str = "cursor_mismatch";
/// Client exceeded bounded delivery window without cumulative ACK.
pub const SLOW_CONSUMER: &str = "slow_consumer";
/// Post-subscribe control frame was malformed.
pub const MALFORMED_STREAM_FRAME: &str = "malformed_stream_frame";
/// Client cancelled the subscription stream.
pub const CLIENT_CANCELLED: &str = "client_cancelled";
/// A persisted syncbat receipt event failed canonical decode.
pub const RECEIPT_DECODE_FAILED: &str = "receipt_decode_failed";
/// A server-internal fault ended the stream (the store-backed watcher bridge
/// failed or disconnected). Distinct from the client/protocol codes above so
/// a peer can tell "the server failed" from "your request was bad" — routing
/// an internal fault through a cursor/protocol code would misreport a
/// server-side failure as a client error.
pub const INTERNAL: &str = "internal_error";
}
/// Error returned by the subscription runtime engine.
#[derive(Debug)]
pub enum SubscriptionRuntimeError {
/// Subscription id grammar or length is invalid.
InvalidSubscriptionId {
/// Stable reason token.
reason: &'static str,
},
/// Subscription id is already present in the registry.
DuplicateSubscription {
/// Duplicated subscription id.
id: String,
},
/// Subscription route declaration is invalid.
InvalidRoute {
/// Stable reason token.
reason: &'static str,
},
/// Runtime configuration cannot support a valid stream.
InvalidConfig {
/// Stable reason token.
reason: &'static str,
},
/// Subscription id is not present in the registry.
UnknownSubscription {
/// Requested subscription id.
id: String,
},
/// Resume cursor decode or validation failed.
CursorInvalid {
/// Stable reason token.
reason: &'static str,
},
/// Resume cursor does not match the subscription route binding.
CursorMismatch {
/// Stable reason token.
reason: &'static str,
},
/// Store read or envelope encoding failed.
Store(batpak::store::StoreError),
/// Canonical envelope encoding failed.
EnvelopeEncoding(String),
/// Runtime worker thread could not be started or stopped unexpectedly.
Worker(String),
/// Cumulative ACK referenced an unknown delivery index or cursor.
AckInvalid {
/// Stable reason token.
reason: &'static str,
},
}
impl fmt::Display for SubscriptionRuntimeError {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::InvalidSubscriptionId { reason } => {
write!(f, "invalid subscription id: {reason}")
}
Self::DuplicateSubscription { id } => {
write!(f, "duplicate subscription route: {id}")
}
Self::InvalidRoute { reason } => write!(f, "invalid subscription route: {reason}"),
Self::InvalidConfig { reason } => write!(f, "invalid subscription config: {reason}"),
Self::UnknownSubscription { id } => write!(f, "unknown subscription: {id}"),
Self::CursorInvalid { reason } => write!(f, "cursor invalid: {reason}"),
Self::CursorMismatch { reason } => write!(f, "cursor mismatch: {reason}"),
Self::Store(error) => write!(f, "store error: {error}"),
Self::EnvelopeEncoding(detail) => write!(f, "envelope encoding failed: {detail}"),
Self::Worker(detail) => write!(f, "subscription worker failed: {detail}"),
Self::AckInvalid { reason } => write!(f, "ack invalid: {reason}"),
}
}
}
impl Error for SubscriptionRuntimeError {}
impl From<batpak::store::StoreError> for SubscriptionRuntimeError {
fn from(error: batpak::store::StoreError) -> Self {
Self::Store(error)
}
}