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
use std::net::SocketAddr;
use crate::server::participant::ParticipantServiceFatal;
/// Error taxonomy for standalone liminal server deployment failures.
#[derive(Debug, thiserror::Error)]
pub enum ServerError {
/// The configuration file could not be read or parsed.
#[error("configuration load failed: {message}")]
ConfigLoad { message: String },
/// The configuration file was read but failed semantic validation.
#[error("configuration validation failed: {message}")]
ConfigValidation { message: String },
/// The server could not bind its configured listener address.
#[error("listener bind failed for {address}: {source}")]
ListenerBind {
/// Address the server attempted to bind.
address: SocketAddr,
/// Underlying operating-system bind failure.
#[source]
source: std::io::Error,
},
/// The server listener failed while accepting an inbound connection.
#[error("listener accept failed: {message}")]
ListenerAccept { message: String },
/// Durable participant-incarnation startup or allocation failed before its
/// result could be published.
#[error("participant incarnation {phase} failed: {message}")]
ParticipantIncarnation {
/// Exact server seam that failed.
phase: &'static str,
/// Underlying durable or bounded-bridge diagnostic.
message: String,
},
/// A durable connection-fate Open could not be completed in this process.
///
/// The normal runtime shutdown path returns this typed fatal only after it has
/// stopped both listeners and run the ordinary connection drain/flush sequence.
#[error("participant service fatal: {fatal}")]
ParticipantServiceFatal {
/// First process-wide post-Open failure; later failures cannot replace it.
fatal: ParticipantServiceFatal,
},
/// Startup found durable connection-fate work that the Decision A/C producer must complete
/// before listener or semantic-service publication.
#[error(
"participant startup requires recovery of {open_count} connection-fate Opens beginning at sequence {first_open_sequence}"
)]
ConnectionFateRecoveryRequired {
/// Number of replay-validated unmatched Opens.
open_count: usize,
/// Lowest unmatched Open sequence.
first_open_sequence: u64,
},
/// The durable server-incarnation namespace has no successor.
#[error("participant server-incarnation namespace is exhausted")]
ServerIncarnationExhausted,
/// Production participant startup restore failed: the durable
/// conversation streams could not be scanned or replayed, so the
/// server-scope capacity ledger cannot be made exact and the server
/// refuses to start over state it cannot account for.
#[error("participant startup restore failed: {message}")]
ParticipantStartupRestore {
/// Underlying durable, replay, or bridge diagnostic.
message: String,
},
/// The current durable server incarnation has no collision-free connection
/// ordinal left, so the accepted socket was not admitted.
#[error(
"connection incarnation exhausted for server incarnation {attempted_server_incarnation}"
)]
ConnectionIncarnationExhausted {
/// Server incarnation whose complete ordinal suffix was examined.
attempted_server_incarnation: u64,
},
/// A connection registration found a record already present under its pid.
/// Enforces the pids-fresh-per-spawn supervision invariant loudly: a silent
/// replace would drop the displaced record's teardown-held fd guard outside
/// the single record-removal funnel (orphaning a live connection's stream)
/// and increment the `liminal_connections_active` gauge a second time with
/// no paired decrement. The whole registration is refused instead, leaving
/// the prior record — and its teardown route — intact.
#[error("connection registration refused: pid {pid} already has a live registry record")]
ConnectionPidCollision {
/// The connection pid that already owned a registry record.
pid: u64,
},
/// A frame requested an operation the configured services profile does not
/// serve (e.g. ordinary publish/subscribe/conversation traffic against the
/// capability-scoped worker front door). Server-internal taxonomy, not wire
/// vocabulary: the connection process renders it as the operation's existing
/// typed error frame with this error's text as the message.
#[error("{operation} is not supported by the {profile} services profile")]
UnsupportedOperation {
/// Human-readable description of the refused operation.
operation: String,
/// The configured services profile that refused it.
profile: &'static str,
},
/// A server→client push reply slot was dropped before a correlated reply
/// arrived — the connection closed (the prompt worker-death signal). Distinct
/// from [`Self::PushReplyTimeout`] so consumers can tell a worker that DIED
/// (fast failover) from one that is merely SLOW, by type rather than message.
#[error(
"push correlation {correlation_id} did not complete: the connection closed before sending a correlated push reply"
)]
PushReplyDisconnected {
/// Correlation id of the push whose reply will never arrive.
correlation_id: u64,
},
/// A server→client push reply did not arrive within the awaiter's timeout —
/// the worker is still connected but did not reply in this wait quantum. This
/// is BENIGN: the reply slot survives untouched and the caller may re-arm
/// [`PushReplyAwaiter::receive`](crate::server::connection::PushReplyAwaiter::receive)
/// indefinitely. It is not a worker-death signal.
#[error(
"push correlation {correlation_id} did not complete: no correlated push reply arrived within the timeout"
)]
PushReplyTimeout {
/// Correlation id of the push whose wait quantum elapsed.
correlation_id: u64,
},
/// A server→client push carrying an explicit reply deadline (via
/// [`push_to_connection_with_deadline`](crate::server::connection::ConnectionSupervisor::push_to_connection_with_deadline))
/// reached that deadline before a correlated reply arrived. Unlike
/// [`Self::PushReplyTimeout`] this is TERMINAL: the reply slot has been
/// removed and its §5 `max_pending_pushes_per_connection` cap admission
/// released. Returned PROMPTLY once the deadline is due — a `receive` call
/// does not hold a due expiry until its caller's quantum ends, so the
/// terminal outcome is independent of how the caller polls. Distinct
/// variant so callers classify by type, not message.
#[error(
"push correlation {correlation_id} did not complete: its reply deadline passed before a correlated push reply arrived"
)]
PushReplyExpired {
/// Correlation id of the push whose reply deadline passed.
correlation_id: u64,
},
/// A server→client push was never queued for the wire: its encoded frame is
/// larger than the WHOLE of the connection's bounded outbound buffer, so no
/// amount of draining could ever make room for it. TERMINAL and CERTAIN — the
/// reply slot is removed, its §5 `max_pending_pushes_per_connection` admission
/// released, and the client never saw a `Push` frame for this correlation id.
///
/// Deliberately NARROW. A frame that would fit an empty buffer but not the
/// current free space is CONGESTION, not a size defect: it keeps the
/// connection-teardown path and its retryable
/// [`Self::PushReplyDisconnected`] reading, because waiting is the correct
/// answer there. Sharing one outcome between the certain case and the
/// recoverable one would leave the caller unable to tell them apart — which
/// is the exact fault this variant exists to end. Distinct variant so callers
/// classify by type, not message.
///
/// The WORDS carry the bound's value and say the bound was chosen, because a
/// reader who meets this without that context blames the server for a number
/// an operator set. Since every outbound bound now comes from
/// [`max_connection_outbound_bytes`](crate::config::types::LimitsConfig::max_connection_outbound_bytes),
/// that sentence is true on every occurrence rather than usually.
#[error(
"push correlation {correlation_id} was never sent: its {needed}-byte frame exceeds this \
connection's whole outbound buffer of {capacity} bytes, which is operator-set by \
`limits.max_connection_outbound_bytes`"
)]
PushFrameExceedsOutboundCapacity {
/// Correlation id of the push that could not be queued.
correlation_id: u64,
/// Encoded size of the rejected frame. Exceeds `capacity` on its own.
needed: usize,
/// The outbound buffer's whole capacity, which `needed` exceeds.
capacity: usize,
/// Bytes already queued when the frame was rejected. Context for the log
/// only: the decision is `needed` against `capacity` alone, so this frame
/// would not have fitted an empty buffer either.
queued: usize,
},
/// The server could not join the configured beamr distribution cluster.
#[error("cluster join failed: {message}")]
ClusterJoin { message: String },
/// Cluster state propagation through beamr distribution failed.
#[error("cluster sync failed: {message}")]
ClusterSync { message: String },
/// Graceful shutdown did not drain within the configured timeout.
#[error("shutdown timed out: {message}")]
ShutdownTimeout { message: String },
/// Durable state could not be flushed during graceful shutdown.
#[error("shutdown flush failed: {message}")]
ShutdownFlush { message: String },
/// The health endpoint failed to start or serve requests.
#[error("health endpoint failed: {message}")]
HealthEndpoint { message: String },
/// A new connection was refused because the configured `max_connections`
/// cap (§5) is already reached. The listener drops the freshly accepted
/// stream, refusing the connection, rather than admitting an unbounded fleet.
#[error(
"connection refused: the configured max_connections limit of {limit} live connections is reached"
)]
ConnectionLimitReached {
/// The configured `max_connections` cap that was hit.
limit: usize,
},
/// An admission-time per-connection cap (§5) refused an operation: too many
/// subscriptions, conversations, in-flight pushes, or pending conversation
/// replies on one connection. The connection process renders it as the
/// operation's existing typed error frame with this text as the message.
#[error("{operation} refused: the per-connection {cap} limit of {limit} is reached")]
ConnectionCapReached {
/// Human-readable description of the refused operation.
operation: String,
/// The name of the cap that refused it (a `limits.*` config key).
cap: &'static str,
/// The configured cap value that was hit.
limit: usize,
},
}