reliar-store-postgres

Reliar's PostgreSQL provider for the transactional outbox: the schema, the explicit migrate() API, and PostgresOutboxStore — the only crate in the workspace where an sqlx/Postgres type appears.

Requirements

Requirements: PostgreSQL 18 or later. Reliar does not check the server version; behaviour on older servers is undefined. Neither a store constructor nor migrate() probes it — the schema uses uuidv7() and uuid_extract_timestamp(), so migrate() on an older server fails inside the first migration with PostgreSQL's own function uuidv7() does not exist (ADR 0047 Amendment B).

MSRV 1.94, set by sqlx 0.9 — six releases above the workspace floor of 1.88. Provider crates may carry their driver's MSRV; the pure reliar-core/reliar-outbox crates stay on 1.88 (ADR 0025).

Guarantees this store honours

PostgresOutboxStore implements reliar-outbox's contract, so its two headline guarantees are reliar-outbox's, not restated differently here — see reliar-outbox's README for the full text. In short:

• Durable at-least-once publication, never exactly-once. A consumer built on Reliar must be idempotent. Three windows produce a duplicate and all three are unavoidable in this release: the crash window (a publish reaches the broker, the worker crashes before complete persists, the lease expires, and another worker republishes), the slow-batch window (a batch outlives its lease while the worker is still healthily publishing, so a second worker reclaims and republishes the tail), and the drain window (cancellation drains in-flight publishes for at most drain_timeout; one still unresolved at the timeout is released rather than awaited further, carrying the same duplicate risk, just triggered by shutdown).
• No ordering by default. Ordering::Unordered (the only value this release supports) guarantees nothing about order — not globally, not per conversation_id, not per aggregate, not even approximately: acquire's SKIP LOCKED claim, concurrent publishing, per-message backoff and multiple dispatcher instances each reorder freely.

Features

Additive, checked with cargo hack check --feature-powerset.

search_path setup

Every Reliar object lives in one configurable schema, reliar by default, with unprefixed table names (outbox). sqlx::query! checks SQL at compile time, so every identifier in every statement is a static, unqualified literal — the schema is resolved at connection time through search_path, never compiled in (ADR 0017).

1. Put reliar first on the connection URL the host passes to its own pool:

   postgres://user:pw@host/app?options=-c%20search_path%3Dreliar,public
   

2. Behind a transaction-mode pooler — any pooler that drops startup options needs a server-side default instead, which every pooler mode honours (verify which yours does: PgDog, the pooler the suite tests against, passes them through). This is the only remedy for a pooler that drops options; there is no library-side alternative:

   ALTER ROLE app SET search_path = reliar, public;
   

3. Reliar does not verify this at startup (ADR 0047). Constructing a store issues no query; an unresolvable search_path surfaces at the first store call as PostgresOutboxError::NotMigrated/PostgresInboxError::NotMigrated, whose message names both the migrate() and the ALTER ROLE remedy. Reliar never sets search_path on a pool it does not own — not at construction, not per call.

4. migrate() does not depend on the caller's search_path — it creates the schema itself and sets search_path on its own dedicated connection before running the migration files, so it works even against a pool whose URL never set one (ADR 0018).

Usage

# #[derive(serde::Serialize, serde::Deserialize)]
# struct OrderCreated;
# impl reliar_core::Message for OrderCreated {
#     const TYPE: &'static str = "orders.created";
#     const VERSION: u16 = 1;
# }
# async fn run() -> Result<(), Box<dyn std::error::Error>> {
use reliar_outbox::OutboxEnqueue; // brings `enqueue`/`enqueue_envelope` into scope

let pool = sqlx::PgPool::connect("postgres://...").await?;

reliar_store_postgres::migrate(&pool, reliar_store_postgres::MigrateOptions::default()).await?;

let store = reliar_store_postgres::PostgresOutboxStore::new(pool.clone());

