ruststream_sqlx/inbox/error.rs
1//! The inbox broker's one error.
2
3use ruststream_sqlx_dialect::StatementError;
4use thiserror::Error;
5
6/// What can go wrong in the inbox broker.
7///
8/// Every variant about a subscription names the subscription, the table and the row type, so one
9/// line of a log says where to look.
10///
11/// # Examples
12///
13/// ```
14/// # #[cfg(feature = "postgres")]
15/// # mod demo {
16/// use ruststream::runtime::PublishError;
17/// use ruststream_sqlx::prelude::*;
18/// use ruststream_sqlx::{RepositoryPublisher, SqlxBrokerError};
19/// use serde::Serialize;
20/// use sqlx::Postgres;
21/// # use ruststream::OutgoingMessage;
22/// # use sqlx::PgConnection;
23/// # #[derive(Inbox, sqlx::FromRow)]
24/// # #[inbox(table = "export_jobs")]
25/// # pub struct ExportJob { #[field(id, generated)] id: i64, #[field(payload)] payload: Vec<u8> }
26/// # impl Publish<Postgres> for ExportJob {
27/// # async fn publish(_: &mut PgConnection, _: &OutgoingMessage<'_>) -> Result<(), sqlx::Error> { Ok(()) }
28/// # }
29///
30/// #[derive(Serialize, Outgoing)]
31/// #[outgoing(name = "exports")]
32/// pub struct Export {
33/// account: u64,
34/// }
35///
36/// /// The status the service's HTTP endpoint answers when it schedules an export.
37/// pub async fn schedule(exports: &RepositoryPublisher<Postgres, ExportJob>, export: &Export) -> u16 {
38/// match exports.message(export).publish().await {
39/// Ok(()) => 202,
40/// // The service is shutting down: the client tries again later.
41/// Err(PublishError::Publish(SqlxBrokerError::Closed)) => 503,
42/// Err(error) => {
43/// tracing::error!(%error, "the export was not scheduled");
44/// 500
45/// }
46/// }
47/// }
48/// # }
49/// # fn main() {}
50/// ```
51#[derive(Debug, Error)]
52#[non_exhaustive]
53pub enum SqlxBrokerError {
54 /// `connect` could not take a connection from the pool.
55 #[error("the inbox broker could not take a connection from the pool: {source}")]
56 Connect {
57 /// The pool's error.
58 #[source]
59 source: sqlx::Error,
60 },
61 /// `connect` found the pool on a database no built-in dialect serves: an `AnyPool`'s backend
62 /// whose feature is off.
63 #[error(
64 "no built-in dialect serves the `{backend}` backend of this `AnyPool`: enable its feature, \
65 or pass a dialect with `SqlxBroker::with_dialect`"
66 )]
67 Backend {
68 /// The backend's name, as its sqlx driver reports it.
69 backend: String,
70 },
71 /// A statement failed while the subscription ran.
72 #[error(
73 "subscription `{subscription}` on table `{table}` ({row}): `{statement}` failed: {source}"
74 )]
75 Sqlx {
76 /// The subscription.
77 subscription: String,
78 /// The table, qualified with its schema.
79 table: String,
80 /// The row type.
81 row: &'static str,
82 /// The statement, or the event a service's own impl ran.
83 statement: &'static str,
84 /// The database's error.
85 #[source]
86 source: Box<sqlx::Error>,
87 },
88 /// A statement failed to prepare at startup: the table does not match the struct.
89 ///
90 /// Preparing checks the names of the table and its columns, not the column types; a claimed
91 /// row whose columns do not decode goes to the subscription's decode-failure policy instead.
92 #[error(
93 "subscription `{subscription}` on table `{table}` ({row}): the database refused \
94 `{statement}`: {source}"
95 )]
96 Schema {
97 /// The subscription.
98 subscription: String,
99 /// The table, qualified with its schema.
100 table: String,
101 /// The row type.
102 row: &'static str,
103 /// The statement that failed to prepare.
104 statement: &'static str,
105 /// The database's error.
106 #[source]
107 source: Box<sqlx::Error>,
108 },
109 /// The dialect refused to build a statement for the table.
110 #[error("subscription `{subscription}` on table `{table}` ({row}): {source}")]
111 Dialect {
112 /// The subscription.
113 subscription: String,
114 /// The table, qualified with its schema.
115 table: String,
116 /// The row type.
117 row: &'static str,
118 /// What the dialect could not build.
119 #[source]
120 source: StatementError,
121 },
122 /// The server is older than the statements of the subscription's form need.
123 ///
124 /// A subscription reads the server's version when it opens, where its dialect asks for it:
125 /// MySQL claims rows with `SKIP LOCKED` in the row lock and lease forms, which MySQL 8.0.1 and
126 /// MariaDB 10.6 added, and a table in the advisory lock form shares that floor.
127 #[error(
128 "subscription `{subscription}` on table `{table}` ({row}): the server reports \
129 `{server}`, and this form needs {required} or later"
130 )]
131 ServerTooOld {
132 /// The subscription.
133 subscription: String,
134 /// The table, qualified with its schema.
135 table: String,
136 /// The row type.
137 row: &'static str,
138 /// The version the server reports.
139 server: String,
140 /// The oldest server the subscription's statements run on.
141 required: &'static str,
142 },
143 /// The registration declared what the table cannot carry: a retry, or transactional mode at an
144 /// isolation level its form cannot keep it at.
145 #[error("subscription `{subscription}` on table `{table}` ({row}): {reason}")]
146 Declaration {
147 /// The subscription.
148 subscription: String,
149 /// The table, qualified with its schema.
150 table: String,
151 /// The row type.
152 row: &'static str,
153 /// What to change.
154 reason: String,
155 },
156 /// A subscription on dedicated threads whose connections would take the service past the
157 /// broker's [`connection_limit`](crate::SqlxBroker::connection_limit): the subscription does
158 /// not open, and the service does not start.
159 #[error(
160 "subscription `{subscription}`: the pool's {pool} connections and the dedicated threads' \
161 {threads} pass the connection limit of {limit}; lower the threads' connections or raise \
162 the limit"
163 )]
164 ConnectionLimit {
165 /// The subscription that would pass the limit.
166 subscription: String,
167 /// The limit the broker sets.
168 limit: u32,
169 /// The size of the service's pool.
170 pool: u64,
171 /// The connections of the dedicated threads of every subscription so far, this one's
172 /// included.
173 threads: u64,
174 },
175 /// The connection already reads this queue: two subscriptions would compete for its rows.
176 #[error(
177 "subscription `{subscription}` on table `{table}` ({row}) is open already on this \
178 connection: one queue has one subscription per process; scale it with `workers(n)`"
179 )]
180 AlreadySubscribed {
181 /// The subscription.
182 subscription: String,
183 /// The table, qualified with its schema.
184 table: String,
185 /// The row type.
186 row: &'static str,
187 },
188 /// A settlement of a batch whose transaction an earlier settlement's failure rolls back.
189 ///
190 /// A batch that holds a row whose statement always fails rolls back and returns all of its
191 /// rows on every attempt, until the service's SQL or schema is fixed.
192 #[error(
193 "subscription `{subscription}` on table `{table}` ({row}): an earlier settlement of this \
194 batch failed, so the batch rolls back and its rows return to the queue"
195 )]
196 BatchRolledBack {
197 /// The subscription.
198 subscription: String,
199 /// The table, qualified with its schema.
200 table: String,
201 /// The row type.
202 row: &'static str,
203 },
204 /// A settlement in the lease form found its row under another lease, so it took no effect.
205 ///
206 /// Ownership of a leased row ends with its lease, not with its holder. The subscription
207 /// extends the lease of every delivery in work each half lease, so a lease runs out under a
208 /// running handler only when its extensions fail or stop (the database out of reach, the
209 /// subscription closed or the broker shut down). The handler then shares the row with the
210 /// next claim, and the lease, the ownership token, only stops the late holder from settling.
211 ///
212 /// A settlement whose statement fails keeps the row under its lease until the lease runs
213 /// out, as a crash does; a delivery dropped unsettled releases its row at once.
214 #[error(
215 "subscription `{subscription}` on table `{table}` ({row}): the lease on row {id} ran out \
216 and another claim took it; this settlement did not take effect"
217 )]
218 LeaseLost {
219 /// The subscription.
220 subscription: String,
221 /// The table, qualified with its schema.
222 table: String,
223 /// The row type.
224 row: &'static str,
225 /// The row's id, as logs name it.
226 id: String,
227 },
228 /// A settlement in transactional mode found the delivery's transaction still lent to its
229 /// handler's [`Tx`](crate::Tx), which the handler moved somewhere that outlived it.
230 ///
231 /// The settlement took no effect. The transaction ends when that `Tx` drops: its connection
232 /// closes, the server rolls back what the handler wrote, and the row returns to the queue. In
233 /// the lease form the row returns once its lease runs out, which the broker extends no more.
234 #[error(
235 "subscription `{subscription}` on table `{table}` ({row}): the handler still holds the \
236 transaction of row {id}, so its settlement took no effect; the transaction rolls back \
237 when the handler's `Tx` drops"
238 )]
239 TransactionHeld {
240 /// The subscription.
241 subscription: String,
242 /// The table, qualified with its schema.
243 table: String,
244 /// The row type.
245 row: &'static str,
246 /// The row's id, as logs name it.
247 id: String,
248 },
249 /// The broker with `listen_notify` could not listen on the channel of the subscription's
250 /// table as it opened.
251 #[error(
252 "subscription `{subscription}` on table `{table}` ({row}): the broker could not listen \
253 for the table's notifications: {source}"
254 )]
255 Listen {
256 /// The subscription.
257 subscription: String,
258 /// The table, qualified with its schema: the channel.
259 table: String,
260 /// The row type.
261 row: &'static str,
262 /// The database's error.
263 #[source]
264 source: Box<sqlx::Error>,
265 },
266 /// The service's `Publish` failed.
267 #[error("publishing to `{name}` into table `{table}` ({row}) failed: {source}")]
268 Publish {
269 /// The name published to.
270 name: String,
271 /// The table, qualified with its schema.
272 table: String,
273 /// The row type.
274 row: &'static str,
275 /// The database's error.
276 #[source]
277 source: Box<sqlx::Error>,
278 },
279 /// The publish carries a header the table cannot hold byte for byte, so it was refused.
280 #[error(
281 "publishing to `{name}` into table `{table}` ({row}) was refused: the table cannot hold \
282 header `{header}` byte for byte"
283 )]
284 Header {
285 /// The name published to.
286 name: String,
287 /// The table, qualified with its schema.
288 table: String,
289 /// The row type.
290 row: &'static str,
291 /// The header.
292 header: String,
293 },
294 /// No route leads the name to a table.
295 #[error("no route leads `{name}` to a table: add `.route::<Row>(\"{name}\")` to the broker")]
296 NoRoute {
297 /// The name published to.
298 name: String,
299 },
300 /// The broker was shut down; the handle outlived it.
301 #[error("the inbox broker is shut down")]
302 Closed,
303}
304
305#[cfg(test)]
306mod tests {
307 use super::SqlxBrokerError;
308
309 #[test]
310 fn every_message_names_where_to_look() {
311 let error = SqlxBrokerError::AlreadySubscribed {
312 subscription: "emails".to_owned(),
313 table: "app.email_jobs".to_owned(),
314 row: "SendEmail",
315 };
316 assert_eq!(
317 error.to_string(),
318 "subscription `emails` on table `app.email_jobs` (SendEmail) is open already on this \
319 connection: one queue has one subscription per process; scale it with `workers(n)`"
320 );
321 assert_eq!(
322 SqlxBrokerError::Closed.to_string(),
323 "the inbox broker is shut down"
324 );
325 let backend = SqlxBrokerError::Backend {
326 backend: "MySQL".to_owned(),
327 };
328 assert_eq!(
329 backend.to_string(),
330 "no built-in dialect serves the `MySQL` backend of this `AnyPool`: enable its feature, \
331 or pass a dialect with `SqlxBroker::with_dialect`"
332 );
333 let lost = SqlxBrokerError::LeaseLost {
334 subscription: "emails".to_owned(),
335 table: "email_jobs".to_owned(),
336 row: "SendEmail",
337 id: "7".to_owned(),
338 };
339 assert_eq!(
340 lost.to_string(),
341 "subscription `emails` on table `email_jobs` (SendEmail): the lease on row 7 ran out \
342 and another claim took it; this settlement did not take effect"
343 );
344 let held = SqlxBrokerError::TransactionHeld {
345 subscription: "emails".to_owned(),
346 table: "email_jobs".to_owned(),
347 row: "SendEmail",
348 id: "7".to_owned(),
349 };
350 assert_eq!(
351 held.to_string(),
352 "subscription `emails` on table `email_jobs` (SendEmail): the handler still holds the \
353 transaction of row 7, so its settlement took no effect; the transaction rolls back \
354 when the handler's `Tx` drops"
355 );
356 let unheard = SqlxBrokerError::Listen {
357 subscription: "emails".to_owned(),
358 table: "email_jobs".to_owned(),
359 row: "SendEmail",
360 source: Box::new(sqlx::Error::PoolClosed),
361 };
362 assert_eq!(
363 unheard.to_string(),
364 "subscription `emails` on table `email_jobs` (SendEmail): the broker could not listen \
365 for the table's notifications: attempted to acquire a connection on a closed pool"
366 );
367 let old = SqlxBrokerError::ServerTooOld {
368 subscription: "emails".to_owned(),
369 table: "email_jobs".to_owned(),
370 row: "SendEmail",
371 server: "10.5.23-MariaDB".to_owned(),
372 required: "MariaDB 10.6",
373 };
374 assert_eq!(
375 old.to_string(),
376 "subscription `emails` on table `email_jobs` (SendEmail): the server reports \
377 `10.5.23-MariaDB`, and this form needs MariaDB 10.6 or later"
378 );
379 }
380}