reliar-outbox

The storage-agnostic transactional outbox: the OutboxStore/OutboxDeadLetters capability traits (plus reliar_core::Publisher, re-exported here for convenience), the request/result types that cross their boundary, a pure RetryPolicy, the feature's settings (OutboxSettings), the OutboxMetrics hook, and the OutboxDispatcher worker loop.

MSRV 1.88, the workspace floor.

Depends only on reliar-core — no sqlx, no Postgres, no broker client. A provider crate (reliar-store-postgres) implements the traits here; this crate never depends on one.

Guarantees

Durable at-least-once publication. Never exactly-once. Duplicate delivery is expected, and a consumer built on Reliar must be idempotent. Three windows produce a duplicate, and all three are unavoidable in this release:

1. The crash window — a publish reaches the broker, the worker crashes before complete persists the outcome, the lease expires, and another worker republishes the same message. A crash is not the only way complete never lands: lease is also the outcome-write retry budget. A complete/fail call that keeps failing or timing out is retried on every loop iteration, but only for up to lease — past that, the row is abandoned to its lease (dropped from tracking, no longer renewed) rather than retried forever, so a perfectly healthy worker with a persistently failing complete produces exactly this same duplicate, no crash required.
2. The slow-batch window — no crash at all. A worker claims a batch under a lease shorter than the batch takes to drain; the lease expires while the worker is still healthily publishing, a second worker reclaims and republishes the tail, and the first worker's later complete/fail is rejected by the locked_by guard. Lease renewal and a per-publish timeout make this rare, not impossible — in practice it is the common window.
3. The drain window — on cancellation, run() drains in-flight publishes for at most DispatcherSettings::drain_timeout. A publish still unresolved at the timeout is released rather than awaited further, and its eventual outcome — success or failure — carries the same duplicate risk as the other two, just triggered by shutdown instead of a lease.

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. SKIP LOCKED, concurrent publishing, per-message backoff and multiple dispatcher instances each reorder freely (ADR 0013). Ordering::PerKey is a configuration error before 0.2.

Pure retry. RetryPolicy is I/O-free and clock-free — it returns a Duration, never a timestamp. The store applies it as available_at = now() + delay in SQL, so a worker's clock skew can never hot-loop a row or park it in the future (ADR 0009).

The library never reads the environment implicitly. Only OutboxSettings::from_env touches std::env, and only when called (ADR 0019).

See docs/architecture/outbox.md for the full delivery-path walkthrough and docs/architecture/phase1-contract.md for the frozen signatures.

Quickstart: enqueue vs publish

The object names the guarantee — there is no facade type joining the two (ADR 0036 amendment B). A provider store implements [OutboxEnqueue] directly: call store.enqueue (a bare message) or store.enqueue_envelope (propagating ids from an inbound request) for the durable path. A transport's own Publisher::publish sends now, with no Reliar guarantee at all — call it directly, with no wrapper in between.

# use reliar_core::{ContentType, Envelope, Message, Publisher as _, Serializer};
# use reliar_outbox::{InMemoryOutboxStore, InMemoryTransaction, OutboxEnqueue, RecordingPublisher};
# #[derive(serde::Serialize, serde::Deserialize)]
# struct OrderCreated;
# impl Message for OrderCreated {
#     const TYPE: &'static str = "orders.created";
#     const VERSION: u16 = 1;
# }
# struct RawJson;
# impl Serializer for RawJson {
#     type Error = serde_json::Error;
#     fn content_type(&self) -> &ContentType { &ContentType::JSON }
#     fn serialize<T: Message>(&self, body: &T) -> Result<bytes::Bytes, Self::Error> {
#         serde_json::to_vec(body).map(bytes::Bytes::from)
#     }
#     fn deserialize<T: Message>(&self, bytes: &[u8]) -> Result<T, Self::Error> {
#         serde_json::from_slice(bytes)
#     }
# }
# #[tokio::main(flavor = "current_thread")]
# async fn main() -> Result<(), Box<dyn std::error::Error>> {
let store = InMemoryOutboxStore::default();
let publisher = RecordingPublisher::default();

