Documentation
¶
Overview ¶
Package outbox implements a transactional outbox for reliable event delivery to declared durable consumers.
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 delivers each committed row to every declared durable consumer with at-least-once semantics.
Two delivery lanes ¶
Delivery is split across two disjoint lanes, so neither duplicates the other:
- Real-time lane (best-effort, ephemeral): the live event bus is notified post-commit by the caller (e.g. crud.EmitEvent), feeding SSE streams and ephemeral On/Subscribe handlers. Lossy by design.
- Durable lane (tracked per consumer): the Relay delivers each row to the consumers declared via Outbox.Consume. The Relay does NOT touch the live bus.
Per-consumer delivery & sibling isolation ¶
Each (parent row, declared consumer) pair has its own delivery row in event_outbox_delivery. A consumer that errors or panics is retried with exponential backoff and eventually dead-lettered INDEPENDENTLY of its siblings — one broken consumer never blocks another or fails the whole row. A consumer that is removed (no handler on any replica) has its deliveries abandoned once they age past the handler grace, so it can't block completion. A parent row is marked dispatched once it has no pending deliveries left (all dispatched/dead/abandoned); it may complete with some deliveries dead. Outbox.Replay / Outbox.ReplayConsumer resurrect dead or abandoned deliveries.
At-least-once delivery ¶
The Relay claims a batch of pending deliveries, invokes the consumer's handler, and marks the delivery dispatched only after the handler returns nil. A crash between the two re-delivers (consumers dedup by event.Event.ID, which the Relay stamps from the outbox row id).
Multi-replica safety ¶
The claim takes a lease (the claimed_until column) at the delivery grain. A Relay that dies mid-batch holds only its claimed deliveries until the lease expires; after expiry another Relay — or the same process after a restart — reclaims and re-delivers them.
Layering ¶
outbox is an L3 leaf package with two deliberate intra-L3 edges: outbox → event (uses event.EventHandler / event.Event) and outbox → db (uses db.Executor so Append participates in the caller's transaction). The precedent is slowquery → db.
Index ¶
- type Delivery
- type Option
- type Outbox
- func (o *Outbox) Append(ctx context.Context, ex db.Executor, eventType string, data any) (string, error)
- func (o *Outbox) Consume(name, eventType string, handler event.EventHandler)
- func (o *Outbox) List(ctx context.Context, status string, limit int) ([]Row, error)
- func (o *Outbox) ListDeliveries(ctx context.Context, rowID string) ([]Delivery, error)
- func (o *Outbox) Nudge()
- func (o *Outbox) Replay(ctx context.Context, rowID string) error
- func (o *Outbox) ReplayConsumer(ctx context.Context, rowID, consumer string) error
- func (o *Outbox) StartRelay(ctx context.Context) (stop func())
- type Row
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Delivery ¶ added in v0.16.0
type Delivery struct {
RowID string
Consumer string
Status string
Attempts int
LastError string
NextAttemptAt *time.Time
DispatchedAt *time.Time
}
Delivery is a snapshot of one per-consumer delivery row. Status is one of "pending", "dispatched", "dead" (handler exhausted MaxAttempts), or "abandoned" (no consumer handler existed anywhere within the grace window — a removed consumer). Each (parent row, declared consumer) pair has exactly one delivery row; its lifecycle is independent of every sibling consumer's delivery for the same event (sibling isolation) — one consumer failing never blocks another.
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 WithHandlerGrace ¶ added in v0.16.0
WithHandlerGrace sets the grace window that governs two symmetric, time-based decisions: how long a delivery may stay unhandled (no consumer handler on any replica) before it is abandoned, and how long a parent stays pending before it may be completed/dropped. Both wait the grace so a consumer being rolled out on another replica has time to expand and deliver its rows before anything is settled.
It therefore also bounds parent-completion latency: a fully-delivered parent is not marked dispatched until it is older than the grace (delivery to consumers is unaffected and prompt — only the parent bookkeeping/GC lags).
The grace MUST comfortably exceed your rolling-deploy overlap window PLUS worst-case clock skew between replicas (delivery/parent timestamps are written by one replica and compared on another). Keep a floor of a few minutes even for fast deploys. Defaults to 15m.
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 WithRetention ¶ added in v0.16.0
WithRetention enables automatic purge of fully-settled (dispatched) parent rows and their deliveries once they are older than d. The relay runs the purge as part of its poll cycle. Zero (the default) disables purging — rows are kept forever. Only dispatched parents are ever purged; pending or dead/abandoned deliveries are never deleted out from under a consumer.
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 delivers each to a set of declared durable consumers, independently, with at-least-once semantics. Construct with New, register consumers with Outbox.Consume, then launch the relay with Outbox.StartRelay.
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) Consume ¶ added in v0.16.0
func (o *Outbox) Consume(name, eventType string, handler event.EventHandler)
Consume registers a durable consumer. Must be called before StartRelay. name is a stable identity used to track per-consumer delivery across restarts/replicas; (eventType, name) must be unique — registering the same pair twice panics. handler is invoked with Event.ID set to the outbox row id. Note Event.ID is shared across ALL consumers of that row, so a handler that deduplicates must key on (name, Event.ID), never Event.ID alone, or one consumer's success would suppress a sibling's delivery. A consumer declared after events have already been staged sees only events staged from its registration forward (the relay expands deliveries for pending rows each pump, so it catches up on still-pending rows immediately and misses only rows already dispatched before it was declared).
The handler runs on the relay goroutine: it MUST be side-effecting but prompt, and idempotent (at-least-once delivery means a duplicate is always possible after a crash or lease expiry). A handler that returns an error or panics is retried with exponential backoff and eventually dead-lettered — independently of its sibling consumers for the same event (sibling isolation).
Re-adding a previously-removed consumer resumes delivery for events staged from the re-add forward automatically. Events whose deliveries were abandoned during the removal gap are NOT re-delivered automatically (a terminal abandoned row blocks re-expansion); recover those explicitly with Outbox.ReplayConsumer.
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) ListDeliveries ¶ added in v0.16.0
ListDeliveries returns the per-consumer delivery rows for one parent row, ordered by consumer name. Returns an empty slice (not nil) for a parent with no deliveries (e.g. no declared consumer matched its type).
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 resurrects ALL dead or abandoned deliveries of a parent row and reopens the parent so the relay re-completes it. Resurrected deliveries have attempts cleared and are scheduled immediately; the parent flips back to pending with dispatched_at cleared. Idempotent: only rows that have at least one dead/abandoned delivery are affected, so replaying a fully-dispatched or unknown row is a no-op. Sibling consumers already dispatched are untouched.
func (*Outbox) ReplayConsumer ¶ added in v0.16.0
ReplayConsumer resurrects a single consumer's dead or abandoned delivery for a row and reopens the parent. Like [Replay] but scoped to one consumer — use it to retry just the dead-lettered consumer after fixing its handler, or to re-deliver to a consumer that was removed and re-added (its delivery abandoned in the interim). Idempotent (no-op if the named delivery isn't dead/abandoned).
func (*Outbox) StartRelay ¶
StartRelay launches the Relay goroutine. The relay delivers each pending outbox row to every declared durable consumer INDEPENDENTLY: it expands per-consumer delivery rows, claims a batch, invokes each consumer's handler directly (the relay no longer touches the live event bus), and settles the delivery — dispatched on success, retried with backoff on error or panic, dead after MaxAttempts. A parent row flips to dispatched once it has no pending deliveries left.
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. Register all consumers via Outbox.Consume before calling this; the declared set is read once per pump and must not change after the relay starts.
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 parent outbox row. The parent status is now just "pending" → "dispatched": a row is "pending" until the relay has settled every per-consumer delivery (dispatched/dead/abandoned), then it flips to "dispatched". The parent's attempts/last_error columns are vestigial — per-attempt state lives in event_outbox_delivery (see Delivery).
Source Files