outbox

package
v0.4.0 Latest Latest
Warning

This package is not in the latest version of its module.

Go to latest
Published: Sep 3, 2026 License: MIT Imports: 17 Imported by: 0

Documentation

Overview

Package outbox implements the transactional outbox pattern for PostgreSQL.

Pattern

Application code enqueues events inside the same database transaction as the domain change. A background Relay worker polls the outbox table, delivers events to registered handlers, and marks them as processed — or records failures after exhausting retries.

Guarantees

  • At-least-once delivery: the relay may deliver a message more than once on restart; handlers must be idempotent.
  • Lease ownership: every claim has a token and only its owner may record an outcome. After lease expiry, delivery can happen again (at-least-once), so handlers must remain idempotent; a stale worker cannot overwrite the newer owner's outcome. Active handlers renew their lease; a lost lease cancels the handler context and stops further renewals.
  • Atomic enqueue: Enqueue must be called within the domain transaction.

Index

Constants

View Source
const Schema = `` /* 617-byte string literal not displayed */

Schema bootstraps the outbox table for ephemeral tests only. Production deployments must apply outbox.Migrations() for upgrade-safe evolution.

Variables

This section is empty.

Functions

func Enqueue

func Enqueue(ctx context.Context, tx pgx.Tx, event events.Event, options ...EnqueueOption) error

Enqueue writes event into the official outbox_messages table within tx. It must be called inside the same transaction as the domain change.

func Migrations

func Migrations() []migrate.Source

Migrations returns the migrations required by the transactional outbox. It does not install Jobs infrastructure.

Types

type EnqueueOption

type EnqueueOption func(*enqueueConfig) error

EnqueueOption configures persistence of a single outbox message.

func WithMaxAttempts

func WithMaxAttempts(maxAttempts int) EnqueueOption

WithMaxAttempts sets the message-owned retry limit.

func WithProcessAfter

func WithProcessAfter(processAfter time.Time) EnqueueOption

WithProcessAfter delays delivery until the given time.

type Relay

type Relay struct {
	Pool     *pgxpool.Pool
	Registry *events.Registry
	Logger   *slog.Logger
	Config   RelayConfig
	Tracer   telemetry.Tracer
	Meter    telemetry.Meter
}

Relay is a background worker that polls the outbox table and delivers events. outbox_messages.max_attempts is the sole authority for retry limits.

func (*Relay) Name

func (r *Relay) Name() string

Name implements engine.Worker.

func (*Relay) Run

func (r *Relay) Run(ctx context.Context) error

Run starts the relay loop. It returns when ctx is cancelled.

type RelayConfig

type RelayConfig struct {
	// BatchSize is the number of messages fetched per poll cycle.
	BatchSize int
	// PollInterval is how often the relay checks for pending messages.
	PollInterval time.Duration
	// InitialBackoff is the base delay before the first retry.
	InitialBackoff time.Duration
	// LeaseDuration is the maximum time a relay owns a claimed message.
	LeaseDuration time.Duration
	// LeaseRenewalInterval controls how often an active handler renews its
	// lease. Zero renews at half LeaseDuration. Values at or above the lease
	// duration are clamped to half the lease to prevent expiry before renewal.
	LeaseRenewalInterval time.Duration
}

RelayConfig configures the Relay worker.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL