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::connect 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).await?;
// `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 async fn connect( pool: PgPool, settings: PostgresOutboxSettings, serializer: Ser, ) -> Result<Self, PostgresStoreError>

Wraps pool with settings and serializer. Verifies once at construction, in order: that the connected server’s server_version_num meets crate::MIN_SERVER_VERSION_NUM (ADR 0041 — a wrong server version explains a missing relation, and the reverse is never true), then that the unqualified name outbox resolves to settings.schema: fails fast with PostgresStoreError::UnsupportedServerVersion, 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::UnsupportedServerVersion, PostgresStoreError::NotMigrated, PostgresStoreError::SchemaResolution, or PostgresStoreError::Database for a connection failure during verification.

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::connect(
    pool,
    PostgresOutboxSettings::default(),
    JsonSerializer,
)
.await?;

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).await?;
assert_eq!(store.content_type().as_str(), "application/json");

impl PostgresOutboxStore<JsonSerializer>

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

Available on crate feature json only.

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

Errors

Same as Self::connect.

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

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

Available on crate feature json only.

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

Errors

Same as Self::connect.

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_settings(
    pool,
    PostgresOutboxSettings::default().schema("orders"),
)
.await?;

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: 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, so there is no owner to check against.

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 (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 (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 (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, shown here against the test-support in-memory fake:

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 guarded by locked_by — the batch continues rather than failing outright (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.

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.

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 claim’s lease clause (locked_until IS NULL OR locked_until < now()), 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 locked_until/expires_at), dead at ix_outbox_dead_at, 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 = PostgresStoreError

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