outbox

README

go-outbox

go-outbox is a PostgreSQL-first Go implementation of the transactional outbox pattern. It writes application state and publishable envelopes in the same caller-owned pgx transaction, then relays committed envelopes to a small publisher contract with at-least-once delivery.

Version 1 guarantees the compatibility surfaces described in the compatibility policy. Delivery remains at least once; upgrading the library does not remove the consumer's idempotency duty.

Guarantees

• Atomic application and outbox persistence only when both writes use the same successful pgx.Tx.
• At-least-once relay delivery. Publisher acceptance followed by a failed or ambiguous delivered update can publish the same envelope again.
• Concurrent claims use PostgreSQL row locks and SKIP LOCKED.
• Every mutation of a leased record requires its current opaque lease token.
• Batch, worker, lease, retry, administrative, payload, and polling limits are explicit and bounded.
• Optional ordering-key or topic serialization is enforced at the PostgreSQL claim seam across relay processes. There is no global ordering guarantee.

Consumers must be idempotent. This project does not provide distributed transactions or exactly-once delivery.

Packages

• github.com/faustbrian/go-outbox: envelope construction and validation.
• github.com/faustbrian/go-outbox/postgres: migrations, transactional writer, claims, leases, retries, dead letters, replay, and retention.
• github.com/faustbrian/go-outbox/relay: bounded embedded relay.
• github.com/faustbrian/go-outbox/adapters/goqueue: separately versioned go-queue publisher adapter; importing core does not add go-queue.
• github.com/faustbrian/go-outbox/adapters/gotelemetry: separately versioned metrics and trace-linkage integration compatible with go-telemetry.

Quick start

builder, err := outbox.NewEnvelopeBuilder()
if err != nil {
    return err
}

envelope, err := builder.Build(outbox.NewEnvelopeParams{
    Topic:          "orders.created",
    Payload:        payload,
    OrderingKey:    customerID,
    IdempotencyKey: commandID,
})
if err != nil {
    return err
}

tx, err := pool.Begin(ctx)
if err != nil {
    return err
}
defer tx.Rollback(ctx)

if _, err := tx.Exec(ctx, insertOrderSQL, orderID); err != nil {
    return err
}
writer, err := postgres.NewWriter(postgres.WriterConfig{})
if err != nil {
    return err
}
if err := writer.Insert(ctx, tx, envelope); err != nil {
    return err
}

return tx.Commit(ctx)

The writer never opens or commits a transaction. Passing a pool or standalone connection is impossible because the API requires pgx.Tx.

See the documentation index, full quickstart, delivery guarantees, and architecture and crash matrix.

Development gates

go test ./...
go test -race -tags=integration ./...
go test -tags=integration -coverprofile=coverage.out ./...
(cd adapters/goqueue && go test -race ./...)

Integration tests use ephemeral Testcontainers PostgreSQL instances. They do not use an existing application or production database.

Status

Version 1 includes the core state machine, concurrency tests, payload-safe lifecycle events, health diagnostics, PostgreSQL backlog statistics, go-queue and telemetry adapters, CI matrices, fuzzing, benchmarks, and archive-before-delete retention.

License

MIT. See LICENSE.

Contribution, conduct, support, and vulnerability-reporting policies are in CONTRIBUTING.md, CODE_OF_CONDUCT.md, SUPPORT.md, and SECURITY.md.

Documentation

Overview

Package outbox provides transactional outbox records and relay primitives.

Persisting an Envelope atomically with application state requires passing the same caller-owned database transaction to a transactional store. Publishing is at least once: consumers must tolerate duplicate envelopes.

Example (DuplicateDeliveryRequiresIdempotentConsumer)

package main

import (
	"fmt"

	"github.com/faustbrian/go-outbox"
)

