Documentation
¶
Overview ¶
Package events is the one door a module's events leave through.
A module writes an event with Publish, inside the same transaction as the state change that caused it, so the row and the change commit together or not at all: there is no window in which the state moved and the event was lost. The relay in the worker role reads those rows and hands them to a transport — in-process for a single-process run, JetStream for a fleet.
Delivery is at-least-once, and Consume is what turns that into exactly-once handling: it claims each (event, subscription) pair in platformkit_handled inside the handler's own transaction, so a redelivery of work already done finds the claim taken and skips the handler.
This is also the job queue: durable, retried, transactional background work is what an outbox is, and asking for it twice buys nothing. Periodic work is kit/jobs. Both arguments are docs/adr/0004.
Index ¶
- func Consume(ctx context.Context, conn *db.Conn, t Transport, subs []Subscription) error
- func Publish(ctx context.Context, tx db.Tx[db.Tenant], name string, payload any) error
- func PublishFor(ctx context.Context, tx db.Tx[db.System], tenantID uuid.UUID, name string, ...) error
- func Purge(ctx context.Context, conn *db.Conn) error
- func Relay(ctx context.Context, conn *db.Conn, t Transport) error
- func ValidName(name string) bool
- type Event
- type Handler
- type Sink
- type Subscription
- type Transport
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func Consume ¶
Consume subscribes every handler in subs to its event. Each delivery opens a transaction in the event's own tenant, so a handler reaches the tenant's rows the same way a request handler does and can publish events of its own into the same transaction.
Each delivery is claimed before the handler runs, so a handler sees each event once however many times the transport delivers it. See claim.
func Publish ¶
Publish writes an event into the outbox inside tx. It is not a network call and it cannot fail because a broker is down: the row commits with the state change and the relay carries it from there.
It takes a context only to read the actor off it. A db.Tx carries none — a transaction is a handle, not a scope somebody can cancel through — so the caller's own context is the one thing that knows whose request this is.
func PublishFor ¶
func PublishFor(ctx context.Context, tx db.Tx[db.System], tenantID uuid.UUID, name string, payload any) error
PublishFor writes an event from a cross-tenant transaction, naming the tenant it belongs to.
It exists for one shape of work and is the only place in the program where a tenant is an argument to an event. The control plane creates a tenant, and the event that says so has to be written in the transaction that created it — a transaction that belongs to no tenant, about a tenant that did not exist a statement earlier. Publish cannot express that, and publishing afterwards in a second transaction would be an event that can be lost while its cause is kept, which is the exact failure the outbox exists to remove.
A caller needs a db.Tx[db.System] to reach it, so the audience is the modules that already hold the capability. See docs/adr/0006.
func Purge ¶
Purge deletes published rows older than a week, and the handled marks of the same age. kit/jobs calls it hourly in the worker role. Unpublished rows are never touched, however old: a row that has not gone out is a queue entry, not history.
The two windows are one window on purpose. A mark exists to recognise a redelivery of its own event, and an event whose outbox row is gone cannot be relayed again, so a mark older than the row it guards guards nothing. The exact residue is an event a transport still holds unacknowledged a week after the outbox forgot it, which JetStream's own limits make a deployment choice rather than a possibility this code can rule out.
The cutoff is computed by the database and not by Go: a worker whose clock has drifted would otherwise delete a different week's rows than its neighbour. Dead letters are never purged; a row there is an alert somebody has to read.
func Relay ¶
Relay moves every unpublished row to the transport, a batch at a time, and returns when the queue is empty or ctx is done. kit/jobs calls it every second in the worker role.
It takes no lock, and that is deliberate: FOR UPDATE SKIP LOCKED is the concurrency control, so several workers relaying at once each take rows nobody else holds and none of them waits. An advisory lock around it would add nothing and would take something away — a transport that blocks inside one relay pass would hold that lock, and every other periodic job on every replica would stop behind it.
One transaction per batch rather than one for the whole drain, so the row locks are held for a batch and not for a backlog. The caller bounds the whole thing with a deadline; a pass that runs out of time leaves its rows unstamped and the next tick takes them.
Types ¶
type Event ¶
type Event struct {
// ID is the deduplication key. A handler that has already seen it has
// already done the work; see the package comment.
ID uuid.UUID `json:"id"`
// Name is "<module>.<something>", the module's namespace first.
Name string `json:"name"`
// TenantID is the tenant the event happened in. Consume opens the
// handler's transaction in it.
TenantID uuid.UUID `json:"tenantId"`
// Payload is whatever the publisher marshalled.
Payload json.RawMessage `json:"payload"`
// At is when the outbox row was written, which is when the state changed.
At time.Time `json:"at"`
// Actor is the user whose request caused this, and the nil UUID when
// nothing did: a periodic job, the relay, a handler reacting to another
// event. It is not a field a publisher fills in — kit/tenancy carries it on
// the request context and Publish reads it there — because "remember to
// pass the caller through" is the kind of instruction that is followed
// almost everywhere, and an audit trail with holes in it is worse than none.
Actor uuid.UUID `json:"actor"`
}
Event is one thing that happened in one tenant.
type Handler ¶
Handler is what a module does with an event. It runs inside a transaction scoped to the event's tenant, so anything it writes commits with the acknowledgement of the event and rolls back with a redelivery.
type Sink ¶
type Sink struct {
Handle func(ctx context.Context, ev Event) error
Dead func(ctx context.Context, ev Event, cause error)
}
Sink is what a transport does with one event. Handle runs the handler and an error from it is a negative acknowledgement, so the event comes back; Dead is called instead once the transport has stopped bringing it back.
It is a struct rather than a second parameter because the two belong to one subscription, and a transport that had only Handle could only choose between losing a poison event and retrying it forever.
type Subscription ¶
Subscription is one module's interest in one event. A module lists its subscriptions in its manifest; kit/app refuses to start when one names an event no module publishes.
type Transport ¶
type Transport interface {
Publish(ctx context.Context, ev Event) error
// Subscribe delivers every event called name to sink until ctx is done.
// durable names the subscription, so a consumer that restarts resumes where
// it stopped rather than replaying from the beginning.
Subscribe(ctx context.Context, durable, name string, sink Sink) error
}
Transport carries events between processes. There are two implementations and no third: Memory for a single-process run and its tests, JetStream for a fleet. Both deliver at least once, both give up after maxDeliveries, and both dead-letter what they gave up on — a transport that agreed with the other about everything except when to stop would be two policies, not one.
func JetStream ¶
JetStream is the transport for a fleet: NATS JetStream, one stream, durable consumers, explicit acknowledgement. A handler that returns an error nacks, and JetStream redelivers; a handler that succeeds acks, and the event is that consumer's history.
The returned Transport is an io.Closer, so kit/app releases the connection when the worker stops.
func Memory ¶
func Memory() Transport
Memory is the in-process transport: a single-process deployment and every test that does not want a broker. Delivery is asynchronous — the relay hands the event over and returns, and the handler runs on the subscription's own goroutine — because a handler opens a tenant transaction and the relay's transaction is a system one, which it could not nest in.
Source Files