Struct PostgresOutboxStore 

#[non_exhaustive]
pub struct PostgresOutboxStore<Ser = JsonSerializer> { /* private fields */ }

Reliar’s PostgreSQL outbox provider. Cheap to clone into an AppState — it wraps a PgPool; no outer Arc required. The connection pool stays the host’s: Reliar never owns or reads a DATABASE_URL.

The default type parameter only exists behind the crate’s default json feature: under --no-default-features there is no default, so Self::with_serializer is the only constructor and cargo hack --feature-powerset compiles every combination. This block’s PostgresOutboxStore::new leans on that default, so it only compiles under json; without it this block still shows the shape but is not compiled.

use reliar_store_postgres::{PostgresOutboxStore, migrate};
use sqlx::postgres::PgPoolOptions;

let pool = PgPoolOptions::new()
    .connect(&std::env::var("DATABASE_URL")?)
    .await?;
migrate(&pool, Default::default()).await?;

let store = PostgresOutboxStore::new(pool);
// `store` now implements `OutboxEnqueue`, `OutboxStore` and `OutboxDeadLetters` —
// hand it to an application's write path and to an `OutboxDispatcher`.

Implementations

impl<Ser: Serializer + Send + Sync + 'static> PostgresOutboxStore<Ser>

pub fn with_serializer( pool: PgPool, settings: PostgresOutboxSettings, serializer: Ser, ) -> Self

Wraps pool with settings and serializer. Performs no I/O: it issues no query, opens no connection and verifies nothing about the database. The pool stays the host’s.

Call crate::migrate (or apply the published SQL through your own pipeline) before the first store call, and make sure the connection’s search_path resolves the unqualified name outbox to the migrated schema — see the crate docs. An un-migrated or unreachable table surfaces at the first statement as PostgresOutboxError::NotMigrated, never here.

This example uses reliar_core::JsonSerializer, gated on the default json feature; without it this block still shows the shape but is not compiled.

use reliar_core::JsonSerializer;
use reliar_store_postgres::{PostgresOutboxSettings, PostgresOutboxStore};
use sqlx::postgres::PgPoolOptions;

let pool = PgPoolOptions::new()
    .connect(&std::env::var("DATABASE_URL")?)
    .await?;
let store =
    PostgresOutboxStore::with_serializer(pool, PostgresOutboxSettings::default(), JsonSerializer);

pub fn content_type(&self) -> &ContentType

The ContentType this store writes to every row — Serializer::content_type(). The only way a caller can predict the content_type of an envelope it will later acquire: enqueue writes this value, ignoring whatever envelope.metadata.delivery.content_type held. PostgresOutboxStore::new here leans on the default type parameter, gated on the default json feature; without it this block still shows the shape but is not compiled.

use reliar_store_postgres::PostgresOutboxStore;

let store = PostgresOutboxStore::new(pool);
assert_eq!(store.content_type().as_str(), "application/json");

impl PostgresOutboxStore<JsonSerializer>

pub fn new(pool: PgPool) -> Self

Available on crate feature json only.

Convenience over Self::with_serializer with reliar_core::JsonSerializer and PostgresOutboxSettings::default, behind the crate’s default json feature. Performs no I/O — see Self::with_serializer.

use reliar_store_postgres::PostgresOutboxStore;
use sqlx::postgres::PgPoolOptions;

let pool = PgPoolOptions::new()
    .connect(&std::env::var("DATABASE_URL")?)
    .await?;
let store = PostgresOutboxStore::new(pool);

pub fn with_settings(pool: PgPool, settings: PostgresOutboxSettings) -> Self

Available on crate feature json only.

Convenience over Self::with_serializer with reliar_core::JsonSerializer and explicit settings, behind the crate’s default json feature. Performs no I/O — see Self::with_serializer.

use reliar_store_postgres::{PostgresOutboxSettings, PostgresOutboxStore};
use sqlx::postgres::PgPoolOptions;
use std::time::Duration;

let pool = PgPoolOptions::new()
    .connect(&std::env::var("DATABASE_URL")?)
    .await?;
let store = PostgresOutboxStore::with_settings(
    pool,
    PostgresOutboxSettings::default().statement_timeout(Duration::from_secs(2)),
);

Trait Implementations

