Line data Source code
1 : // SPDX-License-Identifier: EUPL-1.2
2 : //! Transactional outbox: the change event is
3 : //! INSERTed in the SAME transaction as the entity write, so a broker crash
4 : //! between commit and publish can never lose an event. The drain loop
5 : //! publishes rows to the bus with `Nats-Msg-Id` = `seq` for dedup, then acks
6 : //! by deleting exactly the seqs it published — never a range, see [`ack`].
7 : //!
8 : //! A row whose event was too big for the bus is [`retain`]ed instead of
9 : //! deleted: the published message carries a claim-check reference, and this
10 : //! row is the only remaining copy of the bodies it dropped. The consumer
11 : //! reads it back with [`event`] and [`reap_published`] frees it once the bus
12 : //! can no longer be carrying the message.
13 : //!
14 : //! Producer wiring into the entity write paths lands WITH the drain:
15 : //! enqueuing events nothing consumes would only grow the table
16 : //! without bound.
17 :
18 : use antares_model::TenantId;
19 : use serde_json::Value;
20 : use sqlx::postgres::{PgConnection, PgPool};
21 : use sqlx::Row;
22 :
23 : /// Every statement this module issues. Values are bound as `$n` — the strings
24 : /// are compile-time constants, so no request data ever reaches the parser.
25 : const ENQUEUE_SQL: &str = "INSERT INTO outbox (tenant_id, event) VALUES ($1, $2) RETURNING seq";
26 : const ENQUEUE_MANY_SQL: &str = "INSERT INTO outbox (tenant_id, event)
27 : SELECT $1, e FROM jsonb_array_elements($2::jsonb) AS e";
28 : const PEEK_SQL: &str =
29 : "SELECT seq, tenant_id, event FROM outbox WHERE published_at IS NULL ORDER BY seq LIMIT $1";
30 : const ACK_SQL: &str = "DELETE FROM outbox WHERE seq = ANY($1)";
31 : const RETAIN_SQL: &str = "UPDATE outbox SET published_at = now() WHERE seq = ANY($1)";
32 : const EVENT_SQL: &str = "SELECT event FROM outbox WHERE seq = $1 AND tenant_id = $2";
33 : const REAP_SQL: &str =
34 : "DELETE FROM outbox WHERE published_at < now() - make_interval(hours => $1::int)";
35 :
36 : /// Enqueue one event INSIDE the caller's transaction (same-tx INSERT).
37 12 : pub async fn enqueue(
38 12 : tx: &mut PgConnection,
39 12 : tenant: &TenantId,
40 12 : event: &Value,
41 12 : ) -> Result<i64, sqlx::Error> {
42 12 : let row = sqlx::query(ENQUEUE_SQL)
43 12 : .bind(tenant.as_str())
44 12 : .bind(event)
45 12 : .fetch_one(tx)
46 12 : .await?;
47 12 : Ok(row.get::<i64, _>(0))
48 12 : }
49 :
50 : /// Enqueue a whole batch in ONE multi-row INSERT (a
51 : /// per-item loop cost N round-trips inside every batch transaction).
52 6 : pub async fn enqueue_many(
53 6 : tx: &mut PgConnection,
54 6 : tenant: &TenantId,
55 6 : events: &[Value],
56 6 : ) -> Result<(), sqlx::Error> {
57 6 : if events.is_empty() {
58 6 : return Ok(());
59 0 : }
60 0 : sqlx::query(ENQUEUE_MANY_SQL)
61 0 : .bind(tenant.as_str())
62 0 : .bind(Value::Array(events.to_vec()))
63 0 : .execute(tx)
64 0 : .await
65 0 : .map(|_| ())
66 6 : }
67 :
68 : /// Oldest-first page for the drain loop. `seq` is the dedup id.
69 : ///
70 : /// The drain is cross-tenant by nature — it runs under the transaction-scoped
71 : /// `antares.service` escape (0001_init.sql) so it stays correct under a
72 : /// non-superuser role, where the plain tenant policy would silently return
73 : /// zero rows forever (the very failure this table exists to prevent).
74 743 : pub async fn peek(pool: &PgPool, limit: i64) -> Result<Vec<(i64, String, Value)>, sqlx::Error> {
75 743 : let mut tx = pool.begin().await?;
76 743 : crate::store::pg::set_service(&mut tx).await?;
77 743 : let rows = sqlx::query(PEEK_SQL)
78 743 : .bind(limit)
79 743 : .fetch_all(&mut *tx)
80 743 : .await?;
81 743 : tx.commit().await?;
82 743 : Ok(rows
83 743 : .into_iter()
84 743 : .map(|r| (r.get(0), r.get(1), r.get(2)))
85 743 : .collect())
86 743 : }
87 :
88 : /// Ack EXACTLY the published seqs: bigserial
89 : /// allocates at INSERT and commits land out of order, so a blanket
90 : /// `seq <= max` deletes a lower-seq row that commits between peek and ack —
91 : /// an event lost unpublished. Deleting by exact seq can never touch a row
92 : /// the drain did not publish.
93 5 : pub async fn ack(pool: &PgPool, seqs: &[i64]) -> Result<u64, sqlx::Error> {
94 5 : if seqs.is_empty() {
95 2 : return Ok(0);
96 3 : }
97 3 : let mut tx = pool.begin().await?;
98 3 : crate::store::pg::set_service(&mut tx).await?;
99 3 : let n = sqlx::query(ACK_SQL)
100 3 : .bind(seqs)
101 3 : .execute(&mut *tx)
102 3 : .await?
103 3 : .rows_affected();
104 3 : tx.commit().await?;
105 3 : Ok(n)
106 5 : }
107 :
108 : /// Keep the rows of events the bus could not carry whole, and take them out
109 : /// of the drain's page. The published message holds only a claim-check
110 : /// reference; these rows hold the bodies that reference stands for, and the
111 : /// consumer resolves one by `seq`.
112 : ///
113 : /// Under the ROW's tenant, never the `antares.service` escape: this is the
114 : /// only UPDATE any internal job issues against the outbox, and the escape is
115 : /// deliberately absent from the UPDATE policy (0005) because an escaped
116 : /// UPDATE can move a row into another tenant. The drain reads each row's
117 : /// tenant off the page it just peeked, so it has the one this needs.
118 4 : pub async fn retain(pool: &PgPool, tenant: &TenantId, seqs: &[i64]) -> Result<u64, sqlx::Error> {
119 4 : if seqs.is_empty() {
120 0 : return Ok(0);
121 4 : }
122 4 : let mut tx = pool.begin().await?;
123 4 : crate::store::pg::set_tenant(&mut tx, tenant).await?;
124 4 : let n = sqlx::query(RETAIN_SQL)
125 4 : .bind(seqs)
126 4 : .execute(&mut *tx)
127 4 : .await?
128 4 : .rows_affected();
129 4 : tx.commit().await?;
130 4 : Ok(n)
131 4 : }
132 :
133 : /// The whole event behind a claim-check reference. Read under the tenant the
134 : /// consumer decoded from the message and bound in the statement besides, so a
135 : /// reference can only ever resolve inside the tenant that wrote it.
136 6 : pub async fn event(
137 6 : pool: &PgPool,
138 6 : seq: i64,
139 6 : tenant: &TenantId,
140 6 : ) -> Result<Option<Value>, sqlx::Error> {
141 6 : let mut tx = pool.begin().await?;
142 6 : crate::store::pg::set_tenant(&mut tx, tenant).await?;
143 6 : let row = sqlx::query(EVENT_SQL)
144 6 : .bind(seq)
145 6 : .bind(tenant.as_str())
146 6 : .fetch_optional(&mut *tx)
147 6 : .await?;
148 6 : tx.commit().await?;
149 6 : Ok(row.map(|r| r.get::<Value, _>(0)))
150 6 : }
151 :
152 : /// Free retained rows older than `hours`. The window is the consumer's, not
153 : /// the publisher's: the message is on the bus until every durable has acked
154 : /// it, and this row has to outlive that. Deleting it early costs the
155 : /// notification the reference was carrying.
156 401 : pub async fn reap_published(pool: &PgPool, hours: i64) -> Result<u64, sqlx::Error> {
157 401 : let mut tx = pool.begin().await?;
158 401 : crate::store::pg::set_service(&mut tx).await?;
159 401 : let n = sqlx::query(REAP_SQL)
160 401 : .bind(hours)
161 401 : .execute(&mut *tx)
162 401 : .await?
163 401 : .rows_affected();
164 401 : tx.commit().await?;
165 401 : Ok(n)
166 401 : }
167 :
168 : #[cfg(test)]
169 : mod tests {
170 : use super::*;
171 : use sqlx::postgres::PgPoolOptions;
172 :
173 : /// A pool that has never connected: any statement issued through it fails
174 : /// immediately, so a test that succeeds proves no statement was issued.
175 : /// Constructing one spawns sqlx's idle reaper, so it is built inside the
176 : /// test's own runtime.
177 2 : fn unreachable_pool() -> PgPool {
178 2 : PgPoolOptions::new()
179 2 : .connect_lazy("postgres://nobody@127.0.0.1:1/antares_no_such_db")
180 2 : .expect("lazy pool")
181 2 : }
182 :
183 : /// Acking an empty page must short-circuit: without the guard the drain
184 : /// opens a transaction (and `seq = ANY('{}')` scans) on every idle tick.
185 : #[tokio::test]
186 2 : async fn ack_of_nothing_issues_no_statement() {
187 2 : assert_eq!(ack(&unreachable_pool(), &[]).await.expect("noop ack"), 0);
188 2 : }
189 :
190 : /// The ack deletes the published seqs one by one. A range form
191 : /// (`seq <= max`) also deletes a lower-seq row that committed between peek
192 : /// and ack — an event dropped without ever being published.
193 : #[test]
194 2 : fn ack_deletes_by_exact_seq_never_by_range() {
195 2 : assert!(ACK_SQL.contains("seq = ANY($1)"));
196 2 : assert!(
197 2 : !ACK_SQL.contains("<="),
198 : "range ack loses gap rows: {ACK_SQL}"
199 : );
200 2 : assert!(!ACK_SQL.contains('<'));
201 2 : }
202 :
203 : /// The drain's page size is bound, not interpolated, and the page is
204 : /// oldest-first — publishing order is the commit order the bus dedups on.
205 : #[test]
206 2 : fn peek_is_bounded_and_oldest_first() {
207 2 : assert!(PEEK_SQL.contains("ORDER BY seq"));
208 2 : assert!(PEEK_SQL.contains("LIMIT $1"));
209 2 : assert!(
210 2 : !PEEK_SQL.contains("LIMIT {"),
211 : "page size must not be formatted in"
212 : );
213 2 : }
214 :
215 : /// Both enqueue paths write the tenant column from a bound parameter; the
216 : /// batch form stays ONE statement (a per-item loop would cost N round
217 : /// trips inside every write transaction).
218 : #[test]
219 2 : fn enqueue_binds_the_tenant_and_batches_in_one_statement() {
220 4 : for sql in [ENQUEUE_SQL, ENQUEUE_MANY_SQL] {
221 4 : assert!(sql.contains("INSERT INTO outbox (tenant_id, event)"));
222 4 : assert!(sql.contains("$1"));
223 : }
224 2 : assert!(ENQUEUE_MANY_SQL.contains("jsonb_array_elements($2::jsonb)"));
225 2 : assert_eq!(ENQUEUE_MANY_SQL.matches("INSERT").count(), 1);
226 2 : }
227 : }
|
