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