Outbox

Use outbox to fix the dual-write hazard: a booking row and its booking.created event are two writes, and the row can commit while the event is lost. Store.Record(ctx, tx, msg) persists the message inside the caller’s db.Tx, so both commit or neither does. A relay then publishes it at-least-once, and Inbox makes consumers idempotent.

Use it for domain events, webhook fan-out, cache invalidation, and audit trails. Delivery is at-least-once, so pair it with Inbox (or your own idempotency key) on the consumer.

Record an event and consume it once

One transaction, one event, one idempotent consumer side effect:

main.go

package main

import (
  "context"
  "fmt"

  outboxdb "github.com/zenta-dev/zever/adapters/outbox/db"
  "github.com/zenta-dev/zever/config"
  "github.com/zenta-dev/zever/container"
  coredb "github.com/zenta-dev/zever/core/db"
  "github.com/zenta-dev/zever/core/outbox"
)

func main() {
  cfg := config.Default()
  cfg.DB.Options.Path = ":memory:"
  c := container.New(cfg)

  conn, err := c.DB()
  if err != nil {
    panic(err)
  }

  ob, err := outboxdb.OpenFromDB(conn, outboxdb.Options{
    Publisher: outbox.PublisherFunc(func(_ context.Context, msg outbox.Message) error {
      fmt.Println("published", msg.Topic, string(msg.Payload))
      return nil
    }),
  })
  if err != nil {
    panic(err)
  }
  defer ob.Close()

  ctx := context.Background()
  err = coredb.WithTx(ctx, conn, nil, func(ctx context.Context, tx coredb.Tx) error {
    if _, err := tx.Exec(ctx, `CREATE TABLE bookings (id TEXT PRIMARY KEY)`); err != nil {
      return err
    }
    if _, err := tx.Exec(ctx, `INSERT INTO bookings (id) VALUES (?)`, "b-1"); err != nil {
      return err
    }
    return ob.Record(ctx, tx, outbox.Message{
      ID:      "b-1",
      Topic:   "booking.created",
      Payload: []byte(`{"id":"b-1"}`),
    })
  })
  fmt.Println("recorded:", err)

  inbox, ok := ob.(outbox.Inbox)
  if !ok {
    panic("outbox store does not implement outbox.Inbox")
  }
  err = coredb.WithTx(ctx, conn, nil, func(ctx context.Context, tx coredb.Tx) error {
    return inbox.Process(ctx, tx, "b-1", func(ctx context.Context, tx coredb.Tx) error {
      fmt.Println("side effect ran")
      return nil
    })
  })
  fmt.Println("inbox:", err)
}

recorded: <nil>
side effect ran
inbox: <nil>

The relay is not started here. Start publishes stored messages through the same Publisher.

Record inserts in the caller’s transaction (a nil tx fails with ErrTxRequired) and injects trace context from ctx into the stored headers, so the consumer continues the trace.

Inbox.Process records eventID in the caller’s transaction and runs the side effect only when the ID is new (INSERT ... ON CONFLICT DO NOTHING), so at-least-once delivery applies side effects once.

Understand relay delivery and failure

• The relay claims a batch with an atomic UPDATE ... RETURNING (postgres adds FOR UPDATE SKIP LOCKED, sqlite serializes on a process-local mutex), publishes each message through the Publisher, marks successes processed, and retries failures with shared/retry backoff.
• Messages that exhaust max_attempts are marked failed: that is the dead-letter queue, left for inspection and repair through outbox.Admin (List/Requeue/Purge, plus Deleter on the db adapter), driven by the zever outbox CLI.
• Processed rows older than retention are deleted.
• Status() reports Pending, Processed, Failed, LastError, and Stalled (oldest pending waits too long).
• Any number of replicas can relay one table without double-publishing. If a replica crashes mid-publish, the row returns to pending when its lock_seconds lease expires.
• When an observability provider is wired (Options.Provider, which container.Outbox() injects), the relay emits outbox.* metrics and outbox.publish/outbox.consume spans (see Outbox Ops).

Configure the outbox

