reliar_inbox/store.rs
1//! Transactional deduplication of inbound messages.
2
3use reliar_core::{Classify, MessageId};
4use tracing::Instrument as _;
5
6use crate::claim::{InboxClaim, InboxFailure, InboxOutcome};
7use crate::error::InboxProcessError;
8use crate::handler::InboxHandler;
9use crate::message::InboxMessage;
10use crate::purge::{InboxPurgeReport, InboxPurgeRequest};
11use crate::record::InboxRecord;
12use crate::scope::InboxScope;
13
14/// Transactional deduplication of inbound messages, keyed `(scope, message_id)`.
15///
16/// `Tx` is the provider's transaction type — `sqlx::Transaction<'_, Postgres>` for
17/// `reliar-store-postgres`. A **type parameter**, exactly as on `reliar-outbox`'s
18/// `OutboxEnqueue<Tx>`, so this crate names no storage type.
19///
20/// **Two transaction models on purpose.** [`Self::claim`] and [`Self::complete`] run in the
21/// caller's transaction — that is the guarantee. [`Self::fail`], [`Self::find`] and
22/// [`Self::purge`] run on the provider's own pool: `fail` records the failure of the very
23/// transaction that has just been rolled back, so it cannot live in it (ADR 0042 §4).
24///
25/// [`Self::claim`] then [`Self::complete`] — a **compiled, never-called generic function**
26/// (ADR 0043 §7: this crate ships no store to run a doctest against; `reliar-store-postgres`'s
27/// own docs carry a runnable example over `PostgresInboxStore`):
28///
29/// ```
30/// # use reliar_inbox::{InboxClaim, InboxMessage, InboxScope, InboxStore};
31/// #
32/// async fn claim_then_complete<Tx, S: InboxStore<Tx>>(
33/// store: &S,
34/// tx: &mut Tx,
35/// scope: &InboxScope,
36/// message: InboxMessage<'_>,
37/// ) {
38/// if let Ok(InboxClaim::Claimed { .. }) = store.claim(tx, scope, message).await {
39/// let _ = store.complete(tx, scope, message.id).await;
40/// }
41/// }
42/// ```
43pub trait InboxStore<Tx>: Send + Sync {
44 /// What inbox operations fail with. [`Classify`] so a caller can log a permanent failure
45 /// differently from a transient one without a downcast.
46 type Error: std::error::Error + Send + Sync + 'static + Classify;
47
48 /// Claims `message` for `scope` in the caller's transaction — see [`InboxClaim`] for the
49 /// four answers and what the caller owes each of them.
50 ///
51 /// Call it as the **first** statement of the transaction. `Ok` never leaves `tx` unusable;
52 /// an `Err` may, and whether it does is the provider's contract (with
53 /// `reliar-store-postgres` it does — PostgreSQL aborts the transaction).
54 ///
55 /// Issues no network I/O beyond its own statements, and never commits, rolls back or
56 /// otherwise consumes `tx`.
57 ///
58 /// Non-generic on purpose: a `claim<T>(…, &Envelope<T>)` would monomorphize the provider's
59 /// SQL path per body type (ADR 0042 A.2.2).
60 ///
61 /// # Errors
62 ///
63 /// Provider-defined. Treat any `Err` as *abort this transaction*.
64 ///
65 /// ```
66 /// # use reliar_inbox::{InboxClaim, InboxMessage, InboxScope, InboxStore};
67 /// #
68 /// async fn claim_only<Tx, S: InboxStore<Tx>>(
69 /// store: &S,
70 /// tx: &mut Tx,
71 /// scope: &InboxScope,
72 /// message: InboxMessage<'_>,
73 /// ) -> Option<InboxClaim> {
74 /// store.claim(tx, scope, message).await.ok()
75 /// }
76 /// ```
77 fn claim(
78 &self,
79 tx: &mut Tx,
80 scope: &InboxScope,
81 message: InboxMessage<'_>,
82 ) -> impl Future<Output = Result<InboxClaim, Self::Error>> + Send;
83
84 /// Marks the row completed in the caller's transaction, at database time. The last
85 /// statement before the caller's `commit`.
86 ///
87 /// Preserves `last_error` from earlier failed attempts — a message that eventually
88 /// succeeded keeps the evidence of why it did not the first time.
89 ///
90 /// Its guard is `completed_at IS NULL AND dead_at IS NULL` — the second clause is not
91 /// cosmetic: with `ck_inbox_terminal` in place, completing a dead row would trip the check
92 /// constraint and surface a raw database error where this contract promises a clean
93 /// "no claimed row".
94 ///
95 /// **A caller that commits without calling this** — by calling [`Self::claim`] directly and
96 /// skipping `complete`, or by committing after `complete` itself returned an error — leaves a
97 /// committed, uncompleted row. It is then indistinguishable from a row [`Self::fail`]
98 /// created: the next redelivery's `claim` answers `Claimed { attempt: 1 }` again and the
99 /// handler re-runs over already-committed business writes. [`Self::process`] never risks
100 /// this (it always calls `complete` before returning `Processed`); a caller that drives
101 /// `claim`/`complete` itself must call `complete` before its own commit to keep the guarantee.
102 ///
103 /// Keeps `id: MessageId` — completion writes no new columns, so it needs no [`InboxMessage`]
104 /// view.
105 ///
106 /// # Errors
107 ///
108 /// Provider-defined, plus a provider error for "no claimed row" — reachable by misuse
109 /// (completing without a `Claimed`, or after the transaction aborted), by completing a row
110 /// that has since gone dead, and, more benignly, by a concurrent retention [`Self::purge`]
111 /// deleting the row `claim` adopted in between: the result is a spurious `Err` here, a
112 /// rollback, and a clean redelivery — never a lost or duplicated effect.
113 ///
114 /// ```
115 /// # use reliar_core::MessageId;
116 /// # use reliar_inbox::{InboxScope, InboxStore};
117 /// #
118 /// async fn complete_only<Tx, S: InboxStore<Tx>>(
119 /// store: &S,
120 /// tx: &mut Tx,
121 /// scope: &InboxScope,
122 /// id: MessageId,
123 /// ) {
124 /// let _ = store.complete(tx, scope, id).await;
125 /// }
126 /// ```
127 fn complete(
128 &self,
129 tx: &mut Tx,
130 scope: &InboxScope,
131 id: MessageId,
132 ) -> impl Future<Output = Result<(), Self::Error>> + Send;
133
134 /// Records a failed attempt **on the provider's own pool**, in its own short transaction.
135 /// Call it *after* rolling the handler's transaction back.
136 ///
137 /// Increments `attempts` and stores `error`'s `Display` chain (truncated to 2 KiB at a char
138 /// boundary with a `"…[truncated]"` marker, like [`crate::InboxRecord::last_error`]),
139 /// creating the row if the rollback removed it. A row that is already completed is left
140 /// untouched: a stale attempt cannot un-complete work another consumer finished.
141 ///
142 /// Takes the same [`InboxMessage`] view [`Self::claim`] does, because it **creates** the row
143 /// when the rollback removed it and must supply every `NOT NULL` column — and it returns an
144 /// [`InboxFailure`], because whether the row just went dead decides the caller's next broker
145 /// call.
146 ///
147 /// An implementation SHALL bound recorded failures at a configured `max_attempts` and SHALL
148 /// apply the transition to `dead_at` **atomically with the increment** — reading the count
149 /// and writing the transition back separately races two concurrent `fail`s and can skip the
150 /// transition or apply it twice.
151 ///
152 /// **Best-effort bookkeeping.** Skipping it — or crashing before it — costs an uncounted
153 /// attempt and nothing else; no Reliar decision is taken on the count. `attempts` therefore
154 /// stays a **lower** bound, so `max_attempts` bounds *recorded* failures only, and nothing
155 /// about the row bounds effects **outside** the database.
156 ///
157 /// **Implementors:** `error: &dyn Error` is not `Send` (`&T: Send` requires `T: Sync`, and
158 /// `dyn Error` is not `Sync`), so a plain `async fn` that carries it across the async block's
159 /// construction will not satisfy this method's `+ Send` return bound. Extract `error`'s
160 /// `Display` chain into an owned `String` synchronously, before the async block is built —
161 /// see `reliar-store-postgres`'s `PostgresInboxStore::fail` for the reference shape.
162 /// `InboxMessage<'_>` itself may be captured: `MessageType` and `CorrelationId` are both
163 /// `Sync`.
164 ///
165 /// # Errors
166 ///
167 /// Provider-defined. An `Err` here changes nothing about the message's fate; log it and
168 /// `nak`.
169 ///
170 /// ```
171 /// # use reliar_inbox::{InboxFailure, InboxMessage, InboxScope, InboxStore};
172 /// #
173 /// # #[derive(Debug)]
174 /// # struct Boom;
175 /// # impl std::fmt::Display for Boom {
176 /// # fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
177 /// # f.write_str("boom")
178 /// # }
179 /// # }
180 /// # impl std::error::Error for Boom {}
181 /// #
182 /// async fn fail_only<Tx, S: InboxStore<Tx>>(
183 /// store: &S,
184 /// scope: &InboxScope,
185 /// message: InboxMessage<'_>,
186 /// ) -> Option<InboxFailure> {
187 /// store.fail(scope, message, &Boom).await.ok()
188 /// }
189 /// ```
190 fn fail(
191 &self,
192 scope: &InboxScope,
193 message: InboxMessage<'_>,
194 error: &(dyn std::error::Error + 'static),
195 ) -> impl Future<Output = Result<InboxFailure, Self::Error>> + Send;
196
197 /// Reads a row for diagnostics. No Reliar code path calls it.
198 ///
199 /// # Errors
200 ///
201 /// Provider-defined.
202 ///
203 /// ```
204 /// # use reliar_core::MessageId;
205 /// # use reliar_inbox::{InboxRecord, InboxScope, InboxStore};
206 /// #
207 /// async fn find_only<Tx, S: InboxStore<Tx>>(
208 /// store: &S,
209 /// scope: &InboxScope,
210 /// id: MessageId,
211 /// ) -> Option<InboxRecord> {
212 /// store.find(scope, id).await.ok().flatten()
213 /// }
214 /// ```
215 fn find(
216 &self,
217 scope: &InboxScope,
218 id: MessageId,
219 ) -> impl Future<Output = Result<Option<InboxRecord>, Self::Error>> + Send;
220
221 /// Deletes rows past retention, bounded by `request.batch_size`, on the provider's own pool.
222 /// Idempotent and safe to run concurrently with consumers. `Tx` is inert here.
223 ///
224 /// **Retention is the redelivery window, not a storage budget** — see [`InboxPurgeRequest`].
225 ///
226 /// # Errors
227 ///
228 /// Provider-defined.
229 ///
230 /// ```
231 /// # use reliar_inbox::{InboxPurgeReport, InboxPurgeRequest, InboxStore};
232 /// #
233 /// async fn purge_only<Tx, S: InboxStore<Tx>>(store: &S) -> Option<InboxPurgeReport> {
234 /// store.purge(InboxPurgeRequest::default()).await.ok()
235 /// }
236 /// ```
237 fn purge(
238 &self,
239 request: InboxPurgeRequest,
240 ) -> impl Future<Output = Result<InboxPurgeReport, Self::Error>> + Send;
241
242 /// The happy path in one call: claim, branch, run `handler`, complete.
243 ///
244 /// **Never commits and never calls [`Self::fail`]** — it holds only `&mut Tx`, and `fail`
245 /// needs a different connection than the transaction being rolled back. The caller owns
246 /// both, and the ordering is the part that must not be improvised — **six** branches, and
247 /// the `Err(Handler)` row branches again on what `fail` returned:
248 ///
249 /// | Result | Caller does |
250 /// |---|---|
251 /// | `Ok(Processed(v))` | `tx.commit()`, **then** ack |
252 /// | `Ok(AlreadyCompleted { .. })` | `tx.rollback()`, ack |
253 /// | `Ok(InProgress)` | `tx.rollback()`, `nak` with a delay |
254 /// | `Ok(Dead { id, .. })` | `tx.rollback()`, log `id`, **`term`** — never re-run the handler |
255 /// | `Err(Handler(e))` | `tx.rollback()`, then `self.fail(scope, message, &e)` and follow its [`InboxFailure`]: `Recorded` ⇒ `nak`, `Dead` ⇒ `term`, `AlreadyCompleted` ⇒ ack |
256 /// | `Err(Store(e))` | `tx.rollback()`, log, `nak` |
257 ///
258 /// A host that ignores `Dead` will `nak` forever and the message will bounce until the
259 /// broker's own `max_deliver`.
260 ///
261 /// Acking before the commit succeeds turns at-least-once into at-most-once: the message is
262 /// gone and its effects were rolled back. Following this table keeps [`Self::complete`]'s own
263 /// "commit without complete" duplicate window from ever opening: `process` calls `complete`
264 /// on every path that reaches `tx.commit()`, so a caller that only ever commits on
265 /// `Ok(Processed(_))` never needs `complete`'s rustdoc to protect itself.
266 ///
267 /// Implementors **SHALL NOT** override this method — it is a fixed spelling of
268 /// `claim`/`complete`, not an extension point.
269 ///
270 /// # Errors
271 ///
272 /// [`InboxProcessError::Handler`] when `handler` fails, [`InboxProcessError::Store`] for any
273 /// claim or completion failure.
274 ///
275 /// ```
276 /// # use reliar_inbox::{InboxHandler, InboxMessage, InboxOutcome, InboxProcessError, InboxScope, InboxStore};
277 /// #
278 /// # struct RecordOrder;
279 /// # impl<Tx: Send> InboxHandler<Tx> for RecordOrder {
280 /// # type Output = &'static str;
281 /// # type Error = std::convert::Infallible;
282 /// # async fn handle(&self, _tx: &mut Tx) -> Result<Self::Output, Self::Error> {
283 /// # Ok("order recorded")
284 /// # }
285 /// # }
286 /// #
287 /// async fn process_only<Tx: Send, S: InboxStore<Tx>>(
288 /// store: &S,
289 /// tx: &mut Tx,
290 /// scope: &InboxScope,
291 /// message: InboxMessage<'_>,
292 /// ) -> Result<InboxOutcome<&'static str>, InboxProcessError<S::Error, std::convert::Infallible>> {
293 /// store.process(tx, scope, message, &RecordOrder).await
294 /// }
295 /// ```
296 // Block form: a provided method with an `impl Future` signature must use it (conventions §3;
297 // also reason (a) — the span below is opened eagerly, at call time, not on first poll).
298 #[allow(
299 clippy::type_complexity,
300 reason = "the return type names exactly the two outcomes `process` can produce; a type \
301 alias would need `Self::Error` and `H::Error` as parameters and read no clearer"
302 )]
303 fn process<H>(
304 &self,
305 tx: &mut Tx,
306 scope: &InboxScope,
307 message: InboxMessage<'_>,
308 handler: &H,
309 ) -> impl Future<
310 Output = Result<InboxOutcome<H::Output>, InboxProcessError<Self::Error, H::Error>>,
311 > + Send
312 where
313 H: InboxHandler<Tx> + Sync,
314 Tx: Send,
315 {
316 let span = tracing::info_span!(
317 "reliar.inbox.process",
318 inbox.scope = %scope,
319 message.id = %message.id,
320 message.r#type = %message.message_type,
321 inbox.outcome = tracing::field::Empty,
322 inbox.record_id = tracing::field::Empty,
323 );
324 let recording_span = span.clone();
325
326 async move {
327 let outcome = match self
328 .claim(tx, scope, message)
329 .await
330 .map_err(InboxProcessError::Store)?
331 {
332 InboxClaim::AlreadyCompleted { completed_at } => {
333 InboxOutcome::AlreadyCompleted { completed_at }
334 }
335
336 InboxClaim::InProgress => InboxOutcome::InProgress,
337
338 InboxClaim::Dead {
339 id,
340 attempts,
341 dead_at,
342 } => InboxOutcome::Dead {
343 id,
344 attempts,
345 dead_at,
346 },
347
348 InboxClaim::Claimed { .. } => {
349 let output = handler
350 .handle(tx)
351 .await
352 .map_err(InboxProcessError::Handler)?;
353
354 self.complete(tx, scope, message.id)
355 .await
356 .map_err(InboxProcessError::Store)?;
357
358 InboxOutcome::Processed(output)
359 }
360 };
361
362 recording_span.record("inbox.outcome", outcome_label(&outcome));
363
364 if let InboxOutcome::Dead { id, .. } = &outcome {
365 recording_span.record("inbox.record_id", tracing::field::display(id));
366 }
367
368 Ok(outcome)
369 }
370 .instrument(span)
371 }
372}
373
374/// The `inbox.outcome` span field value for [`InboxStore::process`]'s success path. Never
375/// recorded on an error path — the field stays empty, and the caller's own log carries the
376/// failure.
377fn outcome_label<T>(outcome: &InboxOutcome<T>) -> &'static str {
378 match outcome {
379 InboxOutcome::Processed(_) => "processed",
380 InboxOutcome::AlreadyCompleted { .. } => "already_completed",
381 InboxOutcome::InProgress => "in_progress",
382 InboxOutcome::Dead { .. } => "dead",
383 }
384}