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 (contract §4, review 1 B2): under --no-default-features there is no default, so Self::connect is the only constructor and cargo hack --feature-powerset compiles every combination.

Implementations

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

pub async fn connect( pool: PgPool, settings: PostgresOutboxSettings, serializer: Ser, ) -> Result<Self, PostgresStoreError>

Wraps pool with settings and serializer. Verifies once at construction that the unqualified name outbox resolves to settings.schema: fails fast with PostgresStoreError::SchemaResolution (search_path problem) or PostgresStoreError::NotMigrated (the relation is missing entirely) rather than surprising the first acquire. Logs a tracing::warn! when a same-named table also exists in another schema on the path.

Errors

Returns PostgresStoreError::NotMigrated, PostgresStoreError::SchemaResolution, or PostgresStoreError::Database for a connection failure during verification.

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 (contract §4).

pub async fn enqueue<T: Message>( &self, tx: &mut Transaction<'_, Postgres>, envelope: &Envelope<T>, ) -> Result<MessageId, EnqueueError<Ser::Error>>

Stages a message in the application’s own transaction — atomicity is visible in the signature. Plain INSERT, no ON CONFLICT: a reused MessageId aborts the caller’s transaction rather than silently losing a message. Returns the id it wrote, so the caller can use it as the next message’s causation_id in the same transaction.

Errors

Returns EnqueueError::Serialize if the configured Serializer rejects the body, EnqueueError::Duplicate for a reused MessageId, or EnqueueError::Database for any other sqlx failure.

pub async fn enqueue_with<T: Message>( &self, tx: &mut Transaction<'_, Postgres>, envelope: &Envelope<T>, options: EnqueueOptions<'_>, ) -> Result<MessageId, EnqueueError<Ser::Error>>

Same as Self::enqueue, with provider-side options (currently EnqueueOptions::ordering_key).

Errors

Same as Self::enqueue.

impl PostgresOutboxStore<JsonSerializer>

pub async fn new(pool: PgPool) -> Result<Self, PostgresStoreError>

Convenience over Self::connect, behind the crate’s default json feature.

Errors

Same as Self::connect.

pub async fn with_settings( pool: PgPool, settings: PostgresOutboxSettings, ) -> Result<Self, PostgresStoreError>

Convenience over Self::connect with explicit settings, behind the crate’s default json feature.

Errors

Same as Self::connect.

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 sequence ASC is normative (contract §3.4): after_sequence is a keyset cursor over sequence, the column ix_outbox_dead orders by; message_type/ tenant_id/dead_before are filters only, expressed as ($n::type IS NULL OR ...) so one static statement serves every combination. The cursor returned is the largest sequence scanned, poisoned rows included, so a poisoned tail cannot loop the caller forever.

async fn retry_dead(&self, refs: &[MessageRef]) -> 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 worker-guarded — a dead row holds no lease (contract §3.4).

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

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

type Error = PostgresStoreError

A failure of the call.

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 (SRS §24.1, 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 guarded by locked_by — the batch continues (§19.5, ADR 0008).

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

Marks rows published, worker-guarded (locked_by = $2). A row already completed or reclaimed by another worker contributes nothing to the count — a shortfall is logged at debug, never an error (ADR 0008).

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

Applies each item’s FailureOutcome, worker-guarded. 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: &[MessageRef], ) -> Result<u64, Self::Error>

Clears the lease for rows this worker still owns. available_at and attempts are untouched — a release is not a failure (SRS §26.1).

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

Renews locked_until = now() + lease for rows this worker still owns. Best-effort: a shortfall means the lease already expired (§21.1).

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

One bounded pass, three statements, each capped at request.batch_size (contract §7 G1): 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 claim’s lease clause (locked_until IS NULL OR locked_until < now()), so it never transitions a row a live worker still owns (contract §7 G2) — 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 FILTER-qualified aggregates over a single scan of outbox (contract §4, S8 EXPLAIN comparison, RELIAR-17 card Log). An earlier version issued four separate statements, one per ix_outbox_pending/ix_outbox_dead_at/ix_outbox_expires — but pending’s and the min(available_at) row’s predicates aren’t a strict subset of ix_outbox_pending (they also filter on available_at/locked_until/expires_at, none of which the partial index’s WHERE clause covers), so the planner chose a Seq Scan for both anyway on a realistic seeded table (20k rows, a 25/25/25/25 pending/dead/ published/expired-pending mix) — meaning the four-statement form paid for that same Seq Scan twice (once for pending, once for the min/now() row) plus three extra round trips, for a strictly worse total (Execution Time 4.65 ms combined, Buffers: shared hit 836) than one statement computing all four aggregates from one scan (Execution Time 2.77 ms, Buffers: shared hit 412) — see the card Log for both full EXPLAIN (ANALYZE, BUFFERS) plans.

type Error = PostgresStoreError

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