rullst_orm/
outbox.rs

1//! Durable, database-backed transactional outbox.
2
3use serde_json::Value;
4use std::fmt;
5
6use crate::{Error, FromRow, Orm};
7
8const MAX_KEY_LEN: usize = 128;
9const MAX_PAYLOAD_BYTES: usize = 1_048_576;
10const MAX_ERROR_LEN: usize = 512;
11
12/// Result of an idempotent outbox enqueue operation.
13#[derive(Clone, Debug, Eq, PartialEq)]
14pub struct EnqueuedOutboxEvent {
15    /// Stable database identifier of the new or pre-existing event.
16    pub id: i64,
17    /// Whether this call inserted the row rather than finding its idempotency key.
18    pub inserted: bool,
19}
20
21/// Event exclusively claimed by one worker until its lease expires.
22#[derive(Clone, FromRow)]
23pub struct ClaimedOutboxEvent {
24    /// Stable database identifier.
25    pub id: i64,
26    /// Explicit application or tenant stream.
27    pub stream: String,
28    /// Idempotency key unique within the stream.
29    pub event_key: String,
30    /// Bounded event kind used for dispatch routing.
31    pub event_kind: String,
32    /// Serialized JSON payload.
33    pub payload_json: String,
34    /// Number of successful claims, including this one.
35    pub attempts: i32,
36    /// Unpredictable token required to acknowledge or fail this claim.
37    pub claim_key: String,
38    /// Unix timestamp when another worker may reclaim this event.
39    pub claim_expires_at_epoch: i64,
40}
41
42impl fmt::Debug for ClaimedOutboxEvent {
43    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
44        formatter
45            .debug_struct("ClaimedOutboxEvent")
46            .field("id", &self.id)
47            .field("stream", &"[REDACTED]")
48            .field("event_key", &"[REDACTED]")
49            .field("event_kind", &self.event_kind)
50            .field("payload_bytes", &self.payload_json.len())
51            .field("attempts", &self.attempts)
52            .field("claim_key", &"[REDACTED]")
53            .field("claim_expires_at_epoch", &self.claim_expires_at_epoch)
54            .finish()
55    }
56}
57
58impl ClaimedOutboxEvent {
59    /// Parses the bounded payload as JSON.
60    pub fn payload(&self) -> Result<Value, Error> {
61        serde_json::from_str(&self.payload_json).map_err(Into::into)
62    }
63}
64
65/// Stateless facade for the durable outbox table.
66pub struct Outbox;
67
68/// Versioned migration for applications that use the built-in migration runner.
69pub struct OutboxMigration;
70
71#[async_trait::async_trait]
72impl crate::schema::migration::Migration for OutboxMigration {
73    fn name(&self) -> &'static str {
74        "m20260830_000001_create_rullst_outbox"
75    }
76
77    async fn up(&self) -> Result<(), Error> {
78        Outbox::install().await
79    }
80
81    async fn down(&self) -> Result<(), Error> {
82        sqlx::query("DROP TABLE IF EXISTS rullst_outbox")
83            .execute(Orm::pool()?)
84            .await?;
85        Ok(())
86    }
87}
88
89impl Outbox {
90    /// Installs the portable outbox table and its delivery index.
91    ///
92    /// Release migrations should normally own this DDL. This helper is useful
93    /// for explicit application setup and tests; it never runs implicitly.
94    ///
95    /// On MySQL/MariaDB the `stream`, `event_key` and `claim_key` columns use
96    /// the binary `ascii_bin` collation, so keys compare case-sensitively as on
97    /// PostgreSQL and SQLite. An existing table whose key columns still use
98    /// another collation (created by an earlier version) is converted in place
99    /// with `ALTER TABLE`, which rebuilds the table.
100    pub async fn install() -> Result<(), Error> {
101        let driver = Orm::driver()?;
102        let table_sql = match driver {
103            "postgres" => POSTGRES_TABLE,
104            "mysql" => MYSQL_TABLE,
105            _ => SQLITE_TABLE,
106        };
107        sqlx::query(table_sql).execute(Orm::pool()?).await?;
108        if driver == "mysql" {
109            let (case_insensitive_keys,): (i64,) = sqlx::query_as(MYSQL_KEY_COLLATION_CHECK)
110                .fetch_one(Orm::pool()?)
111                .await?;
112            if case_insensitive_keys > 0 {
113                sqlx::query(MYSQL_KEY_COLLATION_UPGRADE)
114                    .execute(Orm::pool()?)
115                    .await?;
116            }
117        } else {
118            sqlx::query(
119                "CREATE INDEX IF NOT EXISTS rullst_outbox_delivery_idx \
120                 ON rullst_outbox (stream, status, available_at_epoch, claim_expires_at_epoch, id)",
121            )
122            .execute(Orm::pool()?)
123            .await?;
124        }
125        Ok(())
126    }
127
128    /// Enqueues an event inside the current [`Orm::transaction`].
129    ///
130    /// Refusing to open an implicit transaction prevents the domain mutation
131    /// and its outbox row from accidentally committing independently.
132    pub async fn enqueue(
133        stream: impl Into<String>,
134        event_key: impl Into<String>,
135        event_kind: impl Into<String>,
136        payload: &Value,
137    ) -> Result<EnqueuedOutboxEvent, Error> {
138        crate::__transaction_access::ensure_allowed()?;
139        let stream = stream.into();
140        let event_key = event_key.into();
141        let event_kind = event_kind.into();
142        let transaction = crate::CURRENT_TX
143            .try_with(Clone::clone)
144            .map_err(|_| missing_transaction_error())?;
145        let mut transaction = transaction.lock().await;
146        let transaction = transaction.as_mut().ok_or_else(missing_transaction_error)?;
147        Self::enqueue_with_tx_inner(transaction, &stream, &event_key, &event_kind, payload).await
148    }
149
150    /// Enqueues through a caller-owned SQLx transaction.
151    pub async fn enqueue_with_tx(
152        transaction: &mut crate::db::Transaction<'_>,
153        stream: impl Into<String>,
154        event_key: impl Into<String>,
155        event_kind: impl Into<String>,
156        payload: &Value,
157    ) -> Result<EnqueuedOutboxEvent, Error> {
158        let stream = stream.into();
159        let event_key = event_key.into();
160        let event_kind = event_kind.into();
161        Self::enqueue_with_tx_inner(transaction, &stream, &event_key, &event_kind, payload).await
162    }
163
164    async fn enqueue_with_tx_inner(
165        transaction: &mut crate::db::Transaction<'_>,
166        stream: &str,
167        event_key: &str,
168        event_kind: &str,
169        payload: &Value,
170    ) -> Result<EnqueuedOutboxEvent, Error> {
171        validate_key("stream", stream)?;
172        validate_key("event_key", event_key)?;
173        validate_key("event_kind", event_kind)?;
174        let payload_json = serde_json::to_string(payload)?;
175        if payload_json.len() > MAX_PAYLOAD_BYTES {
176            return Err(Error::Validation(format!(
177                "outbox payload exceeds {MAX_PAYLOAD_BYTES} bytes"
178            )));
179        }
180        let now = unix_now()?;
181        let insert_token = uuid::Uuid::new_v4().simple().to_string();
182        let driver = Orm::driver()?;
183        let insert_sql = match driver {
184            "postgres" => POSTGRES_INSERT,
185            "mysql" => MYSQL_INSERT,
186            _ => SQLITE_INSERT,
187        };
188        sqlx::query(insert_sql)
189            .bind(stream)
190            .bind(event_key)
191            .bind(event_kind)
192            .bind(&payload_json)
193            .bind("pending")
194            .bind(0_i32)
195            .bind("")
196            .bind(0_i64)
197            .bind("")
198            .bind(now)
199            .bind(now)
200            .bind(&insert_token)
201            .execute(&mut **transaction)
202            .await?;
203        let select_sql = match driver {
204            "postgres" => POSTGRES_ENQUEUED_SELECT,
205            "mysql" => MYSQL_ENQUEUED_SELECT,
206            _ => SQLITE_ENQUEUED_SELECT,
207        };
208        let (id, stored_kind, stored_payload, stored_insert_token) =
209            sqlx::query_as::<_, (i64, String, String, String)>(select_sql)
210                .bind(stream)
211                .bind(event_key)
212                .fetch_optional(&mut **transaction)
213                .await?
214                .ok_or_else(|| {
215                    Error::DatabaseError(
216                        "outbox event could not be read back after its idempotent insert"
217                            .to_string(),
218                    )
219                })?;
220        if stored_kind != event_kind || stored_payload != payload_json {
221            return Err(Error::Validation(format!(
222                "outbox idempotency key '{event_key}' already exists in stream '{stream}' with different content"
223            )));
224        }
225        let inserted = stored_insert_token == insert_token;
226        Ok(EnqueuedOutboxEvent { id, inserted })
227    }
228
229    /// Claims one pending or lease-expired event for a stream.
230    pub async fn claim_next(
231        stream: impl Into<String>,
232        worker_id: impl Into<String>,
233        lease_seconds: i64,
234        max_attempts: i32,
235    ) -> Result<Option<ClaimedOutboxEvent>, Error> {
236        let stream = stream.into();
237        let worker_id = worker_id.into();
238        Self::claim_next_at_inner(
239            &stream,
240            &worker_id,
241            unix_now()?,
242            lease_seconds,
243            max_attempts,
244        )
245        .await
246    }
247
248    #[doc(hidden)]
249    pub async fn claim_next_at(
250        stream: impl Into<String>,
251        worker_id: impl Into<String>,
252        now_epoch_seconds: i64,
253        lease_seconds: i64,
254        max_attempts: i32,
255    ) -> Result<Option<ClaimedOutboxEvent>, Error> {
256        let stream = stream.into();
257        let worker_id = worker_id.into();
258        Self::claim_next_at_inner(
259            &stream,
260            &worker_id,
261            now_epoch_seconds,
262            lease_seconds,
263            max_attempts,
264        )
265        .await
266    }
267
268    async fn claim_next_at_inner(
269        stream: &str,
270        worker_id: &str,
271        now_epoch_seconds: i64,
272        lease_seconds: i64,
273        max_attempts: i32,
274    ) -> Result<Option<ClaimedOutboxEvent>, Error> {
275        validate_key("stream", stream)?;
276        validate_key("worker_id", worker_id)?;
277        if now_epoch_seconds <= 0
278            || !(1..=3_600).contains(&lease_seconds)
279            || !(1..=100).contains(&max_attempts)
280        {
281            return Err(Error::Validation(
282                "outbox claim timestamp, lease or attempt limit is outside its bound".to_string(),
283            ));
284        }
285        let claim_expires_at_epoch = now_epoch_seconds
286            .checked_add(lease_seconds)
287            .ok_or_else(|| Error::Validation("outbox claim lease overflowed".to_string()))?;
288        let claim_key = uuid::Uuid::new_v4().simple().to_string();
289        let driver = Orm::driver()?;
290        let mut transaction = Orm::begin_transaction().await?;
291        let exhaust_sql = if driver == "postgres" {
292            POSTGRES_EXHAUST
293        } else {
294            PORTABLE_EXHAUST
295        };
296        sqlx::query(exhaust_sql)
297            .bind("dead_letter")
298            .bind("")
299            .bind("")
300            .bind(0_i64)
301            .bind("claim attempt limit reached")
302            .bind(stream)
303            .bind(max_attempts)
304            .bind("pending")
305            .bind("processing")
306            .bind(now_epoch_seconds)
307            .execute(&mut *transaction)
308            .await?;
309
310        let select_sql = match driver {
311            "postgres" => POSTGRES_CLAIM_SELECT,
312            "mysql" => MYSQL_CLAIM_SELECT,
313            _ => PORTABLE_CLAIM_SELECT,
314        };
315        let candidate = sqlx::query_scalar::<_, i64>(select_sql)
316            .bind(stream)
317            .bind("pending")
318            .bind(now_epoch_seconds)
319            .bind("processing")
320            .bind(now_epoch_seconds)
321            .bind(max_attempts)
322            .fetch_optional(&mut *transaction)
323            .await?;
324        let Some(id) = candidate else {
325            transaction.commit().await?;
326            return Ok(None);
327        };
328
329        let update_sql = if driver == "postgres" {
330            POSTGRES_CLAIM_UPDATE
331        } else {
332            PORTABLE_CLAIM_UPDATE
333        };
334        let claimed = sqlx::query(update_sql)
335            .bind("processing")
336            .bind(worker_id)
337            .bind(&claim_key)
338            .bind(claim_expires_at_epoch)
339            .bind("")
340            .bind(id)
341            .bind(stream)
342            .bind("pending")
343            .bind(now_epoch_seconds)
344            .bind("processing")
345            .bind(now_epoch_seconds)
346            .bind(max_attempts)
347            .execute(&mut *transaction)
348            .await?
349            .rows_affected()
350            == 1;
351        if !claimed {
352            transaction.rollback().await?;
353            return Ok(None);
354        }
355
356        let fetch_sql = if driver == "postgres" {
357            POSTGRES_CLAIM_FETCH
358        } else {
359            PORTABLE_CLAIM_FETCH
360        };
361        let event = sqlx::query_as::<_, ClaimedOutboxEvent>(fetch_sql)
362            .bind(id)
363            .bind(stream)
364            .bind("processing")
365            .bind(&claim_key)
366            .fetch_one(&mut *transaction)
367            .await?;
368        transaction.commit().await?;
369        Ok(Some(event))
370    }
371
372    /// Marks an event delivered only when the exact claim token still owns it.
373    pub async fn acknowledge(id: i64, claim_key: impl Into<String>) -> Result<bool, Error> {
374        let claim_key = claim_key.into();
375        transition(id, &claim_key, Transition::Delivered).await
376    }
377
378    /// Releases an event for retry or dead-letters it at the attempt limit.
379    pub async fn fail(
380        id: i64,
381        claim_key: impl Into<String>,
382        error: impl Into<String>,
383        max_attempts: i32,
384        retry_delay_seconds: i64,
385    ) -> Result<bool, Error> {
386        let claim_key = claim_key.into();
387        let error = error.into();
388        let retry_at_epoch = unix_now()?
389            .checked_add(retry_delay_seconds)
390            .ok_or_else(|| Error::Validation("outbox retry timestamp overflowed".to_string()))?;
391        transition(
392            id,
393            &claim_key,
394            Transition::Failed {
395                error: &error,
396                max_attempts,
397                retry_at_epoch,
398                retry_delay_seconds,
399            },
400        )
401        .await
402    }
403}
404
405enum Transition<'a> {
406    Delivered,
407    Failed {
408        error: &'a str,
409        max_attempts: i32,
410        retry_at_epoch: i64,
411        retry_delay_seconds: i64,
412    },
413}
414
415async fn transition(id: i64, claim_key: &str, transition: Transition<'_>) -> Result<bool, Error> {
416    if id <= 0 {
417        return Err(Error::Validation("outbox id must be positive".to_string()));
418    }
419    validate_key("claim_key", claim_key)?;
420    let driver = Orm::driver()?;
421    let now_epoch_seconds = unix_now()?;
422    let result = match transition {
423        Transition::Delivered => {
424            let sql = if driver == "postgres" {
425                POSTGRES_ACK
426            } else {
427                PORTABLE_ACK
428            };
429            sqlx::query(sql)
430                .bind("delivered")
431                .bind("")
432                .bind("")
433                .bind(0_i64)
434                .bind(now_epoch_seconds)
435                .bind(id)
436                .bind("processing")
437                .bind(claim_key)
438                .bind(now_epoch_seconds)
439                .execute(Orm::pool()?)
440                .await?
441        }
442        Transition::Failed {
443            error,
444            max_attempts,
445            retry_at_epoch,
446            retry_delay_seconds,
447        } => {
448            if !(1..=100).contains(&max_attempts)
449                || !(0..=86_400).contains(&retry_delay_seconds)
450                || error.is_empty()
451                || error.len() > MAX_ERROR_LEN
452                || error.chars().any(char::is_control)
453            {
454                return Err(Error::Validation(
455                    "outbox failure policy is outside its bound".to_string(),
456                ));
457            }
458            let sql = if driver == "postgres" {
459                POSTGRES_FAIL
460            } else {
461                PORTABLE_FAIL
462            };
463            sqlx::query(sql)
464                .bind(max_attempts)
465                .bind("dead_letter")
466                .bind("pending")
467                .bind(retry_at_epoch)
468                .bind("")
469                .bind("")
470                .bind(0_i64)
471                .bind(error)
472                .bind(id)
473                .bind("processing")
474                .bind(claim_key)
475                .bind(now_epoch_seconds)
476                .execute(Orm::pool()?)
477                .await?
478        }
479    };
480    Ok(result.rows_affected() == 1)
481}
482
483fn validate_key(field: &str, value: &str) -> Result<(), Error> {
484    if value.is_empty()
485        || value.len() > MAX_KEY_LEN
486        || !value
487            .bytes()
488            .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_' | b':' | b'.'))
489    {
490        return Err(Error::Validation(format!(
491            "outbox {field} must contain 1-{MAX_KEY_LEN} ASCII letters, digits, '.', '-', '_' or ':'"
492        )));
493    }
494    Ok(())
495}
496
497fn unix_now() -> Result<i64, Error> {
498    let elapsed = std::time::SystemTime::now()
499        .duration_since(std::time::UNIX_EPOCH)
500        .map_err(|_| Error::Internal("system clock is before the Unix epoch".to_string()))?;
501    i64::try_from(elapsed.as_secs())
502        .map_err(|_| Error::Internal("Unix timestamp exceeds i64".to_string()))
503}
504
505fn missing_transaction_error() -> Error {
506    Error::Validation(
507        "Outbox::enqueue must run inside Orm::transaction; use enqueue_with_tx for a caller-owned transaction"
508            .to_string(),
509    )
510}
511
512const POSTGRES_TABLE: &str = "CREATE TABLE IF NOT EXISTS rullst_outbox (id BIGSERIAL PRIMARY KEY, stream VARCHAR(128) NOT NULL, event_key VARCHAR(128) NOT NULL, event_kind VARCHAR(128) NOT NULL, payload_json TEXT NOT NULL, status VARCHAR(16) NOT NULL, attempts INTEGER NOT NULL, claimed_by VARCHAR(128) NOT NULL, claim_key VARCHAR(128) NOT NULL, claim_expires_at_epoch BIGINT NOT NULL, last_error VARCHAR(512) NOT NULL, available_at_epoch BIGINT NOT NULL, created_at_epoch BIGINT NOT NULL, delivered_at_epoch BIGINT, insert_token VARCHAR(32) NOT NULL, UNIQUE (stream, event_key))";
513// MySQL/MariaDB default collations fold case; keys are case-sensitive ASCII.
514const MYSQL_TABLE: &str = "CREATE TABLE IF NOT EXISTS rullst_outbox (id BIGINT AUTO_INCREMENT PRIMARY KEY, stream VARCHAR(128) CHARACTER SET ascii COLLATE ascii_bin NOT NULL, event_key VARCHAR(128) CHARACTER SET ascii COLLATE ascii_bin NOT NULL, event_kind VARCHAR(128) NOT NULL, payload_json LONGTEXT NOT NULL, status VARCHAR(16) NOT NULL, attempts INT NOT NULL, claimed_by VARCHAR(128) NOT NULL, claim_key VARCHAR(128) CHARACTER SET ascii COLLATE ascii_bin NOT NULL, claim_expires_at_epoch BIGINT NOT NULL, last_error VARCHAR(512) NOT NULL, available_at_epoch BIGINT NOT NULL, created_at_epoch BIGINT NOT NULL, delivered_at_epoch BIGINT NULL, insert_token VARCHAR(32) NOT NULL, UNIQUE KEY rullst_outbox_stream_event_unique (stream, event_key), INDEX rullst_outbox_delivery_idx (stream, status, available_at_epoch, claim_expires_at_epoch, id))";
515const MYSQL_KEY_COLLATION_CHECK: &str = "SELECT COUNT(*) FROM information_schema.COLUMNS WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'rullst_outbox' AND COLUMN_NAME IN ('stream', 'event_key', 'claim_key') AND (COLLATION_NAME IS NULL OR COLLATION_NAME <> 'ascii_bin')";
516const MYSQL_KEY_COLLATION_UPGRADE: &str = "ALTER TABLE rullst_outbox MODIFY stream VARCHAR(128) CHARACTER SET ascii COLLATE ascii_bin NOT NULL, MODIFY event_key VARCHAR(128) CHARACTER SET ascii COLLATE ascii_bin NOT NULL, MODIFY claim_key VARCHAR(128) CHARACTER SET ascii COLLATE ascii_bin NOT NULL";
517const SQLITE_TABLE: &str = "CREATE TABLE IF NOT EXISTS rullst_outbox (id INTEGER PRIMARY KEY AUTOINCREMENT, stream TEXT NOT NULL, event_key TEXT NOT NULL, event_kind TEXT NOT NULL, payload_json TEXT NOT NULL, status TEXT NOT NULL, attempts INTEGER NOT NULL, claimed_by TEXT NOT NULL, claim_key TEXT NOT NULL, claim_expires_at_epoch BIGINT NOT NULL, last_error TEXT NOT NULL, available_at_epoch BIGINT NOT NULL, created_at_epoch BIGINT NOT NULL, delivered_at_epoch BIGINT, insert_token TEXT NOT NULL, UNIQUE (stream, event_key))";
518
519const POSTGRES_INSERT: &str = "INSERT INTO rullst_outbox (stream, event_key, event_kind, payload_json, status, attempts, claimed_by, claim_expires_at_epoch, last_error, available_at_epoch, created_at_epoch, insert_token, claim_key) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, '') ON CONFLICT (stream, event_key) DO NOTHING";
520const MYSQL_INSERT: &str = "INSERT INTO rullst_outbox (stream, event_key, event_kind, payload_json, status, attempts, claimed_by, claim_expires_at_epoch, last_error, available_at_epoch, created_at_epoch, insert_token, claim_key) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, '') ON DUPLICATE KEY UPDATE id = LAST_INSERT_ID(id)";
521const SQLITE_INSERT: &str = "INSERT INTO rullst_outbox (stream, event_key, event_kind, payload_json, status, attempts, claimed_by, claim_expires_at_epoch, last_error, available_at_epoch, created_at_epoch, insert_token, claim_key) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, '') ON CONFLICT (stream, event_key) DO NOTHING";
522
523const POSTGRES_ENQUEUED_SELECT: &str = "SELECT id, event_kind, payload_json, insert_token FROM rullst_outbox WHERE stream = $1 AND event_key = $2";
524// InnoDB's default REPEATABLE READ serves a plain SELECT from the snapshot of
525// the caller's first read, which misses a key committed later by a concurrent
526// enqueue. A locking read returns the latest committed row, which the
527// preceding upsert has already locked.
528const MYSQL_ENQUEUED_SELECT: &str = "SELECT id, event_kind, payload_json, insert_token FROM rullst_outbox WHERE stream = ? AND event_key = ? FOR UPDATE";
529const SQLITE_ENQUEUED_SELECT: &str = "SELECT id, event_kind, payload_json, insert_token FROM rullst_outbox WHERE stream = ? AND event_key = ?";
530
531const POSTGRES_EXHAUST: &str = "UPDATE rullst_outbox SET status = $1, claimed_by = $2, claim_key = $3, claim_expires_at_epoch = $4, last_error = $5 WHERE stream = $6 AND attempts >= $7 AND (status = $8 OR (status = $9 AND claim_expires_at_epoch <= $10))";
532const PORTABLE_EXHAUST: &str = "UPDATE rullst_outbox SET status = ?, claimed_by = ?, claim_key = ?, claim_expires_at_epoch = ?, last_error = ? WHERE stream = ? AND attempts >= ? AND (status = ? OR (status = ? AND claim_expires_at_epoch <= ?))";
533const POSTGRES_CLAIM_SELECT: &str = "SELECT id FROM rullst_outbox WHERE stream = $1 AND ((status = $2 AND available_at_epoch <= $3) OR (status = $4 AND claim_expires_at_epoch <= $5)) AND attempts < $6 ORDER BY id ASC LIMIT 1 FOR UPDATE SKIP LOCKED";
534const MYSQL_CLAIM_SELECT: &str = "SELECT id FROM rullst_outbox WHERE stream = ? AND ((status = ? AND available_at_epoch <= ?) OR (status = ? AND claim_expires_at_epoch <= ?)) AND attempts < ? ORDER BY id ASC LIMIT 1 FOR UPDATE SKIP LOCKED";
535const PORTABLE_CLAIM_SELECT: &str = "SELECT id FROM rullst_outbox WHERE stream = ? AND ((status = ? AND available_at_epoch <= ?) OR (status = ? AND claim_expires_at_epoch <= ?)) AND attempts < ? ORDER BY id ASC LIMIT 1";
536const POSTGRES_CLAIM_UPDATE: &str = "UPDATE rullst_outbox SET status = $1, attempts = attempts + 1, claimed_by = $2, claim_key = $3, claim_expires_at_epoch = $4, last_error = $5 WHERE id = $6 AND stream = $7 AND ((status = $8 AND available_at_epoch <= $9) OR (status = $10 AND claim_expires_at_epoch <= $11)) AND attempts < $12";
537const PORTABLE_CLAIM_UPDATE: &str = "UPDATE rullst_outbox SET status = ?, attempts = attempts + 1, claimed_by = ?, claim_key = ?, claim_expires_at_epoch = ?, last_error = ? WHERE id = ? AND stream = ? AND ((status = ? AND available_at_epoch <= ?) OR (status = ? AND claim_expires_at_epoch <= ?)) AND attempts < ?";
538const POSTGRES_CLAIM_FETCH: &str = "SELECT id, stream, event_key, event_kind, payload_json, attempts, claim_key, claim_expires_at_epoch FROM rullst_outbox WHERE id = $1 AND stream = $2 AND status = $3 AND claim_key = $4";
539const PORTABLE_CLAIM_FETCH: &str = "SELECT id, stream, event_key, event_kind, payload_json, attempts, claim_key, claim_expires_at_epoch FROM rullst_outbox WHERE id = ? AND stream = ? AND status = ? AND claim_key = ?";
540const POSTGRES_ACK: &str = "UPDATE rullst_outbox SET status = $1, claimed_by = $2, claim_key = $3, claim_expires_at_epoch = $4, delivered_at_epoch = $5 WHERE id = $6 AND status = $7 AND claim_key = $8 AND claim_expires_at_epoch > $9";
541const PORTABLE_ACK: &str = "UPDATE rullst_outbox SET status = ?, claimed_by = ?, claim_key = ?, claim_expires_at_epoch = ?, delivered_at_epoch = ? WHERE id = ? AND status = ? AND claim_key = ? AND claim_expires_at_epoch > ?";
542const POSTGRES_FAIL: &str = "UPDATE rullst_outbox SET status = CASE WHEN attempts >= $1 THEN $2 ELSE $3 END, available_at_epoch = $4, claimed_by = $5, claim_key = $6, claim_expires_at_epoch = $7, last_error = $8 WHERE id = $9 AND status = $10 AND claim_key = $11 AND claim_expires_at_epoch > $12";
543const PORTABLE_FAIL: &str = "UPDATE rullst_outbox SET status = CASE WHEN attempts >= ? THEN ? ELSE ? END, available_at_epoch = ?, claimed_by = ?, claim_key = ?, claim_expires_at_epoch = ?, last_error = ? WHERE id = ? AND status = ? AND claim_key = ? AND claim_expires_at_epoch > ?";
544
545#[cfg(test)]
546mod tests {
547    use super::{MYSQL_KEY_COLLATION_UPGRADE, MYSQL_TABLE};
548
549    #[test]
550    fn mysql_outbox_keys_compare_case_sensitively() {
551        for column in ["stream", "event_key", "claim_key"] {
552            let declaration =
553                format!("{column} VARCHAR(128) CHARACTER SET ascii COLLATE ascii_bin NOT NULL");
554            assert!(MYSQL_TABLE.contains(&declaration), "{column} table DDL");
555            assert!(
556                MYSQL_KEY_COLLATION_UPGRADE.contains(&format!("MODIFY {declaration}")),
557                "{column} upgrade DDL"
558            );
559        }
560    }
561}
rullst_orm/outbox.rs

rullst_orm/
outbox.rs
