Documentation
¶
Overview ¶
Package outbox implements the transactional outbox: the pattern that makes a state change and the events announcing it atomic without a distributed transaction.
A unit of work writes aggregate state and outbox rows in ONE database transaction; a separate relay — a lifecycle participant, leader-elected — drains those rows to the broker afterwards. If the transaction rolls back the rows roll back with it, and if the process dies between commit and publish the rows are still there for the next drain. The honest guarantee is at-least-once publication, which is why the inbox dedupes on Message.ID.
Index ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func Sink ¶
Sink returns the commit hook that turns an aggregate's drained events into outbox records: encode each one, append them all in the caller's transaction. It is the bridge between persistence and outbox, and every application was writing it by hand.
uow.OnCommit(outbox.Sink(store, outbox.JSONEncoder()))
Appending inside the transaction is the whole pattern: the aggregate's new state and the rows announcing it commit together or not at all.
Types ¶
type Durable ¶ added in v0.2.0
type Durable interface {
// Durable reports whether records outlive this process.
Durable() bool
}
Durable is implemented by a Store that survives the process — one backed by a database rather than by memory.
It exists so the relay can tell apart the two configurations that use the Standalone elector: a modular monolith, where always-leading is right, and several replicas over one table, where it silently duplicates every event. A Store that does not implement this is assumed NOT durable, which is true of the in-process store and makes the warning impossible to trigger spuriously.
type Elector ¶
Elector grants the exclusive right to drain the outbox. Lead acquires leadership, runs fn with a context cancelled when leadership is lost, and returns when fn returns.
func Standalone ¶
func Standalone() Elector
Standalone is the default Elector: it always leads. Correct for a single instance and for the modular monolith.
With several replicas and a DURABLE store it is wrong, and quietly: every replica drains the same table, each marks records published, and each delivers to its own broker. A field test lost 50% of its events to exactly this, with no error anywhere — the rows said published, and they were.
So the relay warns at startup when this elector is paired with a store that reports itself durable, naming the fix. It cannot know how many replicas there are; it can know that the combination is only safe at one.
type Electors ¶ added in v0.2.0
type Electors interface {
// Elector returns the leadership called name, claiming it for this
// process.
//
// CALL IT FROM A CONSTRUCTOR. A claim refused is then a boot failure
// naming both claimants, and a claim not refused is the guarantee that
// nothing else in this binary is quietly competing for the same lock.
// Called lazily on a request path it would still work and would prove
// nothing.
Elector(name string) (Elector, error)
}
Electors mints leaderships by name.
A NAME is a leadership. Components with different names lead at the same time on the same replica; components sharing a name contend for one — and that is the only thing a name is for.
It exists because one Elector is one lock. README used to say a scheduler that wants leader-only "injects the same one", and a field test did exactly that: an SLA sweeper and the outbox relay on one elector, so whichever goroutine woke first took the lock and the other did nothing for the life of the process. Four runs of one binary, two each way, both reporting /readyz 200.
el, err := electors.Elector("ticket/sla-sweeper")
Warren's container resolves by TYPE, so this is a second type rather than a second binding of the first: there is no keyed resolution to reach for.
func StandaloneElectors ¶ added in v0.2.0
func StandaloneElectors() Electors
StandaloneElectors returns an Electors whose every elector always leads. Correct for one instance and for the modular monolith; with several replicas, every replica runs the work.
It is also what a test substitutes for the real thing:
warrentest.Replace[outbox.Electors](outbox.StandaloneElectors())
It claims names exactly as strictly as a real registry does. A double that permits what production refuses is a test that passes for a service which cannot start.
type EncodeOption ¶
type EncodeOption func(*jsonEncoder)
EncodeOption configures JSONEncoder.
type Encoder ¶
Encoder turns a domain event into the record the outbox stores.
func JSONEncoder ¶
func JSONEncoder(opts ...EncodeOption) Encoder
JSONEncoder encodes the concrete event value with encoding/json. The message's Key is the event's AggregateID — the reason domain.Event flattens identity to a string, and what preserves per-aggregate order through a partitioned broker.
type MemoryOption ¶
type MemoryOption func(*memoryStore)
MemoryOption configures NewMemoryStore.
func WithClock ¶
func WithClock(now func() time.Time) MemoryOption
WithClock injects the time source — how tests drive the store without sleeping.
type Record ¶
Record is one outbox row: a broker.Message plus the topic it publishes to. broker.Message deliberately carries no topic — a subscription's topic is boot-time state — but an outbox row is the one place the topic must travel with the message.
type Relay ¶
type Relay struct {
// contains filtered or unexported fields
}
Relay drains the outbox to the broker. It is constructed by the module and driven by the lifecycle; DrainOnce is exported so a test can drive one pass deterministically, with no goroutine and no clock.
func NewRelay ¶
func NewRelay(store Store, pub broker.Publisher, opts ...RelayOption) *Relay
NewRelay returns a relay over store and pub.
func (*Relay) DrainOnce ¶
DrainOnce publishes one batch and returns how many records were dispatched. It publishes in insertion order, batching consecutive records that share a topic into one call, and stops at the first failure without publishing anything behind it: global order is a stronger guarantee than per-aggregate order and it is what makes the promise one sentence long.
Head-of-line blocking is the accepted cost — a publish failure is nearly always broker-wide, so in the common case nothing is behind it — and the disposition rules bound the pathological case: a transient failure leaves the record for the next poll, a rejection parks it immediately, and an unknown failure retries until the policy stops and then parks.
func (*Relay) Flush ¶
Flush drains what is pending one last time, bounded by FlushTimeout. It is the relay's OnStop: shutdown step 4, after consumers stop and before connections close. Rows written after it are simply published by the next process — which is precisely what an outbox is for.
func (*Relay) Run ¶
Run drains the outbox until ctx is cancelled: it acquires leadership, then loops — drain everything pending, then wait for the store's signal (or the poll interval) and drain again. It returns when leadership ends or ctx is cancelled, which is what makes it a lifecycle participant rather than a goroutine someone forgot about.
Register it from a constructor that injects lifecycle.Lifecycle:
ctx, cancel := context.WithCancel(context.Background())
lc.Append(lifecycle.Hook{
Name: "outbox relay",
OnStart: func(context.Context) error { go relay.Run(ctx); return nil },
OnStop: func(c context.Context) error { cancel(); return relay.Flush(c) },
})
The loop's context must NOT be OnStart's: that one is the boot context, which outlives boot only under Run and is cancelled immediately under Start/Stop — so a relay started on it would leak the goroutine in every test.
A drain error does not end the loop — a broker outage is transient by definition, and the disposition rules already decide what happens to the records.
type RelayOption ¶
type RelayOption func(*Relay)
RelayOption configures a Relay.
func Backoff ¶
func Backoff(p app.RetryPolicy) RelayOption
Backoff sets the retry policy for records whose publish failed transiently; when it stops, the record is parked. Default ExponentialBackoff(10).
func BatchSize ¶
func BatchSize(n int) RelayOption
BatchSize caps how many records one drain publishes. Default 100.
func FlushTimeout ¶
func FlushTimeout(d time.Duration) RelayOption
FlushTimeout bounds the final drain at shutdown, inside the lifecycle's force-exit budget. Default 10s.
func LeaderElection ¶
func LeaderElection(e Elector) RelayOption
LeaderElection sets who may drain. The default is Standalone(), which always leads — correct for one instance and for the modular monolith, and wrong for several replicas over a durable store, where every replica would drain.
func PollInterval ¶
func PollInterval(d time.Duration) RelayOption
PollInterval is how long Run waits between drains when the store offers no Waiter, and the safety net when it does — a missed signal delays publication, never loses it. Default 1s.
func ReportTo ¶ added in v0.2.0
func ReportTo(l *slog.Logger) RelayOption
ReportTo sets where the relay reports a drain it could not complete — most importantly a PARKED record, whose diagnostic names the record, the topic, the key, the attempt count, and the fact that later records for that key are now out of order.
It exists because Run used to discard that error. Its inner loop was `if err != nil || n == 0 { break }`, so the best diagnostic in the package was written and then dropped, and the scaffolded platform module swallows Run's own return as well. The net effect in every new Warren app was an event parked for ever with nothing printed anywhere.
The default is slog.Default() at emit time, so an application that installs its handler in main is covered without wiring anything.
type Store ¶
type Store interface {
// Append writes records in the caller's ambient transaction. It opens no
// connection of its own: if the transaction rolls back, the records roll
// back with it — that atomicity is the entire pattern. A record whose
// Message.ID is empty is assigned one by the store, stable across
// republishes, which is the key the inbox dedupes on.
Append(ctx context.Context, recs ...Record) error
// Pending returns up to limit undispatched records in insertion order,
// parked records excluded.
Pending(ctx context.Context, limit int) ([]Record, error)
// MarkPublished marks records dispatched.
MarkPublished(ctx context.Context, ids ...string) error
// MarkFailed parks a record: kept for inspection, never returned by
// Pending again.
MarkFailed(ctx context.Context, id string, cause error) error
}
Store is the outbox's one port. Append is the writer: a role the unit of work plays through this method, not a second type. Persistence adapters implement it, because the rows live in their database.
func NewMemoryStore ¶
func NewMemoryStore(opts ...MemoryOption) Store
NewMemoryStore returns an in-process Store implementing Waiter: a slice, a mutex, and a clock.
Test and modular-monolith use only. Undispatched records do not survive a restart, which is the one guarantee an outbox exists to give — a durable store is the persistence adapters' business.
type Waiter ¶
Waiter is the optional low-latency seam a Store may implement: Wait blocks until a record is appended or ctx is done. Relay.Run asserts for it once at start and falls back to PollInterval, so a missed signal delays publication and never loses it. Appending is the signal — Postgres delivers it with LISTEN/NOTIFY, the memory store with a channel.
Source Files