impl<Ser> Clone for PostgresOutboxStore<Ser>

Manual impl, never derived: a derived Clone would condition on Ser: Clone. The serializer is held as Arc<Ser> — stateless and cheap to share — so cloning the store never requires the serializer itself to be Clone.

fn clone(&self) -> Self

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl<Ser> Debug for PostgresOutboxStore<Ser>

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl<Ser: Serializer + Send + Sync + 'static> OutboxDeadLetters for PostgresOutboxStore<Ser>

async fn list_dead( &self, query: DeadQuery, ) -> Result<DeadLetterPage, Self::Error>

ORDER BY dead_at ASC, id ASC is normative: after is a composite keyset cursor over the columns ix_outbox_dead_cursor orders by; message_type/tenant_id/ dead_before are filters only. The cursor returned comes from the last row scanned, poisoned rows included, so a poisoned tail cannot loop the caller forever.

async fn retry_dead(&self, refs: &[RecordRef]) -> Result<u64, Self::Error>

Returns dead rows to pending: clears the lease that already isn’t there, resets attempts to 0 (the only operation that does), keeps last_error for audit. Not guarded by claim_token at all — a dead row holds no claim, so there is nothing to check against (ADR 0046 Amendment A). Also clears claim_token on the resurrected row: the row died holding whatever token its last claim stamped, and a stale outcome write from that pre-death claim must not be able to match the row it resurrects into.

async fn purge_dead(&self, refs: &[RecordRef]) -> Result<u64, Self::Error>

Deletes dead rows by reference, regardless of PurgeRequest::dead_retention.

type Error = PostgresOutboxError

A failure of the call.

impl<'c, Ser> OutboxEnqueue<Transaction<'c, Postgres>> for PostgresOutboxStore<Ser>

where Ser: Serializer + Send + Sync + 'static,

PostgresOutboxStore’s OutboxEnqueue implementation: reuses insert_row. Implements only OutboxEnqueue::enqueue_envelope — the provided enqueue (bare T: Message) calls back into it, so the reliar.outbox.enqueue span fires exactly once per row for either spelling (ADR 0037 amendment A).

Ser: 'static — needed because the method reaches self.serializer (held as Arc<Ser>) across the .await in insert_row; without it the future fails to type-check (a borrowed type must outlive the generic parameters it references). Every concrete serializer (JsonSerializer or any owned one) is 'static, so no real host is excluded.

A single lifetime, 'c, quantified by the impl. With &mut Tx in the trait’s own method signature the reborrow lifetime is quantified by the method itself, so the higher-ranked “implementation is not general enough” trap an earlier OutboxEnqueueIn<&'a mut Transaction<'c, _>> shape had — where an explicit, implied-looking where 'c: 'a bound broke every tokio::spawn/Axum call site — cannot arise here; there is no second lifetime to accidentally bound. That regression guard lives as outbox_enqueue::enqueue_is_send_through_tokio_spawn in this crate’s Postgres suite.

Renamed from OutboxStaging/stage in 0.4.0; the store’s own typed enqueue/enqueue_with were folded into this impl since no inherent method may share a name with a trait method — a caller now needs use reliar_outbox::OutboxEnqueue; in scope to call store.enqueue(..)/store.enqueue_envelope(..). The trait’s serialized twin, enqueue_serialized, was cut before shipping.

