ruststream_sqlx/inbox/transactional.rs
1//! Transactional mode: a handler that writes through its delivery's transaction, which
2//! acknowledgement commits; the step that switches a registration to it, the transaction the
3//! handler borrows, and the book a subscription lends its deliveries' transactions from.
4
5use ruststream::runtime::{Declared, SubscriberBuilder, SubscriberSettings, WorkersStep};
6
7use super::queue::InboxQueue;
8use super::threads::{InboxThreads, ThreadsStep};
9
10mod book;
11pub(crate) mod settle;
12mod tx;
13pub(crate) mod writers;
14
15pub(crate) use book::{Returned, TxBook, TxHold};
16pub(crate) use tx::Lender;
17pub use tx::Tx;
18
19/// A subscription whose handler leaves the delivery's transaction alone: the default mode.
20///
21/// [`InboxQueue::new`] opens a subscription in it.
22///
23/// # Examples
24///
25/// ```
26/// # #[cfg(feature = "postgres")]
27/// # mod demo {
28/// use ruststream_sqlx::prelude::*;
29/// use serde::Deserialize;
30/// use sqlx::PgPool;
31///
32/// #[derive(Inbox, sqlx::FromRow)]
33/// #[inbox(table = "report_jobs")]
34/// pub struct Report {
35/// #[field(id, generated)]
36/// id: i64,
37/// #[field(group)]
38/// month: String,
39/// #[field(payload)]
40/// payload: Vec<u8>,
41/// }
42///
43/// #[derive(Deserialize)]
44/// struct Request {
45/// format: String,
46/// }
47///
48/// // `InboxQueue::<Report>::new(..)` is an `InboxQueue<Report, Plain>`: the claim settles the row,
49/// // and the handler writes through connections of its own.
50/// #[subscriber(InboxQueue::<Report>::new("2026-10"))]
51/// async fn render(request: &Request) -> HandlerOutcome {
52/// tracing::info!(format = %request.format, "rendering October");
53/// HandlerOutcome::ack()
54/// }
55///
56/// pub fn app(pool: PgPool) -> RustStream {
57/// RustStream::new(AppInfo::new("reports", "0.1.0")).with_broker(SqlxBroker::new(pool), |b| {
58/// b.include(render);
59/// })
60/// }
61/// # }
62/// # fn main() {}
63/// ```
64#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Hash)]
65pub struct Plain;
66
67/// A subscription whose handler writes through the delivery's transaction, which acknowledgement
68/// commits.
69///
70/// The mount-site step [`transactional`](InboxSettings::transactional) switches a registration to
71/// it: the handler then takes the transaction with `Ctx<keys::Tx<DB>>`.
72///
73/// # Examples
74///
75/// ```
76/// # #[cfg(feature = "postgres")]
77/// # mod demo {
78/// use ruststream_sqlx::prelude::*;
79/// use sqlx::{PgPool, Postgres};
80///
81/// #[derive(Inbox, sqlx::FromRow)]
82/// #[inbox(table = "signup_jobs")]
83/// pub struct Signup {
84/// #[field(id)]
85/// id: i64,
86/// #[field(payload)]
87/// payload: Vec<u8>,
88/// }
89///
90/// #[derive(serde::Deserialize)]
91/// struct Account {
92/// email: String,
93/// }
94///
95/// #[subscriber(InboxQueue::<Signup>::new("signups"))]
96/// async fn open_account(account: &Account, Ctx(mut tx): Ctx<keys::Tx<Postgres>>) -> HandlerOutcome {
97/// let opened = sqlx::query("INSERT INTO accounts (email) VALUES ($1)")
98/// .bind(&account.email)
99/// .execute(&mut *tx)
100/// .await;
101/// if opened.is_ok() { HandlerOutcome::ack() } else { HandlerOutcome::retry() }
102/// }
103///
104/// pub fn app(pool: PgPool) -> RustStream {
105/// RustStream::new(AppInfo::new("accounts", "0.1.0")).with_broker(SqlxBroker::new(pool), |b| {
106/// // The registration's `InboxQueue<Signup, Plain>` becomes
107/// // `InboxQueue<Signup, Transactional>`.
108/// b.include(open_account.transactional());
109/// })
110/// }
111/// # }
112/// # fn main() {}
113/// ```
114#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Hash)]
115pub struct Transactional;
116
117mod sealed {
118 pub trait Sealed {}
119
120 impl Sealed for super::Plain {}
121
122 impl Sealed for super::Transactional {}
123}
124
125/// The mode of a subscription as a type: [`Plain`] or [`Transactional`]. Machinery; a descriptor,
126/// its subscriber and its deliveries carry it.
127#[doc(hidden)]
128pub trait InboxMode: sealed::Sealed + Send + Sync + 'static {
129 /// Whether the handler writes through the delivery's transaction.
130 const TRANSACTIONAL: bool;
131}
132
133impl InboxMode for Plain {
134 const TRANSACTIONAL: bool = false;
135}
136
137impl InboxMode for Transactional {
138 const TRANSACTIONAL: bool = true;
139}
140
141/// The inbox's own mount-site steps, on a `#[subscriber]` definition and on its settings builder.
142///
143/// The prelude brings it in, beside the framework's own settings.
144///
145/// # Examples
146///
147/// ```
148/// # #[cfg(feature = "postgres")]
149/// # mod demo {
150/// use ruststream_sqlx::prelude::*;
151/// use sqlx::{PgPool, Postgres};
152///
153/// #[derive(Inbox, sqlx::FromRow)]
154/// #[inbox(table = "invoice_jobs")]
155/// pub struct InvoiceJob {
156/// #[field(id)]
157/// id: i64,
158/// #[field(payload)]
159/// payload: Vec<u8>,
160/// }
161///
162/// #[derive(serde::Deserialize)]
163/// struct Invoice {
164/// number: i64,
165/// }
166///
167/// #[subscriber(InboxQueue::<InvoiceJob>::new("invoices"))]
168/// async fn book(invoice: &Invoice, Ctx(mut tx): Ctx<keys::Tx<Postgres>>) -> HandlerOutcome {
169/// let booked = sqlx::query("INSERT INTO booked (number) VALUES ($1)")
170/// .bind(invoice.number)
171/// .execute(&mut *tx)
172/// .await;
173/// if booked.is_ok() { HandlerOutcome::ack() } else { HandlerOutcome::retry() }
174/// }
175///
176/// pub fn app(pool: PgPool) -> RustStream {
177/// RustStream::new(AppInfo::new("billing", "0.1.0")).with_broker(SqlxBroker::new(pool), |b| {
178/// // The inbox's step chains with the framework's.
179/// b.include(book.transactional().workers(nonzero!(4)));
180/// })
181/// }
182/// # }
183/// # fn main() {}
184/// ```
185pub trait InboxSettings: Declared {
186 /// Switches an `InboxQueue` subscription to transactional mode: the handler takes the
187 /// delivery's transaction (`Ctx<keys::Tx<DB>>`), and acknowledgement commits what it wrote.
188 ///
189 /// The handler writes through [`Tx`], which dereferences to the connection inside the
190 /// delivery's transaction. Acknowledgement commits the handler's writes together with the
191 /// settlement of the row. A retry, a delayed retry, a drop and a dead-letter move roll the
192 /// handler's writes back first, then settle the row as outside transactional mode, and so
193 /// does an acknowledgement after the handler panicked. A delivery dropped unsettled ends its
194 /// transaction, and the row returns at once.
195 ///
196 /// Each form keeps the transaction its own way. The row lock form lends the claim's
197 /// transaction: the subscription sets a savepoint right after each claim, one more statement
198 /// per delivery, so a retry discards the handler's writes and still counts the attempt. The
199 /// lease form opens a transaction of the delivery's own once the claim committed, one more
200 /// statement per delivery: the acknowledgement runs inside it by the lease the delivery holds,
201 /// and a delivery whose lease was lost meanwhile rolls everything back and fails with
202 /// [`SqlxBrokerError::LeaseLost`](crate::SqlxBrokerError::LeaseLost). The advisory lock form
203 /// opens the transaction on the session that holds the row's key, one more statement per
204 /// delivery, and ends it before the key's release. Every transaction opens at the table's
205 /// isolation level or SQLite mode. A lease table on Postgres at `isolation = repeatable_read`
206 /// or `serializable` refuses transactional mode when it starts: its transaction cannot see the
207 /// lease extended after its first statement.
208 ///
209 /// Each delivery in work holds one pool connection, so a handler that takes another connection
210 /// of the same pool needs a pool larger than its `workers(n)`. On SQLite a delivery's
211 /// transaction holds the database's one write lock from its first write, or from its start in
212 /// `immediate` mode, until it settles: every other writer waits for it. A tracked outbox
213 /// publish of the handler from then on fails with `OutboxError::WriteLockHeld`, since its
214 /// record needs that lock. Transactional mode serves single deliveries: a batch handler
215 /// mounted with it does not compile.
216 ///
217 /// # Examples
218 ///
219 /// ```
220 /// # #[cfg(feature = "postgres")]
221 /// # mod demo {
222 /// use ruststream_sqlx::prelude::*;
223 /// use sqlx::{PgPool, Postgres};
224 ///
225 /// #[derive(Inbox, sqlx::FromRow)]
226 /// #[inbox(table = "stock_jobs")]
227 /// pub struct StockJob {
228 /// #[field(id)]
229 /// id: i64,
230 /// #[field(payload)]
231 /// payload: Vec<u8>,
232 /// }
233 ///
234 /// #[derive(serde::Deserialize)]
235 /// struct Restock {
236 /// sku: String,
237 /// count: i32,
238 /// }
239 ///
240 /// #[subscriber(InboxQueue::<StockJob>::new("restocks"))]
241 /// async fn restock(order: &Restock, Ctx(mut tx): Ctx<keys::Tx<Postgres>>) -> HandlerOutcome {
242 /// let raised = sqlx::query("UPDATE stock SET count = count + $1 WHERE sku = $2")
243 /// .bind(order.count)
244 /// .bind(&order.sku)
245 /// .execute(&mut *tx)
246 /// .await;
247 /// // The count rises exactly when the job is finished: both commit together.
248 /// if raised.is_ok() { HandlerOutcome::ack() } else { HandlerOutcome::retry() }
249 /// }
250 ///
251 /// pub fn app(pool: PgPool) -> RustStream {
252 /// RustStream::new(AppInfo::new("stock", "0.1.0")).with_broker(SqlxBroker::new(pool), |b| {
253 /// b.include(restock.transactional());
254 /// })
255 /// }
256 /// # }
257 /// # fn main() {}
258 /// ```
259 fn transactional(self) -> <Self::Settings as TransactionalStep>::Out
260 where
261 Self::Settings: TransactionalStep,
262 {
263 self.declare().apply_transactional()
264 }
265
266 /// Runs an `InboxQueue` subscription's handlers on dedicated threads, as `threads(n)` does,
267 /// with a pool of each thread's own for their queries.
268 ///
269 /// It fills the position `workers(n)` and `threads(n)` fill, so the attribute names neither.
270 /// A handler on a thread reads `Ctx<keys::Pool<..>>` as the thread's pool. Its connections
271 /// open on the thread and close with it, so none outlives the thread in the service's pool.
272 /// Claims, settlements, transactions and publishes take the service's pool. The subscription
273 /// counts the threads' connections against the broker's
274 /// [`connection_limit`](crate::SqlxBroker::connection_limit) when it opens.
275 ///
276 /// # Examples
277 ///
278 /// ```
279 /// # #[cfg(feature = "postgres")]
280 /// # mod demo {
281 /// use ruststream::nonzero;
282 /// use ruststream_sqlx::prelude::*;
283 /// use sqlx::{PgPool, Postgres};
284 ///
285 /// #[derive(Inbox, sqlx::FromRow)]
286 /// #[inbox(table = "scan_jobs")]
287 /// pub struct ScanJob {
288 /// #[field(id)]
289 /// id: i64,
290 /// #[field(payload)]
291 /// payload: Vec<u8>,
292 /// }
293 ///
294 /// #[derive(serde::Deserialize)]
295 /// struct Image {
296 /// path: String,
297 /// }
298 ///
299 /// #[subscriber(InboxQueue::<ScanJob>::new("scans"))]
300 /// async fn scan(image: &Image, Ctx(pool): Ctx<keys::Pool<Postgres>>) -> HandlerOutcome {
301 /// let noted = sqlx::query("INSERT INTO scanned (path) VALUES ($1)")
302 /// .bind(&image.path)
303 /// .execute(&pool)
304 /// .await;
305 /// if noted.is_ok() { HandlerOutcome::ack() } else { HandlerOutcome::retry() }
306 /// }
307 ///
308 /// pub fn app(pool: PgPool) -> RustStream {
309 /// // The service opens 30 connections at most: the pool's and the threads' 4 x 2.
310 /// let broker = SqlxBroker::new(pool).connection_limit(nonzero!(30u32));
311 /// RustStream::new(AppInfo::new("scanner", "0.1.0")).with_broker(broker, |b| {
312 /// b.include(scan.on_threads(InboxThreads::new(nonzero!(4)).connections(nonzero!(2))));
313 /// })
314 /// }
315 /// # }
316 /// # fn main() {}
317 /// ```
318 fn on_threads(
319 self,
320 threads: InboxThreads,
321 ) -> <<Self::Settings as WorkersStep>::Out as ThreadsStep>::Out
322 where
323 Self::Settings: WorkersStep,
324 <Self::Settings as WorkersStep>::Out: ThreadsStep,
325 {
326 self.declare()
327 .apply_workers(threads.workers())
328 .apply_threads(threads)
329 }
330}
331
332impl<Def: Declared> InboxSettings for Def {}
333
334/// Switching a registration to transactional mode. Machinery behind
335/// [`InboxSettings::transactional`]; never named in a service.
336#[doc(hidden)]
337#[diagnostic::on_unimplemented(
338 message = "`.transactional()` switches an `InboxQueue` subscription to transactional mode",
339 label = "this registration does not subscribe through `InboxQueue`",
340 note = "a subscription by name has no transactional mode: subscribe with \
341 `InboxQueue::<Row>::new(..)`"
342)]
343pub trait TransactionalStep {
344 /// The registration in transactional mode.
345 type Out;
346
347 /// Switches the registration's descriptor to transactional mode.
348 fn apply_transactional(self) -> Self::Out;
349}
350
351impl<Def, Row, State, DefCodec> TransactionalStep
352 for SubscriberBuilder<Def, InboxQueue<Row, Plain>, State, DefCodec>
353where
354 Def: Declared,
355{
356 type Out = SubscriberBuilder<Def, InboxQueue<Row, Transactional>, State, DefCodec>;
357
358 fn apply_transactional(self) -> Self::Out {
359 self.map_source(InboxQueue::into_transactional)
360 }
361}