// enqueue: durable, at-least-once — visible only once the caller's own transaction commits, and
// published later by an OutboxDispatcher. A real provider (e.g. PostgresOutboxStore) implements
// the same trait against its own transaction type. A bare message gets default metadata and a
// freshly rooted conversation; use enqueue_envelope + Envelope::builder(..) instead when an id
// must propagate from an inbound request.
let mut tx = InMemoryTransaction;
store.enqueue(&mut tx, OrderCreated).await?;

// publish: bypasses the outbox entirely — sends now, through the transport, one attempt, no
// Reliar guarantee at all. The transport takes an already-serialized envelope, so the caller
// serializes once, exactly as it would for a bare transport publisher.
let envelope = Envelope::builder(OrderCreated).build();
let bytes = RawJson.serialize(&envelope.body)?;
let mut serialized = envelope.map_body(|_| bytes);
serialized.metadata.delivery.content_type = RawJson.content_type().clone();
publisher.publish(&serialized).await?;
# Ok(())
# }

See docs/guides/outbox-enqueue-and-publish.md for the full guarantee comparison and the open-transaction warning around calling a transport publisher directly. An app that only ever needs the bypass path — no durability, no dispatcher — can skip this crate entirely and use a transport publisher (e.g. reliar-transport-nats's NatsPublisher) directly against reliar-core alone.

What this crate ships

• The OutboxStore/OutboxDeadLetters traits and their request/result types (AcquireRequest, AcquiredBatch, CompletedMessage, FailedMessage, FailureOutcome, DeadReason, MessageRef, PurgeRequest/PurgeReport, OutboxStats, DeadQuery/DeadLetterPage, PoisonedRow), plus OutboxRecord and its builder.
• reliar_core::Publisher (re-exported here) + Classify/FailureKind — a publisher's error carries its own transient/permanent verdict; the dispatcher never guesses.
• RetryPolicy and the default ExponentialBackoff.
• OutboxDispatcher/OutboxDispatcherBuilder/DispatchError — bounded-concurrency claim → publish → batch complete/fail, half-lease extend_lease renewal, a stats_interval tick feeding OutboxMetrics, tracing spans (reliar.outbox.claim/publish/retry/dead), and graceful cancellation via tokio_util::sync::CancellationToken.
• OutboxSettings/DispatcherSettings/RetentionSettings, each Default + builder, with an opt-in from_env("RELIAR_OUTBOX_").
• OutboxMetrics/NoopMetrics — a static-dispatch hook, no exporter dependency.
• Behind the test-support feature: InMemoryOutboxStore, RecordingPublisher, ScriptedPublisher, RecordingMetrics — reused by provider crates and examples/.

Settings and environment variables

OutboxSettings::from_env("RELIAR_OUTBOX_") is opt-in — nothing in this crate reads the environment implicitly (ADR 0019). It starts from Default, overrides only the variables present, and returns Err for a present-but-unparseable or out-of-range value, never a silent fallback.

DispatcherSettings (worker-loop tunables):

RetentionSettings (purge tunables):

The provider-specific RELIAR_STORE_POSTGRES_* variables (SCHEMA, ENQUEUE_SETS_SEARCH_PATH, STATEMENT_TIMEOUT_MS) are documented in docs/guides/postgres.md.

Features

Testing

cargo test -p reliar-outbox --all-features. Every test lives in tests/ against the public API: in-memory fakes drive dispatcher behaviour (retry, dead, cancellation, concurrency bounds) under #[tokio::test(start_paused = true)], with no wall-clock sleeps. benches/outbox_throughput.rs (criterion, cargo bench -p reliar-outbox --features test-support) covers the dispatcher's own claim → publish → complete overhead.

License

MIT — see the workspace LICENSE.