Documentation
¶
Overview ¶
Package outbox makes a side effect as durable as the write that caused it.
The problem: a handler writes a row and then publishes an event. Between those two lines the process can die, and you have a resource that exists with nothing downstream knowing. Swap the order and you have an event for a resource that does not exist. There is no ordering of two systems that fixes this — the write and the publish must be ONE atomic act.
The fix: write the message into a table in the SAME transaction as the state change. Either both land or neither does. A relay then reads the table and publishes, retrying until it succeeds.
What this buys is at-least-once delivery, which means a consumer MUST be idempotent — the relay can publish a message twice if it crashes between publishing and marking. Exactly-once does not exist here; at-least-once plus an idempotent consumer is the design.
Index ¶
Constants ¶
This section is empty.
Variables ¶
var ErrDrop = errors.New("outbox: drop this message")
ErrDrop tells the relay to stop retrying a message.
Functions ¶
Types ¶
type Message ¶
type Message struct {
ID uuid.UUID `gorm:"primaryKey;column:id"`
Topic string `gorm:"column:topic"`
Payload []byte `gorm:"column:payload;type:jsonb"`
// TraceID carries the originating request's id, so the eventual publish
// can be correlated back to the API call that caused it — hours later,
// in a different process.
TraceID string `gorm:"column:trace_id"`
Attempts int `gorm:"column:attempts"`
LastError string `gorm:"column:last_error"`
AvailableAt time.Time `gorm:"column:available_at"`
PublishedAt *time.Time `gorm:"column:published_at"`
CreatedAt time.Time `gorm:"column:created_at"`
}
Message is one pending side effect.
Migration:
CREATE TABLE outbox_messages (
id UUID PRIMARY KEY,
topic TEXT NOT NULL,
payload JSONB NOT NULL,
trace_id TEXT,
attempts INTEGER NOT NULL DEFAULT 0,
last_error TEXT,
available_at TIMESTAMPTZ NOT NULL DEFAULT now(),
published_at TIMESTAMPTZ,
created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
CREATE INDEX outbox_pending_idx ON outbox_messages (available_at)
WHERE published_at IS NULL;
type Publisher ¶
Publisher delivers a message to wherever it is going: a broker, a webhook, an email provider, an in-process bus.
An error means "try again later". Return ErrDrop for a message that will never succeed — a malformed payload from an old deployment — so the relay stops retrying it forever.
type PublisherFunc ¶
PublisherFunc adapts a function to Publisher.
type Relay ¶
type Relay struct {
// contains filtered or unexported fields
}
Relay polls the table and publishes.
It is safe to run on every replica: each pass claims rows with FOR UPDATE SKIP LOCKED, so two relays take disjoint batches rather than fighting or double-publishing. That is why this needs no distributed lock.
func (*Relay) OnShutdown ¶
OnShutdown implements lifecycle.Shutdowner.
type RelayOptions ¶
type RelayOptions struct {
DB *gorm.DB
Publisher Publisher
// Batch is how many messages one pass claims. Default 50.
Batch int
// Interval between passes. Default 5s.
Interval time.Duration
// MaxAttempts before a message is parked (left unpublished, with its
// error, for a human). Default 10. Parking rather than deleting is
// deliberate: a message nobody can explain is evidence.
MaxAttempts int
// Backoff is the base delay between attempts; it doubles each time.
// Default 10s.
Backoff time.Duration
}
RelayOptions configures the relay.
Source Files