func main() {
	consumer := idempotentConsumer{seen: make(map[string]struct{})}
	envelope := outbox.Envelope{ID: "evt-42", Topic: "orders.created"}

	consumer.Consume(envelope)
	consumer.Consume(envelope) // relay redelivery after an ambiguous result

	fmt.Println(consumer.effects)
}

type idempotentConsumer struct {
	seen    map[string]struct{}
	effects int
}

func (consumer *idempotentConsumer) Consume(envelope outbox.Envelope) {
	if _, duplicate := consumer.seen[envelope.ID]; duplicate {
		return
	}
	consumer.seen[envelope.ID] = struct{}{}
	consumer.effects++
}

Output:
1

Index

• Variables
• type BacklogStats
• type Envelope
  • func (e Envelope) CanonicalJSON() []byte
  • func (e Envelope) ValidateForInsert(limits Limits) error
• type EnvelopeBuilder
  • func NewEnvelopeBuilder(options ...EnvelopeBuilderOption) (*EnvelopeBuilder, error)
  • func (b *EnvelopeBuilder) Build(params NewEnvelopeParams) (Envelope, error)
• type EnvelopeBuilderOption
  • func WithClock(clock func() time.Time) EnvelopeBuilderOption
  • func WithIDGenerator(generator func() (string, error)) EnvelopeBuilderOption
  • func WithLimits(limits Limits) EnvelopeBuilderOption
• type Event
• type Limits
  • func DefaultLimits() Limits
  • func (limits Limits) Validate() error
• type NewEnvelopeParams
• type Observer
• type ObserverFunc
  • func (observe ObserverFunc) Observe(ctx context.Context, event Event)
• type Operation
• type Outcome

Examples

• Package (DuplicateDeliveryRequiresIdempotentConsumer)

Variables

var (
	ErrIDRequired              = errors.New("outbox: ID is required")
	ErrIDTooLarge              = errors.New("outbox: ID is too large")
	ErrTopicRequired           = errors.New("outbox: topic is required")
	ErrTopicTooLarge           = errors.New("outbox: topic is too large")
	ErrPayloadTooLarge         = errors.New("outbox: payload is too large")
	ErrMetadataTooLarge        = errors.New("outbox: metadata is too large")
	ErrMetadataEntriesTooLarge = errors.New("outbox: metadata has too many entries")
	ErrOrderingKeyTooLarge     = errors.New("outbox: ordering key is too large")
	ErrIdempotencyKeyTooLarge  = errors.New("outbox: idempotency key is too large")
	ErrPayloadVersionRequired  = errors.New("outbox: payload version is required")
	ErrAttemptsInvalid         = errors.New("outbox: new envelope attempts must be zero")
	ErrAvailableAtRequired     = errors.New("outbox: availability time is required")
	ErrCreatedAtRequired       = errors.New("outbox: creation time is required")
	ErrTimestampOutOfRange     = errors.New("outbox: timestamp is outside JSON range")
	ErrInvalidLimits           = errors.New("outbox: limits must be positive")
)

Types

type BacklogStats

type BacklogStats struct {
	Pending         int64
	Leased          int64
	Dead            int64
	OldestPendingAt *time.Time
}

BacklogStats summarizes operationally relevant message states.

type Envelope

type Envelope struct {
	ID             string            `json:"id"`
	Topic          string            `json:"topic"`
	Payload        []byte            `json:"payload"`
	PayloadVersion uint16            `json:"payload_version"`
	Metadata       map[string]string `json:"metadata,omitempty"`
	OrderingKey    string            `json:"ordering_key,omitempty"`
	IdempotencyKey string            `json:"idempotency_key,omitempty"`
	Attempts       int               `json:"attempts"`
	AvailableAt    time.Time         `json:"available_at"`
	CreatedAt      time.Time         `json:"created_at"`
}

Envelope is the stable record transferred from PostgreSQL to a publisher. Attempts is zero when written and increases when a claim is published.

func (Envelope) CanonicalJSON

func (e Envelope) CanonicalJSON() []byte

