Documentation
¶
Overview ¶
Package outbox writes events through a transactional outbox and publishes them with a relay, on any SQL engine that supplies the statements.
The package exports:
- Engine, every statement the outbox runs, compiled for one engine
- Outbox, the outbox on one engine's statements, with New, which checks the engine and returns one; Outbox.Sink, the event sink a recorder writes through; and Outbox.Relay, which returns its relay
- Relay, the reactor source of the committed, unpublished events, with Relay.Receive, which runs its passes, and Relay.Ready, its readiness
- RelayOption, which configures a Relay: Timeout, the bound on one row's handler; Drain, the last pass after cancellation; and Gated, the Gate the relay holds shared for each row
- QuarantineError, the relay's report of a row it quarantined, and PublishError, its report of a transient publish failure
Emitting ¶
An Outbox is built with New over an Engine: every statement the outbox runs, compiled for one engine by that engine's module, such as go-messaging's postgres sub-module. The engine's module also ships the tables as a migration set. This package holds no SQL and imports no driver.
Outbox.Sink is the event.Sink a composition root builds a service's event.Recorder over. It inserts each event a command raises in the command's own transaction, so an event exists exactly when the state it reports was committed.
Relaying ¶
Nothing is published by the sink: the Relay publishes the committed rows afterward, each inside a transaction of its own that locks the row and marks it published once the broker holds the event, so a stop between the commit and the publish loses no event. A row that can never be published, one that does not decode or whose publish fails permanently, is quarantined in its own transaction instead, so it does not hold the outbox back, and the relay reports it as a QuarantineError. The inbox package guards a consumer against handling a redelivered event twice.
Index ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Engine ¶
type Engine struct {
// Emit inserts one event into the outbox. Parameters: source, id,
// header (event.Encode's headers as JSON), and data. An event whose
// source and id are already in the outbox must fail.
Emit query.Statement
// ClaimRow locks the oldest row neither published nor quarantined that
// is not already held by another transaction, skipping held rows, for
// the rest of the transaction. It takes no parameters and returns seq,
// header, and data, in that order, or no row. A statement exposes no
// result columns, so [New] cannot check them; a mismatch fails the
// relay's first claim.
ClaimRow query.Statement
// MarkPublished marks the claimed row published. Parameter: seq.
MarkPublished query.Statement
// Quarantine marks the claimed row quarantined, recording why, so
// ClaimRow never returns it again. Parameters: seq and reason, a
// human-readable account of why the row cannot be published.
Quarantine query.Statement
}
Engine is the SQL a database engine's module supplies: every statement the outbox runs, compiled for that engine. The outbox holds no SQL of its own, so an engine is required, and it must define each statement with exactly the parameters named here. New checks both. Emit runs on the command's transaction, a sqlate *Tx, so it may declare a transaction required.
type Gate ¶ added in v0.2.0
type Gate interface {
}
Gate is what a relay holds while it handles each row: a quiesce gate, such as go-core's reactor.Gate, held shared by background work and exclusively by an operation that must run with that work paused. Shared waits until the gate admits a shared holder and returns its release, or fails, holding nothing; a wait ctx ends returns ctx's error.
type Outbox ¶
type Outbox struct {
// contains filtered or unexported fields
}
Outbox runs the outbox on one engine's statements: the sink a recorder writes events through, and the relay that publishes them.
func New ¶
New returns an outbox over eng, once eng defines every statement with the parameters Engine names.
func (*Outbox) Relay ¶
func (o *Outbox) Relay(db sqlate.Beginner, poll time.Duration, report func(ctx context.Context, err error), opts ...RelayOption) *Relay
Relay returns a relay over the outbox in db that waits poll between passes once a pass finds no row. poll and any Timeout must be positive, and any Drain not negative.
The relay passes each error it survives to report, with the context it was receiving under: each row it quarantines, as a *QuarantineError; each pass a transient publish failure ends, as a *PublishError; and each database error. A nil report discards them. report has messaging.Report's shape, so a messaging.Report can be given. It must not block for long, since the relay waits on it before its next row.
func (*Outbox) Sink ¶
Sink returns the event.Sink a composition root builds a service's event.Recorder over: it writes the events a command raises into the command's own transaction.
type PublishError ¶
type PublishError struct {
// Seq is the row's position in the outbox.
Seq int64
// Event is the row's event.
Event event.Event
// Err is the handler's error.
Err error
}
PublishError is the error the relay reports when a row's handler fails with an error not marked event.Permanent: the row stays unpublished, the pass ends, and the next pass retries the row.
func (*PublishError) Error ¶
func (e *PublishError) Error() string
func (*PublishError) Unwrap ¶
func (e *PublishError) Unwrap() error
type QuarantineError ¶
type QuarantineError struct {
// Seq is the row's position in the outbox.
Seq int64
// Event is the row's event, or the zero Event when the row does not
// decode.
Event event.Event
// Reason is why the row was quarantined, as the engine recorded it.
Reason string
// Err is the failure that quarantined the row: the decoding error, or
// the handler's permanent error.
Err error
}
QuarantineError is the error the relay reports for a row it quarantined: one whose header or event does not decode, or whose publish failed with an error marked event.Permanent. The row's quarantine is committed by the time it is reported.
func (*QuarantineError) Error ¶
func (e *QuarantineError) Error() string
func (*QuarantineError) Unwrap ¶
func (e *QuarantineError) Unwrap() error
type Relay ¶
type Relay struct {
// contains filtered or unexported fields
}
Relay is the source of the outbox's committed, unpublished events. A composition root runs it into a reactor whose handler publishes each event to the broker.
Each row is handled in a transaction of its own. The relay locks the oldest row neither published nor quarantined, skipping rows another relay holds, and calls the handler with its event inside that transaction, so the row stays locked while the event is published. When the handler returns nil, the row is marked published in the same transaction. When it returns an error marked event.Permanent, the row is quarantined in the same transaction instead, and the pass goes on to the next row. A row whose header or event does not decode is quarantined the same way, without calling the handler. When the handler returns any other error, the transaction rolls back, the row stays unpublished, and the pass ends. The next pass, a poll later, retries it, without bound. A stop anywhere before the commit leaves the row to be published again under the same source and id, so delivery is at least once and a broker that deduplicates on them sees it once.
One relay publishes in seq order, which is the order the rows were inserted among those committed, not the order their transactions committed: a long transaction can commit a row after a later one was published. It holds the rest of the outbox back behind a row whose publish fails transiently. Several relays share the rows, and each row is claimed by one relay at a time, but they publish in no particular order.
func (*Relay) Ready ¶
Ready reports whether the relay is receiving and no pass since its last successful one has failed to reach the database. It is true from the start of Receive, before the first pass. A handler's error, such as a broker that is down, leaves the relay ready: the row waits for the next pass, and readiness speaks only for the database.
func (*Relay) Receive ¶
Receive publishes through fn until ctx ends, and returns nil: no row and no failure ends it. A transient handler error or a database error ends the pass, and both are reported; a database error also makes the relay not ready until a pass succeeds. A row whose handler is running when ctx ends is settled before Receive returns, and a relay with a Drain then makes its last pass.
type RelayOption ¶
type RelayOption func(*Relay)
RelayOption configures a Relay.
func Drain ¶
func Drain(d time.Duration) RelayOption
Drain sets how long a relay keeps publishing once it is cancelled. The relay makes one last pass on a context of its own, bounded by d. The pass returns when the outbox is empty or a row fails, so the events committed while the producers drained are published before the process exits rather than at its next start. Once d runs out, the pass claims no further row. A row already claimed settles on its own deadlines, as every row does: its handler's Timeout and its transaction's twice that. The reactor's reactor.CancelAfter cancels the handler first if it runs out sooner, so set d below it. The default, 0, makes no last pass. d must not be negative.
func Gated ¶ added in v0.2.0
func Gated(g Gate) RelayOption
Gated sets the gate the relay holds shared for each row, from before the row's transaction begins until it commits or rolls back, so an exclusive holder of g never runs beside a relay's transaction. A relay waiting on g opens no transaction and claims no row. The wait honors the context the row is claimed under, Receive's or the last pass's, so a wait the stop or the drain ends claims nothing and reports nothing. Any other failure of g ends the pass and is reported, wrapped, but leaves the relay ready, since it is not the database's. A nil g, the default, leaves the relay ungated.
func Timeout ¶
func Timeout(d time.Duration) RelayOption
Timeout sets the deadline on each row's handler context. The row stays locked while its handler runs, so the timeout bounds how long a hung publish holds it. The row's whole transaction is bounded at twice the timeout, so a stalled database, or a handler that ignores its deadline, rolls the transaction back rather than hold the row and the drain. It must be positive.
Source Files