ruststream_sqlx/outbox/error.rs
1//! What can go wrong while the outbox tracks a message.
2
3use std::error::Error as StdError;
4
5use sqlx::Error as SqlError;
6use thiserror::Error;
7
8/// A failure of the outbox: a record it could not write or read, or a republish that did not
9/// reach the broker.
10///
11/// # Examples
12///
13/// ```
14/// # #[cfg(all(feature = "outbox", feature = "postgres"))]
15/// # mod demo {
16/// # use ruststream::OutgoingMessage;
17/// # use ruststream::memory::prelude::*;
18/// # use ruststream_sqlx::{Outbox, outbox};
19/// # use serde::{Deserialize, Serialize};
20/// # use sqlx::postgres::{PgConnection, PgPool, Postgres};
21/// # #[derive(Outbox, sqlx::FromRow)]
22/// # #[outbox(table = "outbox")]
23/// # pub struct OrderOutbox { #[field(id)] id: i64, #[field(name)] name: String, #[field(payload)] payload: Vec<u8> }
24/// # impl outbox::Publish<Postgres> for OrderOutbox {
25/// # async fn publish(conn: &mut PgConnection, msg: &OutgoingMessage<'_>) -> sqlx::Result<i64> {
26/// # sqlx::query_scalar("INSERT INTO outbox (name, payload) VALUES ($1, $2) RETURNING id")
27/// # .bind(msg.name()).bind(msg.payload()).fetch_one(conn).await
28/// # }
29/// # }
30/// # #[derive(Deserialize)] pub struct PlaceOrder { id: u64 }
31/// # #[derive(Serialize, Deserialize, Outgoing)] #[outgoing(name = "orders")] pub struct OrderPlaced { id: u64 }
32/// # #[subscriber("checkout", reply)] async fn place(cmd: &PlaceOrder) -> OrderPlaced { OrderPlaced { id: cmd.id } }
33/// # #[subscriber("orders")] async fn fulfil(_: &OrderPlaced) -> HandlerOutcome { HandlerOutcome::ack() }
34/// use ruststream::runtime::PublishError;
35/// use ruststream_sqlx::outbox::{OutboxError, TrackedPublishError};
36///
37/// pub async fn serve(pool: PgPool) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
38/// let tracking = outbox! { pool: pool, "orders" => OrderOutbox };
39/// let broker = MemoryBroker::new().bindable();
40/// let egress = broker.bind(Publish);
41/// let running = RustStream::new(AppInfo::new("orders", "0.1.0"))
42/// .with_broker(broker, |b| {
43/// b.include(fulfil);
44/// })
45/// .start()
46/// .await?;
47/// let publisher = tracking.wrap(running.publisher(egress).await?);
48///
49/// // The status an HTTP endpoint answers an order with.
50/// let status = match publisher.message(&OrderPlaced { id: 7 }).publish().await {
51/// Ok(()) => 202,
52/// // Recorded but not sent: the next startup sends it.
53/// Err(PublishError::Publish(TrackedPublishError::Publish(_))) => 202,
54/// // Neither recorded nor sent: the client sends the order again.
55/// Err(PublishError::Publish(TrackedPublishError::Outbox(OutboxError::Record { .. }))) => 503,
56/// Err(_) => 500,
57/// };
58/// tracing::info!(status, "order answered");
59///
60/// running.shutdown().await?;
61/// Ok(())
62/// }
63/// # }
64/// # fn main() {}
65/// ```
66#[derive(Debug, Error)]
67#[non_exhaustive]
68pub enum OutboxError {
69 /// A message under a registered name was published before the outbox had a pool.
70 #[error(
71 "the outbox has no pool, so a message under `{name}` cannot be recorded: give it one with \
72 `Outbox::new` or with `set_pool` in `on_startup`"
73 )]
74 NoPool {
75 /// The registered name.
76 name: &'static str,
77 },
78 /// The record of a published message was not written; the message was not sent.
79 #[error("the outbox did not record a message published under `{name}` in `{record}`")]
80 Record {
81 /// The registered name.
82 name: &'static str,
83 /// The record type.
84 record: &'static str,
85 /// The database's error.
86 #[source]
87 source: SqlError,
88 },
89 /// A transactional handler on SQLite published a message under a registered name while its
90 /// own delivery transaction held the database's write lock, which the record needs; the
91 /// message was not sent.
92 ///
93 /// The transaction holds the lock from its first write through `Tx`, or from its begin in an
94 /// immediate or exclusive mode, until the delivery settles. A transaction that took the lock as
95 /// it began fails the publish at once. After a statement through `Tx` the record waits for the
96 /// lock at most the pool's acquire timeout, since SQLite tells whether that statement wrote
97 /// only by the wait. Such a handler publishes its tracked messages before its first write.
98 #[error(
99 "the outbox did not record a message published under `{name}` in `{record}`: the \
100 handler's own delivery transaction holds SQLite's write lock, from its first write \
101 through `Tx` (or its begin in an immediate or exclusive mode) until the delivery \
102 settles; publish tracked messages before that write"
103 )]
104 WriteLockHeld {
105 /// The registered name.
106 name: &'static str,
107 /// The record type.
108 record: &'static str,
109 },
110 /// The unprocessed records of a name were not read at startup.
111 #[error("the outbox did not read the unprocessed records of `{name}` from `{record}`")]
112 Recover {
113 /// The registered name.
114 name: &'static str,
115 /// The record type.
116 record: &'static str,
117 /// The database's error.
118 #[source]
119 source: SqlError,
120 },
121 /// A record read at startup was not published again.
122 #[error("the outbox did not publish record {id} of `{name}` in `{record}` again")]
123 Republish {
124 /// The registered name.
125 name: &'static str,
126 /// The record type.
127 record: &'static str,
128 /// The record's id.
129 id: String,
130 /// The publisher's error.
131 #[source]
132 source: Box<dyn StdError + Send + Sync>,
133 },
134}
135
136/// The outbox's pool was set already: by [`Outbox::new`](super::Outbox::new) or by an earlier
137/// [`set_pool`](super::Outbox::set_pool).
138///
139/// # Examples
140///
141/// ```
142/// # #[cfg(all(feature = "outbox", feature = "postgres"))]
143/// # mod demo {
144/// # use ruststream::OutgoingMessage;
145/// # use ruststream::memory::prelude::*;
146/// # use ruststream_sqlx::{Outbox, outbox};
147/// # use serde::{Deserialize, Serialize};
148/// # use sqlx::postgres::{PgConnection, PgPool, Postgres};
149/// # #[derive(Outbox, sqlx::FromRow)]
150/// # #[outbox(table = "outbox")]
151/// # pub struct OrderOutbox { #[field(id)] id: i64, #[field(name)] name: String, #[field(payload)] payload: Vec<u8> }
152/// # impl outbox::Publish<Postgres> for OrderOutbox {
153/// # async fn publish(conn: &mut PgConnection, msg: &OutgoingMessage<'_>) -> sqlx::Result<i64> {
154/// # sqlx::query_scalar("INSERT INTO outbox (name, payload) VALUES ($1, $2) RETURNING id")
155/// # .bind(msg.name()).bind(msg.payload()).fetch_one(conn).await
156/// # }
157/// # }
158/// # #[derive(Deserialize)] pub struct PlaceOrder { id: u64 }
159/// # #[derive(Serialize, Deserialize, Outgoing)] #[outgoing(name = "orders")] pub struct OrderPlaced { id: u64 }
160/// # #[subscriber("checkout", reply)] async fn place(cmd: &PlaceOrder) -> OrderPlaced { OrderPlaced { id: cmd.id } }
161/// # #[subscriber("orders")] async fn fulfil(_: &OrderPlaced) -> HandlerOutcome { HandlerOutcome::ack() }
162/// use std::io;
163///
164/// pub fn app() -> impl App {
165/// let tracking = outbox! { "orders" => OrderOutbox };
166/// let registry = tracking.clone();
167/// RustStream::new(AppInfo::new("orders", "0.1.0"))
168/// .on_startup(async move |()| {
169/// let pool = PgPool::connect("postgres://localhost/orders")
170/// .await
171/// .map_err(io::Error::other)?;
172/// // A deferred registry takes this one pool; a second one would fail startup here.
173/// registry.set_pool(pool.clone()).map_err(io::Error::other)?;
174/// Ok::<_, io::Error>(pool)
175/// })
176/// .publish_layer(tracking.publish_layer())
177/// .with_broker(MemoryBroker::new(), |b| {
178/// b.include(place).out_reply(Publish);
179/// })
180/// }
181/// # }
182/// # fn main() {}
183/// ```
184#[derive(Debug, Error)]
185#[error("the outbox's pool is set already; `set_pool` takes one pool, once")]
186#[non_exhaustive]
187pub struct PoolAlreadySet;
188
189/// A publish through [`wrap`](super::Outbox::wrap) that failed: the outbox did not record the
190/// message, or the publisher it wraps did not send it.
191///
192/// # Examples
193///
194/// ```
195/// # #[cfg(all(feature = "outbox", feature = "postgres"))]
196/// # mod demo {
197/// # use ruststream::OutgoingMessage;
198/// # use ruststream::memory::prelude::*;
199/// # use ruststream_sqlx::{Outbox, outbox};
200/// # use serde::{Deserialize, Serialize};
201/// # use sqlx::postgres::{PgConnection, PgPool, Postgres};
202/// # #[derive(Outbox, sqlx::FromRow)]
203/// # #[outbox(table = "outbox")]
204/// # pub struct OrderOutbox { #[field(id)] id: i64, #[field(name)] name: String, #[field(payload)] payload: Vec<u8> }
205/// # impl outbox::Publish<Postgres> for OrderOutbox {
206/// # async fn publish(conn: &mut PgConnection, msg: &OutgoingMessage<'_>) -> sqlx::Result<i64> {
207/// # sqlx::query_scalar("INSERT INTO outbox (name, payload) VALUES ($1, $2) RETURNING id")
208/// # .bind(msg.name()).bind(msg.payload()).fetch_one(conn).await
209/// # }
210/// # }
211/// # #[derive(Deserialize)] pub struct PlaceOrder { id: u64 }
212/// # #[derive(Serialize, Deserialize, Outgoing)] #[outgoing(name = "orders")] pub struct OrderPlaced { id: u64 }
213/// # #[subscriber("checkout", reply)] async fn place(cmd: &PlaceOrder) -> OrderPlaced { OrderPlaced { id: cmd.id } }
214/// # #[subscriber("orders")] async fn fulfil(_: &OrderPlaced) -> HandlerOutcome { HandlerOutcome::ack() }
215/// use ruststream::runtime::PublishError;
216/// use ruststream_sqlx::outbox::{OutboxError, TrackedPublishError};
217///
218/// pub async fn serve(pool: PgPool) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
219/// let tracking = outbox! { pool: pool, "orders" => OrderOutbox };
220/// let broker = MemoryBroker::new().bindable();
221/// let egress = broker.bind(Publish);
222/// let running = RustStream::new(AppInfo::new("orders", "0.1.0"))
223/// .with_broker(broker, |b| {
224/// b.include(fulfil);
225/// })
226/// .start()
227/// .await?;
228/// let publisher = tracking.wrap(running.publisher(egress).await?);
229///
230/// // The status an HTTP endpoint answers an order with.
231/// let status = match publisher.message(&OrderPlaced { id: 7 }).publish().await {
232/// Ok(()) => 202,
233/// // Recorded but not sent: the next startup sends it.
234/// Err(PublishError::Publish(TrackedPublishError::Publish(_))) => 202,
235/// // Neither recorded nor sent: the client sends the order again.
236/// Err(PublishError::Publish(TrackedPublishError::Outbox(OutboxError::Record { .. }))) => 503,
237/// Err(_) => 500,
238/// };
239/// tracing::info!(status, "order answered");
240///
241/// running.shutdown().await?;
242/// Ok(())
243/// }
244/// # }
245/// # fn main() {}
246/// ```
247#[derive(Debug, Error)]
248#[non_exhaustive]
249pub enum TrackedPublishError<PublishError> {
250 /// The record was not written, so the message was not sent.
251 #[error(transparent)]
252 Outbox(OutboxError),
253 /// The wrapped publisher's error. The record stays unprocessed, and the next startup
254 /// publishes it again.
255 #[error(transparent)]
256 Publish(PublishError),
257}