let mut tx = pool.begin().await?;
// ... write your own business row(s) in the same transaction ...
// Fire-and-forget: store.enqueue(&mut tx, /* your Message */ OrderCreated).await?;
// Propagating an inbound request's ids instead:
let envelope = reliar_core::Envelope::builder(OrderCreated).build();
store.enqueue_envelope(&mut tx, envelope).await?;
tx.commit().await?;
# Ok(()) }

What this crate ships

• migrations/0001_outbox.sql — the full v0.1 schema, every constraint/index explicitly named (pk_/ck_/ix_).
• migrations/0002_outbox_claimable_index.sql / 0003_drop_ix_outbox_pending.sql — replace ix_outbox_pending with ix_outbox_claimable (ADR 0040 §2): a claim now moves a leased row's available_at past its lease end, so the index the claim scans needs available_at in its own key rather than filtering locked_until from the heap. 0002 is a forward-only, -- no-transaction CREATE INDEX CONCURRENTLY; 0003 refuses to drop the old index unless the new one exists and is valid. If a CONCURRENTLY build is interrupted mid-run it leaves an invalid index — migrate()'s rustdoc carries the recovery step (DROP INDEX CONCURRENTLY ix_outbox_claimable;, then re-run migrate()). Historical: 0002 built the index with INCLUDE (locked_until, expires_at); 0015/0016 reduced it to INCLUDE (expires_at) once locked_until itself was dropped (ADR 0050).
• migrations/0015_outbox_claimable_index_without_locked_until.sql / 0016_drop_outbox_locked_until.sql — drop the locked_until column and ck_outbox_lease (ADR 0050): the lease is locked_by (who) + claim_token (which claim) + available_at (until when), and available_at is the only lease clock. Same two-file create-concurrently-then-swap shape as 0012/0014 — an index depends on its INCLUDE columns exactly as on its key columns, so a plain DROP COLUMN would silently drop ix_outbox_claimable along with it.
• migrations/0005_outbox_record_id.sql … 0010_outbox_primary_key_swap.sql — the outbox row gains its own id (OutboxRecordId, DEFAULT uuidv7(), pk_outbox) separate from the client-minted message_id it used to share one column with (ADR 0044). 0005 renames the old id column to message_id and adds the new, nullable id; 0006 backfills it (id := message_id, restartable — see its own doc comment for the batched escape hatch on a large table) and validates the NOT NULL check; 0007/0008/0009 build ix_outbox_id/ix_outbox_message_id/ix_outbox_dead_cursor CONCURRENTLY; 0010 promotes ix_outbox_id to pk_outbox, requires message_id NOT NULL, and drops the two indexes ix_outbox_dead_cursor supersedes (ix_outbox_dead, ix_outbox_dead_at) — each concurrent build's failure mode and recovery step (DROP INDEX CONCURRENTLY <name>;, then re-run migrate()) is in migrate()'s rustdoc. Run migrate() before starting 0.7.0 — see docs/guides/postgres.md for the rolling-upgrade procedure.
• migrations/0012_outbox_claimable_index_on_id.sql / 0013_outbox_ordering_key_index_on_id.sql / 0014_drop_outbox_sequence.sql — drop the sequence column (ADR 0049 Amendment A): sequence's only job was breaking a tie in ORDER BY available_at, sequence, and id (unique, NOT NULL, pk_outbox) already breaks it. 0012/0013 build ix_outbox_claimable/ix_outbox_ordering_key re-keyed on id under transient names (..._id) CONCURRENTLY; 0014 is an ordinary transactional migration that refuses to drop sequence and its indexes unless both transient indexes exist and are valid, then drops sequence and its indexes and renames the transient ones to the permanent names. If a CONCURRENTLY build is interrupted mid-run it leaves an invalid index and 0014 refuses to proceed, naming both indexes in its error — migrate()'s rustdoc carries the recovery step (DROP INDEX CONCURRENTLY ix_outbox_claimable_id; and/or ix_outbox_ordering_key_id, then re-run migrate()).
• migrate(&pool, MigrateOptions) — isolated bookkeeping in <schema>._migrations, never the shared _sqlx_migrations; serializes concurrent callers with its own poll-based advisory lock (not sqlx::migrate's blocking one, which would deadlock against 0002's CREATE INDEX CONCURRENTLY). MigrateOptions::schema accepts only a lowercase identifier ([a-z_][a-z0-9_$]*, at most 63 bytes) — PostgreSQL folds an unquoted identifier to lowercase, so an uppercase name is rejected rather than silently resolving inconsistently.
• PostgresOutboxStore::new/with_settings/with_serializer and PostgresInboxStore::new/with_settings — synchronous, no I/O (ADR 0047): a store can be built inside a OnceLock, a Default impl, or a synchronous main().
• reliar_outbox::OutboxEnqueue::enqueue/enqueue_envelope — the transactional write path; import the trait to call either.
• The full OutboxStore impl: acquire (the single-statement FOR UPDATE SKIP LOCKED claim, with poisoned-row handling — an undecodable row is moved to dead with DeadReason::Undecodable and reported, never a panic, the rest of the batch still delivers), claim_token-fenced complete/fail/release/extend_lease (ADR 0046 Amendment A), bounded purge (one pass, three LIMIT-capped statements), and stats.
• The full OutboxDeadLetters impl: list_dead (keyset-paginated by (dead_at, id), the row's own OutboxRecordId), retry_dead, purge_dead.
• Per-variant Classify for PostgresOutboxError/EnqueueError, including SQLSTATE-class-based classification of Database errors and 42P01 → NotMigrated on every operational path.
• migrations/0004_inbox.sql and PostgresInboxStore — the transactional inbox (ADR 0042 + Amendments A/B; inbox contract docs/architecture/inbox-contract.md), implementing reliar_inbox::InboxStore + InboxDeadLetters against sqlx::Transaction<'_, Postgres>. A separate type from PostgresOutboxStore sharing the same schema, the same migrate() and the same _migrations bookkeeping table: claim/complete run in the caller's own transaction (the whole of the inbox's guarantee — a handler's business write and the claim/completion commit or roll back together); fail/find/purge/InboxDeadLetters run on this store's own pool. claim's in-flight guard is a two-argument pg_try_advisory_xact_lock, released by the caller's own commit or rollback — never a blocking wait, so a redelivery racing an in-flight claim gets InProgress immediately instead of stalling. The row carries a client-minted UUIDv7 id (pk_inbox) plus the envelope's trace columns (message_type/conversation_id/correlation_id/causation_id), keyed for dedup by the unique index ix_inbox_scope_message_id. fail's single INSERT … ON CONFLICT DO UPDATE bounds recorded failures at PostgresInboxSettings::max_attempts (default 10) and sets dead_at atomically with the increment — a poison message stops being redelivered and its row becomes operator evidence in InboxDeadLetters::list_dead/retry_dead/purge_dead. PostgresInboxSettings mirrors PostgresOutboxSettings's statement_timeout knob plus its own max_attempts; PostgresInboxError is Classify, reusing the crate's SQLSTATE classification table.

Testing

Real-Postgres integration tests live in tests/ (skill testcontainers): one ephemeral postgres:18-alpine container per test binary, or DATABASE_URL when set (CI's service container), with one isolated database per test, migrated via the crate's own public migrate(). Run with Docker available:

cargo test -p reliar-store-postgres --all-features

.sqlx/ offline cache

cd crates/reliar-store-postgres
DATABASE_URL=postgres://user:pw@localhost/db?options=-c%20search_path%3Dreliar,public \
  cargo sqlx prepare -- --all-targets --all-features
git add .sqlx

CI builds with SQLX_OFFLINE=true and runs cargo sqlx prepare --check against a freshly migrated database.

License

MIT — see the workspace LICENSE.