Documentation
¶
Overview ¶
Package outbox is the durable half of the change feed: a table written in the same transaction as the change, and a dispatcher that tails it.
It is the rest.Source ADR-0012 describes and ADR-0045 left a seam for. Swapping rest.Broker for a Dispatcher is a constructor call — the endpoint, the wire format, the `Last-Event-ID` contract and every generated client are unchanged — and what it buys is the two properties the Broker documents itself as lacking:
- **At-least-once instead of at-most-once.** The event and the data commit together. A process that dies between the commit and the fan-out has already written the event, and the dispatcher delivers it when it comes back.
- **Correct across replicas instead of correct on one.** Every replica writes to one table and every dispatcher reads it, so a write served by one replica reaches subscribers connected to another.
The shape ¶
outbox.Install(ctx, pool, outbox.Options{}) // the table and its trigger
ob := outbox.New(pool, outbox.Options{})
rest.Must(rest.PublishChanges[blog.Post](reg, ob)) // unchanged from the Broker
d := outbox.NewDispatcher(ctx, pool, outbox.DispatcherOptions{})
go d.Run(ctx)
rest.Must(rest.Events(api, rest.EventsOptions{Source: d}))
Outbox implements rest.TxPublisher, so `rest.PublishChanges` records into the writing transaction rather than after it. That assertion is the whole of the swap: nothing else in an application changes.
Why a subscriber can be replayed after a restart ¶
The stream position is the outbox row's `id`, so it means the same thing in every process and across every restart. A client reconnecting with `Last-Event-ID: 4210` is answered from the table rather than from a ring buffer that died with the last deployment — which is the difference between a rolling restart costing every connected client a full refetch and costing them nothing.
A position older than the oldest row still retained gets a rest.Reset, the same as it would from a Broker whose history had rolled over. Retention is therefore a delivery guarantee and not only a disk-space setting; see Options.Retention.
Ordering, and what it costs ¶
A tail of `id > cursor ORDER BY id` is only correct if rows become visible in id order, and a bare sequence does not promise that: two transactions can take ids 5 and 6 and commit in the other order, and a dispatcher that read 6 first would advance its cursor past 5 and lose it silently. Losing an invalidation silently is the one failure this whole design exists to avoid.
So Outbox.Record takes a transaction-scoped advisory lock before it inserts. The lock is held until commit, so id order is commit order by construction and the dispatcher's tail needs no reasoning about visibility at all.
**The cost is real and should be read before adopting this.** Writes to published models serialise against each other from the outbox insert to the commit — roughly the commit itself, since the outbox write is the last thing the mutation does. That bounds write throughput on published models at about one transaction per commit latency. For the applications this feature is for it is not the binding constraint; for a write-heavy ingest path it may be, and the answer there is not to publish that model. ADR-0012 carries the revisit trigger.
It is `pg_advisory_xact_lock` rather than the session form deliberately: ADR-0019 forbids anything session-scoped on the query path, because the target deployment runs PgBouncer in transaction pooling. A transaction-scoped lock is released by the commit that returns the connection, so it is safe through a pooler.
What still needs a direct connection ¶
The dispatcher's `LISTEN`, and only that. ADR-0019 measured it: PgBouncer in transaction pooling *accepts* a `LISTEN` and then silently never delivers, so Dispatcher.Run must be given a pool that reaches Postgres directly. `NOTIFY` is transactional and survives the pooler, so the write path — which is everything else — needs no exception.
A dispatcher whose `LISTEN` is silently useless still works, at the poll interval, which is exactly the failure ADR-0012 warns can hide. Dispatcher therefore reports it rather than absorbing it: see [DispatcherOptions.OnError] and Dispatcher.Stats.
Index ¶
Constants ¶
const DefaultChannel = "sqlb_outbox"
DefaultChannel is the `NOTIFY` channel the table's trigger rings when Options.Channel is empty.
It is a doorbell and carries no payload, so the 8000-byte limit on a notification is not a constraint on anything. A dispatcher that never hears it is late rather than wrong, which is what makes the poll a fallback rather than a second mechanism to keep correct.
const DefaultTable = "sqlb_outbox"
DefaultTable is where events are recorded when Options.Table is empty.
Variables ¶
var ErrDispatcherClosed = errors.New("outbox: the dispatcher is closed")
ErrDispatcherClosed reports a subscription to a Dispatcher that has been closed.
Functions ¶
func DDL ¶
DDL renders the table, its index and the trigger that rings the doorbell.
It is `IF NOT EXISTS` throughout and safe to apply repeatedly, which is what lets Install be called at startup. For a project that owns its migrations — which is every project that has got as far as needing this — the right home for this text is a migration file, so that the outbox appears in the same history as everything else it is transactional with.
The trigger is `FOR EACH STATEMENT`, not per row: the notification carries no payload, so one per statement says exactly as much as one per row and a bulk update publishing four hundred events rings once.
The doorbell is a trigger on the outbox table rather than a `pg_notify` in Go, which is ADR-0012's decision and worth restating where someone might undo it. A notify issued from the mutation path is one fewer database object and is forgettable: a new write path that appends to the table and omits the notify passes every test and lags in production, because the fallback poll covers for it.
Types ¶
type Dispatcher ¶
type Dispatcher struct {
// contains filtered or unexported fields
}
Dispatcher tails the outbox table and fans it out to subscribers. It is the rest.Source that replaces rest.Broker.
It takes a *pgxpool.Pool rather than an sqlb.Executor because it needs one thing an Executor cannot express: a connection of its own to hold a `LISTEN` on. That connection is hijacked out of the pool rather than borrowed, so a session carrying a `LISTEN` is never returned for someone else's query.
**The pool must reach Postgres directly.** ADR-0019 measured PgBouncer in transaction pooling accepting a `LISTEN` and then silently never delivering on it, which leaves this working — at the poll interval — while looking broken to nobody. Dispatcher.Run probes for exactly that at startup and reports it through Options.OnError.
func MustDispatcher ¶
func MustDispatcher(d *Dispatcher, err error) *Dispatcher
MustDispatcher is NewDispatcher for a startup path with nowhere to put an error.
func NewDispatcher ¶
func NewDispatcher(ctx context.Context, pool *pgxpool.Pool, opts DispatcherOptions) (*Dispatcher, error)
NewDispatcher returns a Dispatcher over pool, with its starting position already fixed.
The position is read here rather than in [Run], and that is a correctness requirement rather than tidiness. Run is started in a goroutine, so a subscriber can attach before it has executed a line — and if Run then set the cursor to the head of the table, every event written in between would be behind the cursor and delivered to nobody. Fixing the position before the caller has a Dispatcher at all removes the window instead of narrowing it.
It also means a missing outbox table fails here, where a caller is looking at an error, rather than arriving later through Options.OnError from a goroutine nobody is watching.
func (*Dispatcher) Close ¶
func (d *Dispatcher) Close()
Close disconnects every subscriber and refuses further subscriptions.
func (*Dispatcher) Run ¶
func (d *Dispatcher) Run(ctx context.Context) error
Run tails the table until ctx is cancelled. It blocks, and is meant to be the body of a goroutine started once at startup.
It returns ctx.Err() and nothing else: the one failure it could not proceed through — establishing the starting position — happened in NewDispatcher, and everything from here is transient by nature. A poll that failed or a LISTEN that dropped goes to Options.OnError while the loop keeps going, because a dispatcher that gave up on a failed query would be a feed that stops on the first blip and says so only to whoever reads a returned error.
func (*Dispatcher) Stats ¶
func (d *Dispatcher) Stats() Stats
Stats reports what the dispatcher is doing. It exists for a metric and a health check; nothing on the request path reads it.
func (*Dispatcher) Subscribe ¶
Subscribe implements rest.Source.
type DispatcherOptions ¶
type DispatcherOptions struct {
// Options describes the table being tailed, and must match the Options the
// [Outbox] writing it was built with.
Options
// Poll is how often the table is read in the absence of a notification.
// Defaults to 5 seconds; it is a fallback, not the delivery mechanism.
//
// It exists because a lost notification must degrade to latency rather than
// to lost data — a connection pooler in transaction mode swallows LISTEN
// entirely ([ADR-0019]), and this is what keeps the feed correct there. See
// [DispatcherOptions.OnError] for why that is reported rather than silently
// absorbed.
//
// [ADR-0019]: https://github.com/jryannel/sqlb/blob/main/docs/adr/0019-pgbouncer-in-the-path.md
Poll time.Duration
// Buffer is how many events may queue for one subscriber before the
// dispatcher gives up on it and closes its channel. Defaults to 256.
//
// Dropping the subscriber rather than the event is [ADR-0045]'s policy and
// this implementation keeps it: a dropped event is a client that stays wrong
// forever, and a dropped connection is a client that reconnects, is replayed
// from the table, and converges.
//
// [ADR-0045]: https://github.com/jryannel/sqlb/blob/main/docs/adr/0045-the-stream-is-a-seam.md
Buffer int
// Batch is how many rows one tail query reads. Defaults to 512.
Batch int
// MaxReplay is the largest catch-up a reconnecting subscriber is given
// before it is told to refetch instead. Defaults to 1000.
//
// Past some gap a replay stops being cheaper than the thing it saves. A
// client resuming across forty thousand invalidations would receive forty
// thousand messages and then refetch most of its views anyway, so beyond
// this it gets one reset and does that directly.
MaxReplay int
// Replay, when false, refuses to catch a reconnecting subscriber up from
// the table and answers every resumption with a reset.
//
// The default — replay enabled — is the reason to run an outbox at all: a
// position means the same thing in every process, so a rolling restart costs
// connected clients nothing instead of costing each of them a full refetch.
// Turning it off trades that for never running the catch-up query.
DisableReplay bool
// StartAtBeginning dispatches the whole retained table on the first run
// rather than starting at its head.
//
// The default starts at the head, because the rows already in the table were
// delivered by whoever was running before and re-sending them invalidates
// every client's world for no reason. Set this when the dispatcher is the
// first one to run against a table that was already being written.
StartAtBeginning bool
}
DispatcherOptions configures a Dispatcher.
type Options ¶
type Options struct {
// Table is the outbox table, unqualified and lowercase. Defaults to
// [DefaultTable].
Table string
// Channel is the NOTIFY channel the table's trigger rings. Defaults to
// [DefaultChannel].
Channel string
// Retention is how long a delivered event is kept so that a reconnecting
// client can still be replayed from it. Defaults to 24 hours; negative
// disables pruning entirely.
//
// This is a delivery setting before it is a disk setting. A client whose
// Last-Event-ID is older than the oldest retained row is told to refetch
// everything it displays, so the retention window is the longest
// disconnection a client can survive cheaply. A mobile app backgrounded
// overnight wants a day; a dashboard on a wall wants whatever the deploy
// interval is.
Retention time.Duration
// OnError reports a failure that could not be returned to a caller: a
// best-effort record under autocommit, a poll that failed, a LISTEN that
// dropped. Nil discards them, which is the wrong default for anything you
// intend to rely on and is why every constructor's documentation says so.
//
// It may be called from several goroutines.
OnError func(error)
}
Options configures an Outbox and the DDL that backs it.
type Outbox ¶
type Outbox struct {
// contains filtered or unexported fields
}
Outbox records changes into the table, inside the transaction that made them.
It implements rest.TxPublisher, which is what makes it a drop-in for rest.Broker in a rest.PublishChanges call: the assertion in that function finds Record and uses it, so the events land in the writing transaction rather than in a callback after it.
The zero value is not usable; call New.
func New ¶
New returns an Outbox recording into the table Options names.
exec is used only for the fallback path — a write that ran outside a transaction, where there is no transaction to record into and the change is already durable. Every ordinary write records on the transaction it is part of, taken from the context, and never touches this handle.
Set Options.OnError. The fallback path cannot return a failure to anyone — the write it belongs to has already committed — so without it a change that was never recorded is indistinguishable from one that was.
func (*Outbox) Prune ¶
Prune removes events older than Options.Retention and reports how many went.
Dispatcher.Run calls this on a timer, so an application that runs a dispatcher does not need to. It is exported for the one that does not — a worker fleet that publishes but serves no stream still fills the table.
Pruning is what makes the retention window a guarantee rather than an intention, and it is also the one operation here that can make a connected client refetch: an event deleted while a client was disconnected past it turns that client's reconnection into a rest.Reset.
func (*Outbox) Publish ¶
Publish is the fallback for a write that ran outside a transaction, which is what rest.Options.DisableTransactions produces.
There is nothing to be atomic with: the statement committed before the hook ran, so the event is recorded in a transaction of its own. That is at-most-once for this one event — the process can die in between — and it is the guarantee rest.Broker gives for every event, so nothing is worse off than it would be without an outbox. It is simply not what the outbox is for.
The failure has nowhere to go. The write it belongs to is durable and cannot be undone by returning an error, and this signature has no error to return anyway, so it goes to Options.OnError — which is why that field's documentation says to set it.
func (*Outbox) Record ¶
Record writes the events into the transaction carried by ctx.
This is rest.TxPublisher, and the error it returns is load-bearing: it travels back through the hook that called it and rolls the mutation back. A change that could not be recorded is a change no subscriber would ever hear about, and a row that exists while every client believes it does not is worse than a write that failed and said so.
It takes a transaction-scoped advisory lock first. See the package documentation for why the ordering that buys is worth what it costs.
type Stats ¶
type Stats struct {
// Cursor is the highest outbox id dispatched.
Cursor uint64
// Subscribers is how many streams are connected.
Subscribers int
// Listening reports whether the LISTEN connection is currently established.
// It says nothing about whether notifications actually arrive on it, which
// is the failure a pooled connection produces; Notifications is what answers
// that.
Listening bool
// Delivered counts events fanned out since the process started.
Delivered uint64
// Notifications counts doorbells heard on the LISTEN connection.
//
// It is the metric to alert on. Listening true with this flat is a feed
// running entirely on its fallback poll — correct, slow, and identical from
// the outside to one that is working.
Notifications uint64
}
Stats is what a metric or a health check reads off a Dispatcher.
Source Files