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

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<'c, Ser> OutboxEnqueue<Transaction<'c, Postgres>> for PostgresOutboxStore<Ser>

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

PostgresOutboxStore’s OutboxEnqueue implementation: reuses insert_enqueued’s search_path handling and its insert_row helper. 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 (decision #42, ADR 0037 amendment A).

Ser: 'static — needed because the method reaches self.serializer (held as Arc<Ser>) across the .await in insert_enqueued; without it the future fails to type-check (E0310). 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 (decision #34); folded the store’s own typed enqueue/enqueue_with into this impl in decision #37 (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 (decision #38).

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 (pk_outbox violation), 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 (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).