cratefield_core/outbox.rs
1//! Durable outbox (issue #128): the pair of the [`Inbox`](crate::Inbox). Where
2//! the inbox makes *inbound* effects exactly-once, the outbox makes *outbound*
3//! work at-least-once even across a crash.
4//!
5//! `Defer` (Workers `wait_until`, native `tokio::spawn`) is an execution
6//! *opportunity*, not durable delivery: a process exit, an execution deadline
7//! or a downstream failure loses the confirmation mail or the cross-module
8//! action after the database mutation already committed. The outbox closes that
9//! gap:
10//!
11//! 1. A module writes an outbox row **inside the same batch** as its state
12//! change — [`enqueue_statement`](Outbox::enqueue_statement) returns a
13//! [`Statement`] the module appends to its own `db.batch_atomic(..)`, so the row
14//! commits atomically with the change or not at all.
15//! 2. It then uses `Defer` only to *attempt* immediate delivery: lease due rows
16//! with [`claim_due`](Outbox::claim_due), deliver, and
17//! [`complete`](Outbox::complete) (delete) or [`retry_later`](Outbox::retry_later).
18//! 3. The venture's scheduled entry point drains whatever immediate delivery
19//! missed, on the same lease + bounded-retry path.
20//!
21//! Leasing is race-free without a portable `RETURNING`: `claim_due` selects due
22//! rows, then wins each one with a guarded `UPDATE … WHERE locked_until IS NULL
23//! OR locked_until < now` (the same first-writer-wins the inbox uses), so two
24//! drainers never deliver the same row. Consumers must still be idempotent
25//! (pair the topic with an [`Inbox`](crate::Inbox) key) — the outbox guarantees
26//! at-least-once, not exactly-once.
27
28use crate::ports::{Database, DbError, Statement};
29use sea_query::{Alias, Expr, Order, Query};
30
31fn iden(name: &str) -> Alias {
32 Alias::new(name)
33}
34
35/// A leased outbox record handed to a drainer.
36#[derive(Debug, Clone, PartialEq, Eq)]
37pub struct OutboxRecord {
38 pub id: String,
39 /// What kind of work this is (the module routes on it).
40 pub topic: String,
41 /// The opaque payload the module wrote (typically JSON).
42 pub payload: String,
43 /// How many delivery attempts have already failed.
44 pub attempts: i64,
45}
46
47/// A durable work queue over the `Database` port. Construct it with the table
48/// the owning module declares (e.g. `"<module>_outbox"`).
49#[derive(Debug, Clone)]
50pub struct Outbox {
51 table: String,
52}
53
54impl Outbox {
55 #[must_use]
56 pub fn new(table: impl Into<String>) -> Self {
57 Self {
58 table: table.into(),
59 }
60 }
61
62 /// The portable DDL for the outbox table. The owning module ships this as a
63 /// forward-only migration.
64 #[must_use]
65 pub fn create_table_sql(&self) -> String {
66 format!(
67 "CREATE TABLE IF NOT EXISTS {table} (\n \
68 id TEXT PRIMARY KEY,\n \
69 topic TEXT NOT NULL,\n \
70 payload TEXT NOT NULL,\n \
71 subject TEXT,\n \
72 attempts INTEGER NOT NULL DEFAULT 0,\n \
73 next_attempt_at TEXT NOT NULL,\n \
74 locked_until TEXT,\n \
75 created_at TEXT NOT NULL\n);",
76 table = self.table
77 )
78 }
79
80 /// The `INSERT` that enqueues one unit of work. Return it into the module's
81 /// **own** `db.batch_atomic(..)` alongside the state change, so the row is durable
82 /// exactly when the change is. `id` is a caller-supplied ULID; `at` is an
83 /// RFC 3339 timestamp used for both `created_at` and the initial
84 /// `next_attempt_at` (deliver as soon as possible).
85 ///
86 /// `subject` is the person the work is for — an account id, a waitlist
87 /// entry id — or `None` for work that names nobody. Writing it is what
88 /// makes the queued row reachable for export and erasure (issue #266):
89 /// the payload JSON is the module's own dialect and no predicate can
90 /// match into it. Pass it for every per-person job even when it feels
91 /// redundant with the payload; `None` rows drain identically but are
92 /// returned for nobody's subject, so a forgotten `Some` is a silent
93 /// hole in the erasure catalogue rather than an error.
94 #[must_use]
95 pub fn enqueue_statement(
96 &self,
97 id: &str,
98 topic: &str,
99 payload: &str,
100 subject: Option<&str>,
101 at: &str,
102 ) -> Statement {
103 let mut insert = Query::insert();
104 insert
105 .into_table(iden(&self.table))
106 .columns([
107 "id",
108 "topic",
109 "payload",
110 "subject",
111 "attempts",
112 "next_attempt_at",
113 "created_at",
114 ])
115 .values_panic([
116 id.to_owned().into(),
117 topic.to_owned().into(),
118 payload.to_owned().into(),
119 subject.map(str::to_owned).into(),
120 0i64.into(),
121 at.to_owned().into(),
122 at.to_owned().into(),
123 ]);
124 Statement::render(&insert)
125 }
126
127 /// Leases up to `limit` records that are due (`next_attempt_at <= now`) and
128 /// not already leased, marking each `locked_until = lease_until` so a
129 /// concurrent drainer skips it. Returns only the rows this caller won.
130 ///
131 /// # Errors
132 ///
133 /// Returns [`DbError`] if a read or lease write fails.
134 pub async fn claim_due(
135 &self,
136 db: &dyn Database,
137 now: &str,
138 lease_until: &str,
139 limit: u64,
140 ) -> Result<Vec<OutboxRecord>, DbError> {
141 let mut select = Query::select();
142 select
143 .columns([iden("id"), iden("topic"), iden("payload"), iden("attempts")])
144 .from(iden(&self.table))
145 .and_where(Expr::col(iden("next_attempt_at")).lte(now))
146 .and_where(
147 Expr::col(iden("locked_until"))
148 .is_null()
149 .or(Expr::col(iden("locked_until")).lt(now)),
150 )
151 .order_by(iden("next_attempt_at"), Order::Asc)
152 .limit(limit);
153 let rows = db.query(&Statement::render(&select)).await?;
154
155 let mut leased = Vec::new();
156 for row in &rows.rows {
157 let id = row.get::<String>("id").unwrap_or_default();
158 // Win the lease with a guarded update: exactly one drainer's write
159 // takes, and only it processes the row.
160 let mut lease = Query::update();
161 lease
162 .table(iden(&self.table))
163 .value(iden("locked_until"), lease_until)
164 .and_where(Expr::col(iden("id")).eq(id.as_str()))
165 .and_where(
166 Expr::col(iden("locked_until"))
167 .is_null()
168 .or(Expr::col(iden("locked_until")).lt(now)),
169 );
170 if db.execute(&Statement::render(&lease)).await? == 1 {
171 leased.push(OutboxRecord {
172 id,
173 topic: row.get::<String>("topic").unwrap_or_default(),
174 payload: row.get::<String>("payload").unwrap_or_default(),
175 attempts: row.get::<i64>("attempts").unwrap_or(0),
176 });
177 }
178 }
179 Ok(leased)
180 }
181
182 /// Removes a record after successful delivery.
183 ///
184 /// # Errors
185 ///
186 /// Returns [`DbError`] if the delete fails.
187 pub async fn complete(&self, db: &dyn Database, id: &str) -> Result<(), DbError> {
188 let mut delete = Query::delete();
189 delete
190 .from_table(iden(&self.table))
191 .and_where(Expr::col(iden("id")).eq(id));
192 db.execute(&Statement::render(&delete)).await?;
193 Ok(())
194 }
195
196 /// Reschedules a record after a failed delivery: increments `attempts`,
197 /// sets the next attempt time (the caller applies its own backoff), and
198 /// clears the lease so a drainer can pick it up again.
199 ///
200 /// # Errors
201 ///
202 /// Returns [`DbError`] if the update fails.
203 pub async fn retry_later(
204 &self,
205 db: &dyn Database,
206 id: &str,
207 next_attempt_at: &str,
208 ) -> Result<(), DbError> {
209 let mut update = Query::update();
210 update
211 .table(iden(&self.table))
212 .value(iden("attempts"), Expr::col(iden("attempts")).add(1))
213 .value(iden("next_attempt_at"), next_attempt_at)
214 .value(iden("locked_until"), Option::<String>::None)
215 .and_where(Expr::col(iden("id")).eq(id));
216 db.execute(&Statement::render(&update)).await?;
217 Ok(())
218 }
219}
220
221#[cfg(test)]
222mod tests {
223 use super::*;
224
225 #[test]
226 fn create_table_sql_is_portable_ddl() {
227 let sql = Outbox::new("mail_outbox").create_table_sql();
228 assert!(sql.contains("CREATE TABLE IF NOT EXISTS mail_outbox"));
229 assert!(sql.contains("next_attempt_at TEXT NOT NULL"));
230 assert!(sql.contains("locked_until TEXT"));
231 // Nullable, not NOT NULL: the migration story is ADD COLUMN on a
232 // live table, and NOT NULL would force a backfill that cannot be
233 // honest about rows whose payload never parsed.
234 assert!(sql.contains("subject TEXT"));
235 }
236
237 #[test]
238 fn enqueue_statement_inserts_the_row() {
239 let stmt = Outbox::new("mail_outbox").enqueue_statement(
240 "01J",
241 "confirmation",
242 "{\"to\":\"a@b\"}",
243 Some("acct-1"),
244 "2026-09-07T00:00:00Z",
245 );
246 assert!(stmt.sql.contains("INSERT INTO"));
247 assert!(stmt.sql.contains("mail_outbox"));
248 assert!(stmt.sql.contains("subject"));
249 // id, topic, payload, subject, attempts, next_attempt_at, created_at
250 assert_eq!(stmt.values.0.len(), 7);
251 assert_eq!(stmt.values.0[3], "acct-1".into());
252 }
253
254 #[test]
255 fn enqueue_statement_without_a_subject_binds_null() {
256 // The shape a row written before the migration has: the column
257 // exists, the value does not. The drain treats both the same.
258 let stmt = Outbox::new("mail_outbox").enqueue_statement(
259 "01J",
260 "confirmation",
261 "{\"to\":\"a@b\"}",
262 None,
263 "2026-09-07T00:00:00Z",
264 );
265 assert!(stmt.sql.contains("subject"));
266 assert_eq!(
267 stmt.values.0[3],
268 sea_query::Value::from(Option::<String>::None)
269 );
270 }
271}