fn enqueue_envelope<T: Message + Sync>( &self, tx: &mut Transaction<'c, Postgres>, typed_envelope: Envelope<T>, ) -> impl Future<Output = Result<MessageId, Self::Error>> + Send

Serializes envelope.body with this store’s configured Serializer and writes the serializer’s own content_type. Plain INSERT, no ON CONFLICT: a reused MessageId aborts the caller’s transaction rather than silently losing a message.

Errors

EnqueueError::Serialize if the configured Serializer rejects the body, EnqueueError::Duplicate for a reused MessageId (ix_outbox_message_id violation, ADR 0044 §1), or EnqueueError::Database for any other sqlx failure.

EnqueueError::Duplicate/EnqueueError::Database leave tx aborted: the failed INSERT puts the PostgreSQL transaction in the aborted state, so PostgreSQL rejects every subsequent statement on it, and every earlier write in that transaction is rolled back at commit. EnqueueError::Serialize is returned before any statement runs, so tx is untouched and stays usable.

type Error = EnqueueError<<Ser as Serializer>::Error>

What enqueuing fails with.

fn enqueue<T>( &self, tx: &mut Tx, body: T, ) -> impl Future<Output = Result<MessageId, Self::Error>> + Send

where T: Message + Sync,

The fire-and-forget spelling: builds body into an envelope with default metadata and a freshly rooted conversation — exactly Envelope::builder(body).build() — then enqueues it via Self::enqueue_envelope.

impl<Ser: Serializer + Send + Sync + 'static> OutboxStore for PostgresOutboxStore<Ser>

async fn acquire( &self, request: AcquireRequest, ) -> Result<AcquiredBatch, Self::Error>

The canonical single-statement claim (ADR 0006): a CTE SELECT … FOR UPDATE SKIP LOCKED feeding an UPDATE … RETURNING, so the row lock is released before this future resolves and no network I/O to a publisher can ever happen while it is held.

A row this call cannot decode is excluded from records, reported in poisoned, and moved to dead with DeadReason::Undecodable by a follow-up statement fenced by the claim token this claim just stamped (ADR 0046 Amendment A) — the batch continues rather than failing outright (ADR 0008).

async fn complete( &self, worker: &WorkerId, items: &[RecordRef], ) -> Result<u64, Self::Error>

Marks rows published, fenced by each item’s claim token (ADR 0046 Amendment A). A row already completed or reclaimed under a fresh token — by any worker, including this one — contributes nothing to the count; a shortfall is logged at warn, naming the fenced ids, never an error (ADR 0008, ADR 0046 §5).

async fn fail( &self, worker: &WorkerId, items: &[FailedRecord], ) -> Result<u64, Self::Error>

Applies each item’s FailureOutcome, fenced by each item’s claim token. Retry rows get available_at = now() + delay computed in SQL (ADR 0009); dead rows get dead_at/dead_reason set together (ck_outbox_dead_reason). Both increment attempts — on outcome, never on claim.

async fn release( &self, worker: &WorkerId, items: &[RecordRef], ) -> Result<u64, Self::Error>

Clears the lease for rows whose claim token still matches. available_at and attempts are untouched — a release is not a failure.

async fn extend_lease( &self, worker: &WorkerId, items: &[RecordRef], lease: Duration, ) -> Result<u64, Self::Error>

Renews the lease by moving available_at to now() + lease for rows whose claim token still matches, without rotating it (ADR 0046 Amendment A). available_at is the lease clock, and the only one (ADR 0050 §1). Best-effort: a shortfall means the claim was superseded.

async fn purge(&self, request: PurgeRequest) -> Result<PurgeReport, Self::Error>

One bounded pass, three statements, each capped at request.batch_size: published-row delete, dead-row delete, and the expired→dead sweep — none of the three is ever an unbounded DELETE/UPDATE. The sweep’s predicate carries the not-currently-leased guard (locked_by IS NULL OR available_at <= now(), ADR 0050 §2.3), so it never transitions a row a live worker still owns — that worker’s own complete/fail wins, and the row becomes sweepable only once its lease lapses.

async fn stats(&self) -> Result<OutboxStats, Self::Error>

One statement, four independently planned scalar subqueries (ADR 0040 §3; supersedes the earlier single-scan FILTER-aggregate form, which was O(table)). Each subquery is aimed at its own partial index — pending and oldest_pending_available_at at ix_outbox_claimable (an index-only scan can evaluate a filter on its INCLUDEd expires_at), dead at ix_outbox_dead_cursor, expired_pending at ix_outbox_expires — so the cost is O(claimable backlog)/O(dead rows)/O(expired rows), never O(table), and oldest_pending_available_at is a single-row LIMIT. One round trip, one transaction snapshot (now() evaluated once), so as_of and the four values are consistent with each other even though each is planned separately. Measured at 100k rows (mixed pending/leased/published/dead/expired) on a vacuumed table, every subquery plans as an index-only scan with zero heap fetches.

type Error = PostgresOutboxError

A failure of the call — never a property of one row’s content. Must self-classify via crate::Classify so the dispatcher’s run() can tell a transient outage from a permanent one (ADR 0014).