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
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
use alloc::string::String;
/// Error taxonomy returned by the Rust SDK API surface.
///
/// The SDK keeps application code independent from transport, protocol framing,
/// and core implementation error types. Concrete embedded and remote adapters
/// map their internal failures into these variants.
///
/// `Clone` is derived so a failure can be both RETAINED and RETURNED at the same
/// seam. A store failure that bricks a participant handle is exactly that shape:
/// the error travels to the caller of the failing operation, while a copy stays
/// behind as the reason every later call reports
/// `RemoteParticipantError::StateUnavailable`. Without it the cause is
/// observable exactly once and only by whoever happened to make the failing
/// call.
#[derive(Clone, PartialEq, Eq, thiserror::Error)]
pub enum SdkError {
/// A routed worker refused. The code is the worker's closed vocabulary;
/// words remain application data and are omitted from `Debug`.
#[error("{words}")]
WorkerRefused {
/// One of busy, invalid, unavailable or internal.
code: String,
/// The worker's exact words, bounded by the wire contract.
words: String,
},
/// A receiver handle already owns reception, or a direct receive is active.
#[error("push reception is already claimed")]
ReceiverAlreadyClaimed,
/// A `ReadWake` ended a budgeted participant read before a frame arrived;
/// the transport remains usable. See `ReadWake` for which reads it reaches.
#[error("idle receive interrupted by its owner")]
ReadInterrupted,
/// Establishing, keeping, or recovering a client connection failed.
#[error("connection error: {description}")]
Connection {
/// Human-readable context from the failing connection operation.
description: String,
},
/// Encoding, decoding, or interpreting SDK-internal protocol state failed.
#[error("protocol error: {description}")]
Protocol {
/// Human-readable protocol failure context.
description: String,
},
/// Serialising an outbound value or deserialising an inbound value failed.
#[error("serialization error: {description}")]
Serialization {
/// Human-readable serialization failure context.
description: String,
},
/// A typed message did not satisfy the schema declared for its channel.
#[error("type validation failed: {description}")]
TypeValidation {
/// Human-readable type-validation context.
description: String,
},
/// A publish operation encountered application-visible backpressure.
#[error("backpressure: {reason}")]
Backpressure {
/// Human-readable pressure reason.
reason: String,
},
/// A publish whose outcome the server could not resolve: it cannot tell
/// whether the message was stored.
///
/// Decoded from the reply's reason code alone, never from its text. It is
/// neither a rejection nor backpressure: the publish may have been stored,
/// so the safe response is a retry carrying the SAME idempotency key, which
/// answers the original outcome if the row committed.
#[error("publish unresolved: {description}")]
PublishUnresolved {
/// The server's detail for the unresolved publish, verbatim.
description: String,
},
/// A conversation operation failed.
#[error("conversation {conversation_id}: {description}")]
Conversation {
/// Application-visible conversation identifier.
conversation_id: String,
/// Human-readable conversation failure context.
description: String,
},
/// Persisted subscription or recovery state could not be read or written.
#[error("store error: {description}")]
Store {
/// Human-readable store failure context.
description: String,
},
/// A remote operation ran against a configuration whose transport was never
/// connected, so nothing it did could have reached any wire.
///
/// `RemoteConfig::new` installs the unconnected in-process transport. Call
/// `RemoteConfig::connect_tcp`, `RemoteConfig::connect_tcp_with_auth`,
/// `RemoteConfig::connect_websocket`,
/// `RemoteConfig::connect_websocket_with_auth`, or (with the `embedded`
/// feature) `RemoteConfig::connect_loopback` before publishing,
/// subscribing, conversing, or resuming. The SDK refuses here rather than
/// encoding a frame it discards and reporting the success that would imply.
#[error(
"transport not connected: {operation} needs a connected transport; call \
RemoteConfig::connect_tcp or RemoteConfig::connect_websocket first"
)]
NotConnected {
/// The SDK operation that refused.
operation: String,
},
/// An SDK surface exists but has no machinery behind it, so it refused
/// rather than reporting a success it cannot deliver.
///
/// This is the SDK being loud about an absent feature. `seam` records where
/// the gap is written down (a ledger row, or that nothing records it) and
/// `alternative` names the surface that does work today.
#[error("{surface} is not wired ({seam}); {alternative}")]
Unwired {
/// The SDK surface that refused.
surface: String,
/// Where the gap is recorded, or that nothing records it.
seam: String,
/// The working surface a caller should reach for instead.
alternative: String,
},
}
impl SdkError {
/// Classifies the wire reason before retaining any application words.
#[cfg(feature = "std")]
pub(crate) fn conversation_refusal(
conversation_id: u64,
reason_code: u16,
message: Option<String>,
) -> Self {
use alloc::{format, string::ToString};
use liminal::protocol::{WorkerRefusal, WorkerRefusalCode};
if let Some(code) = WorkerRefusalCode::from_reason_code(reason_code) {
let Some(words) = message else {
return Self::Protocol {
description: "worker_refusal_missing_words".to_string(),
};
};
return match WorkerRefusal::new(code.name(), &words) {
Ok(_) => Self::WorkerRefused {
code: code.name().to_string(),
words,
},
Err(error) => Self::Protocol {
description: error
.message()
.unwrap_or("worker_refusal_invalid_record")
.to_string(),
},
};
}
Self::Conversation {
conversation_id: conversation_id.to_string(),
description: format!(
"server rejected conversation {conversation_id} (reason {reason_code}): {}",
message.unwrap_or_else(|| "no detail".to_string())
),
}
}
}
impl core::fmt::Debug for SdkError {
fn fmt(&self, formatter: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
match self {
Self::WorkerRefused { code, words } => formatter
.debug_struct("WorkerRefused")
.field("code", code)
.field("words_bytes", &words.len())
.finish(),
Self::ReceiverAlreadyClaimed => formatter.write_str("ReceiverAlreadyClaimed"),
Self::ReadInterrupted => formatter.write_str("ReadInterrupted"),
Self::Connection { description } => formatter
.debug_struct("Connection")
.field("description_bytes", &description.len())
.finish(),
Self::Protocol { description } => formatter
.debug_struct("Protocol")
.field("description_bytes", &description.len())
.finish(),
Self::Serialization { description } => formatter
.debug_struct("Serialization")
.field("description_bytes", &description.len())
.finish(),
Self::TypeValidation { description } => formatter
.debug_struct("TypeValidation")
.field("description_bytes", &description.len())
.finish(),
Self::Backpressure { reason } => formatter
.debug_struct("Backpressure")
.field("reason_bytes", &reason.len())
.finish(),
Self::PublishUnresolved { description } => formatter
.debug_struct("PublishUnresolved")
.field("description_bytes", &description.len())
.finish(),
Self::Conversation {
conversation_id,
description,
} => formatter
.debug_struct("Conversation")
.field("conversation_id", conversation_id)
.field("description_bytes", &description.len())
.finish(),
Self::Store { description } => formatter
.debug_struct("Store")
.field("description_bytes", &description.len())
.finish(),
Self::NotConnected { operation } => formatter
.debug_struct("NotConnected")
.field("operation", operation)
.finish(),
Self::Unwired {
surface,
seam,
alternative,
} => formatter
.debug_struct("Unwired")
.field("surface", surface)
.field("seam", seam)
.field("alternative", alternative)
.finish(),
}
}
}