CanonicalJSON returns a deterministic JSON representation. The wire struct has a fixed field order, metadata keys are sorted by encoding/json, and timestamps are formatted explicitly so encoding cannot fail on a time year.

func (Envelope) ValidateForInsert

func (e Envelope) ValidateForInsert(limits Limits) error

ValidateForInsert checks a new envelope against limits and persistence invariants. Writers call this again because callers can construct the exported Envelope type without using EnvelopeBuilder.

type EnvelopeBuilder

type EnvelopeBuilder struct {
	// contains filtered or unexported fields
}

EnvelopeBuilder validates and constructs new envelopes.

func NewEnvelopeBuilder

func NewEnvelopeBuilder(options ...EnvelopeBuilderOption) (*EnvelopeBuilder, error)

NewEnvelopeBuilder creates a bounded envelope builder.

func (*EnvelopeBuilder) Build

func (b *EnvelopeBuilder) Build(params NewEnvelopeParams) (Envelope, error)

Build creates a validated envelope and takes ownership by copying mutable payload and metadata values.

type EnvelopeBuilderOption

type EnvelopeBuilderOption func(*EnvelopeBuilder) error

EnvelopeBuilderOption configures an EnvelopeBuilder.

func WithClock

func WithClock(clock func() time.Time) EnvelopeBuilderOption

WithClock injects the clock used for creation and default availability.

func WithIDGenerator

func WithIDGenerator(generator func() (string, error)) EnvelopeBuilderOption

WithIDGenerator injects the envelope ID generator.

func WithLimits

func WithLimits(limits Limits) EnvelopeBuilderOption

WithLimits replaces all envelope size limits.

type Event

type Event struct {
	Operation Operation
	Outcome   Outcome
	Count     int
	MessageID string
	Topic     string
	Attempts  int
	Duration  time.Duration
}

Event contains payload-safe diagnostics for one outbox operation. It never includes payloads, metadata, or error text.

type Limits

type Limits struct {
	MaxIDBytes             int
	MaxTopicBytes          int
	MaxPayloadBytes        int
	MaxMetadataEntries     int
	MaxMetadataBytes       int
	MaxOrderingKeyBytes    int
	MaxIdempotencyKeyBytes int
}

Limits bounds all caller-controlled variable-length envelope fields.

func DefaultLimits

func DefaultLimits() Limits

DefaultLimits returns conservative defaults suitable for most relays.

func (Limits) Validate

func (limits Limits) Validate() error

Validate reports whether every envelope limit is positive.

type NewEnvelopeParams

type NewEnvelopeParams struct {
	Topic          string
	Payload        []byte
	PayloadVersion uint16
	Metadata       map[string]string
	OrderingKey    string
	IdempotencyKey string
	AvailableAt    time.Time
}

NewEnvelopeParams contains caller-controlled data for a new envelope.

type Observer

type Observer interface {
	Observe(context.Context, Event)
}

Observer receives structured lifecycle events.

type ObserverFunc

type ObserverFunc func(context.Context, Event)

ObserverFunc adapts a function to Observer.

func (ObserverFunc) Observe

func (observe ObserverFunc) Observe(ctx context.Context, event Event)

Observe forwards the event to the wrapped function.

type Operation

type Operation string

Operation identifies a bounded outbox lifecycle action.

const (
	OperationClaim       Operation = "claim"
	OperationPublish     Operation = "publish"
	OperationDeliver     Operation = "deliver"
	OperationRetry       Operation = "retry"
	OperationDeadLetter  Operation = "dead_letter"
	OperationRelease     Operation = "release"
	OperationExtendLease Operation = "extend_lease"
	OperationReplay      Operation = "replay"
	OperationPrune       Operation = "prune"
	OperationArchive     Operation = "archive"
)

type Outcome

type Outcome string

Outcome identifies whether an operation completed successfully.

const (
	OutcomeSuccess Outcome = "success"
	OutcomeFailure Outcome = "failure"
)