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. Requires PostgreSQL 18 or later — a hard requirement, with no older-version fallback, checked at connect() and at migrate() (ADR 0041).

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):

   ALTER ROLE app SET search_path = reliar, public;
   

3. PostgresOutboxStore::connect/new verify, once at construction, that the unqualified name outbox resolves to the configured schema. An unresolvable name or a mismatch is a construction error naming the configured schema, the observed search_path, and the ALTER ROLE remedy — never a surprise failure on the first acquire. A same-named table found in another schema on the path is logged as a tracing::warn!.

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()).await?;

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()).
• 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.
• 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/PostgresOutboxSettings::schema accept 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::connect/new/with_settings — fail-fast startup search_path verification.
• 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), worker-guarded complete/fail/release/extend_lease, 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 schema/*_sets_search_path knobs plus 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.