OUTBOX_ADAPTER=db
OUTBOX_DSN=postgres://app:secret@db:5432/app
OUTBOX_TABLE=outbox
OUTBOX_INBOX_TABLE=inbox
OUTBOX_PUBLISHER=eventbus
OUTBOX_POLL_INTERVAL=1s
OUTBOX_BATCH_SIZE=100
OUTBOX_MAX_ATTEMPTS=5
OUTBOX_RETENTION=604800000000000
OUTBOX_LOCK_SECONDS=30
OUTBOX_DEDICATED_POOL=false

OUTBOX_ADAPTER=db
OUTBOX_DSN=postgres://app:secret@db:5432/app
OUTBOX_POLL_INTERVAL=1000000000

Keys are strict snake_case with integer-nanosecond durations. Zero means the adapter default. Validate rejects negative intervals or budgets, unsafe table names, and a publisher outside {eventbus, queue, ""}. All keys: outbox config reference.

Core never imports eventbus or queue. Your wiring bridges a transport to the Publisher interface, and Options.Publisher only records which one.

Choose an adapter

Adapter	Package	Notes
db	adapters/outbox/db	Default. Durable sqlite or postgres polling relay, DLQ, retention. Also implements Inbox.
memory	adapters/outbox/memory	Dev and tests. Publishes synchronously and keeps nothing, so a crash between Record and the publisher’s ack loses the event.
cdc	adapters/outbox/cdc	Postgres logical replication via pg_logical_emit_message. No polling loop. Needs wal_level=logical.

The cdc adapter emits inside the caller’s transaction, so a rollback drops the event. Its consumer tails the WAL through a replication slot and acknowledges the LSN only after a successful publish.

OUTBOX_ADAPTER=cdc
OUTBOX_DSN=postgres://app:secret@db:5432/app
OUTBOX_PREFIX=zever_outbox
OUTBOX_SLOT=zever_outbox_slot
OUTBOX_PUBLICATION=zever_outbox_pub

The cdc adapter requires non-empty dsn, slot, publication, and prefix. Live cdc tests are gated behind POSTGRES_DSN.

Resolve the store from the container

main.go

import (
  "github.com/zenta-dev/zever/config"
  "github.com/zenta-dev/zever/container"
)

func openStore() error {
  c := container.New(config.Default())
  ob, err := c.Outbox()
  if err != nil {
    return err
  }
  defer ob.Close()
  return nil
}

Outbox() resolves lazily and caches; on error the cache clears so the next call retries. It also injects the resolved observability.Provider into Options.Provider, so a container-opened store emits the relay metrics and spans described in Outbox Ops without extra wiring. The container-opened store records messages but cannot Start its relay, because core Options carry no Publisher. Wire one through the adapter constructor (OpenFromDB in adapters/outbox/db, New in adapters/outbox/cdc and adapters/outbox/memory), as above.

Production checklist

• Store.Close() error takes no context. The container joins close errors via errors.Join, and panics become *ClosePanicError.
• The db adapter shares one pool per exact DSN (first opener’s max_conns wins). Use dedicated_pool: true under sustained load.
• Payloads can carry PII: keep them small and reference-backed, and never log raw option maps (use config.RedactedServices).

Troubleshooting

Start fails with “publisher is nil”

Cause: the store was opened through the container or registry path, which carries no Publisher.

Fix: wire one through the adapter constructor (OpenFromDB or New) as in the example.

Messages sit pending forever

Cause: the relay is not running (Start never called), or Status().Stalled is true.

Fix: call Start, and check LastError for the claim or publish error.

Failed rows accumulate

Cause: the destination rejects the payload, or the Publisher errors. max_attempts (default 5) bounds retries.

Fix: inspect last_error on the row, fix the cause, and re-drive the row.

Consumers apply duplicate side effects

Cause: the consumer is not idempotent.

Fix: give every message a stable ID and dedupe through Inbox.Process (or your own idempotency key) in the same transaction as the side effect.

The cdc adapter fails to start

Cause: the server lacks wal_level=logical, or the publication and slot names do not match the configured values.

Fix: set wal_level=logical and align the names. Live tests run with POSTGRES_DSN set.

Related

Events and event busIn-process fan-out, the usual Publisher bridge target.

QueueDurable delivery with ack and retry for must-run work.

WorkflowsDurable multi-step runs.

ObservabilityTrace context injected at Record time.

Outbox OpsRelay metrics, alerting, DLQ runbook, stalled readiness.