Documentation
¶
Overview ¶
Package events is domain events with an outbox.
There is no Publish(). The naive flow loses data in both directions: if the process dies between the write and the publish, the event never leaves; if the publish happens and the transaction rolls back, the rest of the system reacts to something that did not happen.
So an event is stored in the same transaction as the write that produced it, and a relay publishes it afterwards. One way to do it (RULE 9), and the one that cannot lose an event.
Index ¶
- Variables
- type Event
- type Locker
- type Module
- func (m *Module) Close(ctx context.Context) error
- func (m *Module) Diagnose(ctx context.Context) []string
- func (m *Module) Health(ctx context.Context) error
- func (*Module) Migrations() []kernel.Migration
- func (*Module) Name() string
- func (*Module) Routes(*httpx.Router)
- func (m *Module) Start(ctx context.Context) error
- type Outbox
- func (o *Outbox) Lag(ctx context.Context) (time.Duration, error)
- func (o *Outbox) MarkFailed(ctx context.Context, id string, cause error) error
- func (o *Outbox) MarkPublished(ctx context.Context, id string) error
- func (o *Outbox) Park(ctx context.Context, id string, cause error) error
- func (o *Outbox) Parked(ctx context.Context, limit int) ([]Stored, error)
- func (o *Outbox) Pending(ctx context.Context, tenant string, limit int) ([]Stored, error)
- func (o *Outbox) PendingAll(ctx context.Context, limit int) ([]Stored, error)
- func (o *Outbox) Retry(ctx context.Context, id string) error
- func (o *Outbox) Store(ctx context.Context, g security.Grant, list []Event) error
- type Publisher
- type PublisherFunc
- type Recorder
- type Relay
- type RelayOptions
- type Stored
Constants ¶
This section is empty.
Variables ¶
var ErrNoTransaction = errors.New("events: Store must run inside data.Transaction")
ErrNoTransaction is returned when Store is called outside data.Transaction.
It is an error rather than a fallback, and that is the whole guarantee: an event stored next to a row that then rolled back is worse than no event, and an event stored after the commit is one process crash away from being lost.
Functions ¶
This section is empty.
Types ¶
type Event ¶
type Event struct {
// Name identifies the event: "customer.created", "invoice.paid".
Name string
// Aggregate and AggregateID say what it happened to.
Aggregate string
AggregateID string
// Payload is what the consumer needs, serialized as JSON. Keep it to facts
// that were true when the event occurred: an event that says "look it up" is
// an event that reads a row which has already changed.
Payload any
// OccurredAt defaults to the moment it is stored.
OccurredAt time.Time
}
Event is something that happened, in the past tense.
The name is the vocabulary of the domain rather than of the database: "invoice.paid", not "invoice.updated". A consumer that has to diff two rows to learn what happened is a consumer coupled to your schema.
type Locker ¶ added in v0.5.0
Locker keeps N replicas from publishing the same event N times.
It is an alias for kernel.Locker rather than a second declaration: the scheduler needs the same thing, and two identical interfaces in two packages is a signature that can drift in one of them. github.com/arandu-io/kv implements it, and wiring the distributed lock is one line.
type Module ¶
type Module struct {
// contains filtered or unexported fields
}
Module brings the outbox table, and runs the relay when one is wired.
It registers no routes: it exists so the table travels with the framework rather than being copied into every project's migrations. Register it in bootstrap/app.go next to the modules that store events.
func NewModule ¶
func NewModule() *Module
NewModule returns the module with no relay: the table exists, events are stored, and nothing publishes them yet.
That is a useful state rather than a broken one. Storing is what cannot be recovered later; publishing can start on the day there is something to publish to.
func WithRelay ¶ added in v0.5.0
WithRelay returns the module running the relay in this process.
In-process, like the scheduler and for the same reason: a second deployable for background work is a second thing to monitor, page on, and forget to restart. With more than one replica, give the relay a Locker -- otherwise each one publishes every event.
func (*Module) Close ¶ added in v0.5.0
Close stops the relay and waits for the pass in flight.
Waiting matters: a pass interrupted between publishing and marking published delivers the event again on the next start, and that is the duplicate this framework can avoid rather than the one it cannot.
func (*Module) Diagnose ¶ added in v0.6.0
Diagnose says what is wrong with event delivery, in a sentence.
This is the hint doc 27 asks for: "invoice.paid has been waiting four minutes -- is the relay running?". It shows up on the error page, next to the failure somebody is already looking at, which is the moment they are most likely to act on it.
func (*Module) Health ¶ added in v0.5.0
Health fails when the outbox is falling behind.
A relay that stopped looks exactly like a relay with nothing to do, and the age of the oldest pending event is what tells them apart. Without this, the first sign is a customer asking why they never got the email.
func (*Module) Migrations ¶
Migrations returns the outbox table.
func (*Module) Start ¶ added in v0.10.0
Start begins the relay loop, and only the process that serves calls it.
It used to be Boot, which every command calls: each `aru work` replica ran a relay of its own, and so did `aru routes`. The lock made the duplicate harmless rather than correct. See kernel.Background.
type Outbox ¶
type Outbox struct {
// contains filtered or unexported fields
}
Outbox stores events in the same transaction as the write.
func (*Outbox) Lag ¶ added in v0.5.0
Lag is how long the oldest unpublished event has been waiting.
A relay that stopped looks exactly like a relay with nothing to do. This is the only number that tells them apart, which is why it feeds the health check and the hint on the error page rather than a dashboard.
func (*Outbox) MarkFailed ¶
MarkFailed records an attempt that did not deliver.
The count and the message are stored rather than logged, because the thing anyone needs at 3am is "this event failed 12 times with this message", and a log line from six hours ago does not answer it.
func (*Outbox) MarkPublished ¶
MarkPublished records that an event left.
func (*Outbox) Park ¶ added in v0.5.0
Park stops retrying an event and records why.
An event that failed ten times will not succeed on the eleventh, and a relay stuck on it stops delivering everything behind it. Parking keeps the row -- the payload is the only copy of what happened -- and takes it out of the way.
func (*Outbox) Parked ¶ added in v0.5.0
Parked returns the events that gave up, newest failure first.
A dead letter queue nobody can list is a table that grows. `aru doctor` reports the count, because an event that never left is a business process that silently did not happen.
func (*Outbox) PendingAll ¶ added in v0.5.0
PendingAll returns unpublished events across every tenant, oldest first.
It takes no Grant, and that is deliberate rather than an oversight of RULE 17. The authorization already happened, at write time, and it is recorded in the row -- authorized_by and action are right there. The relay decides nothing: it delivers what was already permitted. This is the same shape as the migrator reading its own table, and it is the only read in the framework that works this way.
It is also why the relay is infrastructure and not a route. Nothing here is reachable from a request.
func (*Outbox) Retry ¶ added in v0.5.0
Retry puts a parked event back in line, with its attempt count reset.
The operator fixed the broker, or the consumer, or the payload. Without this the only way back is SQL by hand, which is how a dead letter queue becomes a table nobody touches.
type Publisher ¶ added in v0.5.0
Publisher is where events go once they are committed.
The framework does not pick one. NATS, a webhook, an in-process handler and a queue are all the same shape from here, and the choice belongs to the application -- what the framework guarantees is that whatever you plug in receives every event that was stored, at least once.
type PublisherFunc ¶ added in v0.5.0
PublisherFunc adapts a function to Publisher.
type Recorder ¶
type Recorder struct {
// contains filtered or unexported fields
}
Recorder is what an entity embeds to collect its own events.
The entity records; the service stores. That split is what keeps the entity free of a database handle and keeps the event next to the rule that produced it.
func (*Recorder) PullEvents ¶
PullEvents returns the recorded events and clears them.
Clearing is the point: an entity stored twice must not emit the same event twice, and the caller that pulls is the one that is about to store.
type Relay ¶ added in v0.5.0
type Relay struct {
// contains filtered or unexported fields
}
Relay publishes what the outbox stored.
Delivery is at-least-once, and that is not a limitation to fix -- it is the price of never losing an event. The consumer deduplicates on Stored.ID, which is why the id is stable and why it travels with the event.
func NewRelay ¶ added in v0.5.0
func NewRelay(o *Outbox, p Publisher, opts RelayOptions) *Relay
NewRelay returns the relay.
func (*Relay) Drain ¶ added in v0.5.0
Drain publishes everything pending, once, and returns.
This is what a test uses. There is no synchronous mode -- the test runs the same code path as production, with the relay executed inline instead of on a ticker. "Sync only in tests" is a second way to do one thing, and the second way always leaks into production.
func (*Relay) Lag ¶ added in v0.5.0
Lag is how long the oldest unpublished event has been waiting.
This is the number that matters: a relay that stopped looks exactly like a relay with nothing to do, and only the age of the oldest pending event tells them apart. It feeds the health check and the hint on the error page.
func (*Relay) Parked ¶ added in v0.6.0
Parked returns the events that gave up, for the diagnosis and for whoever is deciding whether to retry them.
func (*Relay) Run ¶ added in v0.5.0
Run polls until the context is cancelled.
It is started by the module at boot and stopped at shutdown, in the same process as the application -- like the scheduler, and for the same reason: a second deployable to run background work is a second thing to monitor, page on, and forget to restart.
type RelayOptions ¶ added in v0.5.0
type RelayOptions struct {
// Interval is how often the outbox is polled. Default 1s.
//
// Polling rather than LISTEN/NOTIFY, and that is a deliberate trade:
// LISTEN/NOTIFY is lower latency and is Postgres-specific, which would put a
// driver dependency in the core and give SQLite and MySQL a second code
// path. One second of latency on a background publish is not the problem
// this framework exists to solve.
Interval time.Duration
// Batch is how many events one pass publishes. Default 100.
Batch int
// MaxAttempts is how many failures an event gets before it is parked.
// Default 10.
MaxAttempts int
// LockTTL bounds how long one pass may hold the lock. Default 30s.
LockTTL time.Duration
// Locker is the distributed lock. Nil means a single replica.
Locker Locker
}
RelayOptions configures the relay.
type Stored ¶
type Stored struct {
ID string
TenantID string
Name string
Aggregate string
AggregateID string
Payload string
AuthorizedBy string
Action string
OccurredAt time.Time
Attempts int
// LastError is why the most recent attempt failed. It is stored rather than
// logged because the thing anyone needs at 3am is "this event failed twelve
// times with this message", and a log line from six hours ago does not
// answer it.
LastError string
}
Stored is one row of the outbox.
Source Files