Documentation
¶
Overview ¶
Package outbox implements a transactional outbox for reliable event delivery.
The in-process event.EventBus emits after the database transaction commits. If the process crashes in the gap between commit and emit, the event is lost. The outbox closes that gap: Append writes the event row inside the caller's transaction (it commits or rolls back with the business write), and a background Relay publishes committed rows to the bus with at-least-once semantics.
At-least-once delivery ¶
The Relay claims a batch of pending rows, calls event.EventBus.Emit synchronously, and marks the row dispatched only after Emit succeeds. A crash between Emit and the mark leaves the row claimable; on restart it is delivered again. Consumers MUST be idempotent and deduplicate by event.Event.ID, which the Relay stamps from the outbox row's primary key. Events emitted directly via Emit/EmitAsync carry an empty ID and have no durable identity.
Multi-replica safety ¶
The claim takes a lease (the claimed_until column). A Relay that dies mid-batch holds its rows until the lease expires; after expiry another Relay — or the same process after a restart — reclaims and re-delivers them. This lets several replicas run a Relay against one shared table without double-processing (modulo the at-least-once caveat above).
Layering ¶
outbox is an L3 leaf package with two deliberate intra-L3 edges: outbox → event (publishes Events) and outbox → db (uses db.Executor so Append participates in the caller's transaction). The precedent is slowquery → db.
Index ¶
- type Emitter
- type Option
- type Outbox
- func (o *Outbox) Append(ctx context.Context, ex db.Executor, eventType string, data any) (string, error)
- func (o *Outbox) List(ctx context.Context, status string, limit int) ([]Row, error)
- func (o *Outbox) Nudge()
- func (o *Outbox) Replay(ctx context.Context, id string) error
- func (o *Outbox) StartRelay(ctx context.Context, bus Emitter) (stop func())
- type Row
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Option ¶
type Option func(*Outbox)
Option configures an Outbox.
func WithBatchSize ¶
WithBatchSize sets the maximum number of rows the Relay claims per pump. Defaults to 100.
func WithMaxAttempts ¶
WithMaxAttempts sets how many Emit attempts a row gets before it is marked "dead". Defaults to 10.
func WithPollInterval ¶
WithPollInterval sets how often the Relay polls for pending rows when no Nudge arrives. Defaults to 1s.
func WithTable ¶
WithTable overrides the default "event_outbox" table name. The name is validated as a safe SQL identifier at construction; an invalid name makes New return an error.
func WithoutEnsureTable ¶
func WithoutEnsureTable() Option
WithoutEnsureTable suppresses the CREATE TABLE IF NOT EXISTS that New otherwise runs at construction. Use it in deployments whose policy forbids unattended DDL (typically alongside framework.WithoutAutoMigrate): you must then create the outbox table via your own migration pipeline before the app stages any event, or the first Append fails. The table schema is documented in framework/docs/content/events.md.
type Outbox ¶
type Outbox struct {
// contains filtered or unexported fields
}
Outbox stores event rows transactionally and relays them to an event bus. Construct with New.
func New ¶
New constructs an Outbox backed by db. It detects the dialect (postgres or sqlite — mirroring battery/queue) and ensures the table and its (status, created_at) index exist.
func (*Outbox) Append ¶
func (o *Outbox) Append(ctx context.Context, ex db.Executor, eventType string, data any) (string, error)
Append inserts a pending row using ex — callers hand in their *sql.Tx (it satisfies db.Executor) so the row commits or rolls back with the business write. data is JSON-marshalled into Payload. Returns the new row's ID, which becomes event.Event.ID on delivery for consumer dedup.
func (*Outbox) List ¶
List returns up to limit rows, newest-first. An empty status returns rows regardless of state; otherwise only rows in that status. limit <= 0 defaults to 100.
func (*Outbox) Nudge ¶
func (o *Outbox) Nudge()
Nudge wakes the Relay immediately (non-blocking send on a cap-1 channel). Callers invoke it right after commit so delivery latency is not bound to PollInterval. Extra nudges coalesce — only one wake is buffered regardless of how many arrive between pumps.
func (*Outbox) Replay ¶
Replay resets a dead row to pending so the Relay picks it up again — attempts cleared, scheduled immediately. The `AND status='dead'` clause makes it idempotent: replaying a pending, dispatched, or unknown row matches nothing and is a no-op (same contract as battery/queue's Replayable).
func (*Outbox) StartRelay ¶
StartRelay launches the Relay goroutine. It claims batches of pending rows, publishes each to bus via the SYNCHRONOUS Emit (first handler error aborts and counts as a failed attempt), then marks the row dispatched. An Emit failure increments Attempts, records LastError, schedules an exponential backoff via next_attempt_at, and — once Attempts reaches MaxAttempts — marks the row dead.
The loop runs until ctx is cancelled. The returned stop func blocks until the loop has fully exited, so callers can drain safely on shutdown.
type Row ¶
type Row struct {
ID string
Type string
Payload []byte // JSON of event.Event.Data
Status string
Attempts int
LastError string
CreatedAt time.Time
DispatchedAt *time.Time
}
Row is a snapshot of one outbox row. Status is one of "pending", "dispatched", or "dead". A row is "pending" while awaiting delivery (including the in-flight window, which is guarded by a lease rather than a separate status), "dispatched" after a successful Emit, and "dead" after exhausting MaxAttempts.
Source Files