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, and the one that cannot lose an event.
This package is a bridge. It is removed in v1.0.0; import github.com/arandu-io/hesape/events directly.
The outbox, the relay and the event types moved to github.com/arandu-io/hesape/events, which also holds the dispatcher this package never had. Nothing here holds an implementation of them: where the name and the signature survived the move it is a Go alias, and where the design diverged it is an envelope that translates and nothing more. The death date above is what keeps this from being a second way to import one type.
The three envelopes, and what diverged:
NewOutbox hesape/events.NewOutbox takes an interface that can be asked
whether the context is in a transaction, and *data.DB answers
that through a package-level function instead
RelayOptions the Locker field became a *cache.Locks, which nothing outside
hesape can build from a Locker
Relay Run drives the locked pass through that Locker, because the
options it would otherwise be handed cannot carry one
Module stays framework code rather than an envelope. It answers the module contract the kernel collects -- Routes and Migrations included -- so the outbox table travels with the module that owns it.
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(*http.Router)
- func (m *Module) Start(ctx context.Context) error
- type Outbox
- type Publisher
- type PublisherFunc
- type Recorder
- type Relay
- type RelayOptions
- type Stored
Constants ¶
This section is empty.
Variables ¶
var ErrNoTransaction = hevents.ErrNoTransaction
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.
The alias is what keeps it one value: a caller comparing against this name matches the error hesape returns.
Functions ¶
This section is empty.
Types ¶
type Event ¶
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 stays an alias for kernel.Locker, which is the one declaration of it in the framework: the scheduler needs the same thing, and two identical interfaces in two packages is a signature that can drift in one of them. One interface is what lets a single value be wired into both the relay and the scheduler.
It is the one thing here hesape has no counterpart for. There the lock is *cache.Locks, a concrete issuer over a store that can acquire and release by owner, and a Locker -- which only knows how to run a function under a lock it takes and gives back itself -- cannot be turned into one. So the name stays, and Relay.Run below is what drives it.
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 ¶
Outbox stores events in the same transaction as the write.
Store takes a Grant it does not otherwise need, and puts it in the row: who authorized it, which action, which tenant. That is a full audit trail without a second table.
func NewOutbox ¶
NewOutbox returns an outbox over the application's database handle.
An envelope rather than a call through: hesape/events.NewOutbox takes an interface, whose fourth method asks whether the context is inside a transaction on this handle, and *data.DB answers that question through data.InTransaction instead of through a method. The signature here is the one the framework has always had, so every service that builds an outbox from its repository's handle is untouched.
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
type PublisherFunc = hevents.PublisherFunc
PublisherFunc adapts a function to Publisher.
type Recorder ¶
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.
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.
It is an envelope over hesape/events.Relay, which is the code that publishes. What this adds is the Locker: hesape takes a *cache.Locks and there is no way to build one from a Locker, so a relay wired with one runs its ticker here and gives each pass to hesape's Drain under the lock. A relay without a Locker -- which is every relay in a single-replica deployment -- is hesape's loop unchanged.
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.
Without a Locker it is hesape's loop. With one, the loop is here and each tick gives one pass to hesape under the lock, because the lock cannot travel there.
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.
It stays declared here rather than aliasing hesape/events.RelayOptions, whose last field is a *cache.Locks. An alias would change the field every caller that wires a distributed lock is written against -- one line in bootstrap/app.go in every project -- and a bridge that changes a signature is not a bridge.
Source Files