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 /// The unprocessed records of a name were not read at startup.
90 #[error("the outbox did not read the unprocessed records of `{name}` from `{record}`")]
91 Recover {
92 /// The registered name.
93 name: &'static str,
94 /// The record type.
95 record: &'static str,
96 /// The database's error.
97 #[source]
98 source: SqlError,
99 },
100 /// A record read at startup was not published again.
101 #[error("the outbox did not publish record {id} of `{name}` in `{record}` again")]
102 Republish {
103 /// The registered name.
104 name: &'static str,
105 /// The record type.
106 record: &'static str,
107 /// The record's id.
108 id: String,
109 /// The publisher's error.
110 #[source]
111 source: Box<dyn StdError + Send + Sync>,
112 },
113}
114
115/// The outbox's pool was set already: by [`Outbox::new`](super::Outbox::new) or by an earlier
116/// [`set_pool`](super::Outbox::set_pool).
117///
118/// # Examples
119///
120/// ```
121/// # #[cfg(all(feature = "outbox", feature = "postgres"))]
122/// # mod demo {
123/// # use ruststream::OutgoingMessage;
124/// # use ruststream::memory::prelude::*;
125/// # use ruststream_sqlx::{Outbox, outbox};
126/// # use serde::{Deserialize, Serialize};
127/// # use sqlx::postgres::{PgConnection, PgPool, Postgres};
128/// # #[derive(Outbox, sqlx::FromRow)]
129/// # #[outbox(table = "outbox")]
130/// # pub struct OrderOutbox { #[field(id)] id: i64, #[field(name)] name: String, #[field(payload)] payload: Vec<u8> }
131/// # impl outbox::Publish<Postgres> for OrderOutbox {
132/// # async fn publish(conn: &mut PgConnection, msg: &OutgoingMessage<'_>) -> sqlx::Result<i64> {
133/// # sqlx::query_scalar("INSERT INTO outbox (name, payload) VALUES ($1, $2) RETURNING id")
134/// # .bind(msg.name()).bind(msg.payload()).fetch_one(conn).await
135/// # }
136/// # }
137/// # #[derive(Deserialize)] pub struct PlaceOrder { id: u64 }
138/// # #[derive(Serialize, Deserialize, Outgoing)] #[outgoing(name = "orders")] pub struct OrderPlaced { id: u64 }
139/// # #[subscriber("checkout", reply)] async fn place(cmd: &PlaceOrder) -> OrderPlaced { OrderPlaced { id: cmd.id } }
140/// # #[subscriber("orders")] async fn fulfil(_: &OrderPlaced) -> HandlerOutcome { HandlerOutcome::ack() }
141/// use std::io;
142///
143/// pub fn app() -> impl App {
144/// let tracking = outbox! { "orders" => OrderOutbox };
145/// let registry = tracking.clone();
146/// RustStream::new(AppInfo::new("orders", "0.1.0"))
147/// .on_startup(async move |()| {
148/// let pool = PgPool::connect("postgres://localhost/orders")
149/// .await
150/// .map_err(io::Error::other)?;
151/// // A deferred registry takes this one pool; a second one would fail startup here.
152/// registry.set_pool(pool.clone()).map_err(io::Error::other)?;
153/// Ok::<_, io::Error>(pool)
154/// })
155/// .publish_layer(tracking.publish_layer())
156/// .with_broker(MemoryBroker::new(), |b| {
157/// b.include(place).out_reply(Publish);
158/// })
159/// }
160/// # }
161/// # fn main() {}
162/// ```
163#[derive(Debug, Error)]
164#[error("the outbox's pool is set already; `set_pool` takes one pool, once")]
165#[non_exhaustive]
166pub struct PoolAlreadySet;
167
168/// A publish through [`wrap`](super::Outbox::wrap) that failed: the outbox did not record the
169/// message, or the publisher it wraps did not send it.
170///
171/// # Examples
172///
173/// ```
174/// # #[cfg(all(feature = "outbox", feature = "postgres"))]
175/// # mod demo {
176/// # use ruststream::OutgoingMessage;
177/// # use ruststream::memory::prelude::*;
178/// # use ruststream_sqlx::{Outbox, outbox};
179/// # use serde::{Deserialize, Serialize};
180/// # use sqlx::postgres::{PgConnection, PgPool, Postgres};
181/// # #[derive(Outbox, sqlx::FromRow)]
182/// # #[outbox(table = "outbox")]
183/// # pub struct OrderOutbox { #[field(id)] id: i64, #[field(name)] name: String, #[field(payload)] payload: Vec<u8> }
184/// # impl outbox::Publish<Postgres> for OrderOutbox {
185/// # async fn publish(conn: &mut PgConnection, msg: &OutgoingMessage<'_>) -> sqlx::Result<i64> {
186/// # sqlx::query_scalar("INSERT INTO outbox (name, payload) VALUES ($1, $2) RETURNING id")
187/// # .bind(msg.name()).bind(msg.payload()).fetch_one(conn).await
188/// # }
189/// # }
190/// # #[derive(Deserialize)] pub struct PlaceOrder { id: u64 }
191/// # #[derive(Serialize, Deserialize, Outgoing)] #[outgoing(name = "orders")] pub struct OrderPlaced { id: u64 }
192/// # #[subscriber("checkout", reply)] async fn place(cmd: &PlaceOrder) -> OrderPlaced { OrderPlaced { id: cmd.id } }
193/// # #[subscriber("orders")] async fn fulfil(_: &OrderPlaced) -> HandlerOutcome { HandlerOutcome::ack() }
194/// use ruststream::runtime::PublishError;
195/// use ruststream_sqlx::outbox::{OutboxError, TrackedPublishError};
196///
197/// pub async fn serve(pool: PgPool) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
198/// let tracking = outbox! { pool: pool, "orders" => OrderOutbox };
199/// let broker = MemoryBroker::new().bindable();
200/// let egress = broker.bind(Publish);
201/// let running = RustStream::new(AppInfo::new("orders", "0.1.0"))
202/// .with_broker(broker, |b| {
203/// b.include(fulfil);
204/// })
205/// .start()
206/// .await?;
207/// let publisher = tracking.wrap(running.publisher(egress).await?);
208///
209/// // The status an HTTP endpoint answers an order with.
210/// let status = match publisher.message(&OrderPlaced { id: 7 }).publish().await {
211/// Ok(()) => 202,
212/// // Recorded but not sent: the next startup sends it.
213/// Err(PublishError::Publish(TrackedPublishError::Publish(_))) => 202,
214/// // Neither recorded nor sent: the client sends the order again.
215/// Err(PublishError::Publish(TrackedPublishError::Outbox(OutboxError::Record { .. }))) => 503,
216/// Err(_) => 500,
217/// };
218/// tracing::info!(status, "order answered");
219///
220/// running.shutdown().await?;
221/// Ok(())
222/// }
223/// # }
224/// # fn main() {}
225/// ```
226#[derive(Debug, Error)]
227#[non_exhaustive]
228pub enum TrackedPublishError<PublishError> {
229 /// The record was not written, so the message was not sent.
230 #[error(transparent)]
231 Outbox(OutboxError),
232 /// The wrapped publisher's error. The record stays unprocessed, and the next startup
233 /// publishes it again.
234 #[error(transparent)]
235 Publish(PublishError),
236}