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