sqlbus

package module
v0.0.0-...-ae434bf Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jul 3, 2026 License: MIT Imports: 19 Imported by: 0

Documentation

Overview

Package sqlbus extends the in-process eventbus across application nodes by using a shared SQL database (PostgreSQL or SQLite) as the transport. It is the distributed sibling of the outbox package: where the outbox guarantees that a locally subscribed listener eventually processes an event, sqlbus guarantees that an event published on one node reaches listeners attached on any node of the cluster.

An event is stored as one message row, written inside the caller's business transaction when a unit of work is active. Listeners attach through a Bridge under one of two delivery modes: competing listeners process each event exactly once cluster-wide (arbitrated by a guarded claim in the database), while broadcast listeners process each event once per node, which suits node-local concerns such as cache invalidation. Every node runs a Dispatcher that materializes delivery rows for the listeners it hosts, claims them, delivers through the local in-memory bus, and settles the outcome.

Delivery is at-least-once: a crash between delivering an event and settling its delivery leads to a redelivery after the claim lease expires, so listeners must be idempotent. Polling is the correctness mechanism; wake signals (for example PostgreSQL LISTEN/NOTIFY wired by the caller) are strictly latency hints. The nodes must keep reasonably synchronized clocks: the materialization grace absorbs clock skew, like commit latency, only up to its configured size.

The package is decoupled from the eventbus core: eventbus keeps its zero-dependency guarantee, and applications that do not need cross-node delivery never import sqlbus. An event type must be routed either through the outbox or through sqlbus, never both, or its listeners process it twice.

Index

Constants

View Source 
const SchemaHistoryTable = "event_message_schema_history"

SchemaHistoryTable is the name of the schema-history table that engine-backed Migrator adapters use to record applied migrations. It is exported so an adapter (such as the goway adapter) writes to the same table across upgrades, preserving the on-disk migration-history contract for databases that were previously migrated by that engine.

Variables

View Source 
var (
	// ErrUnknownEventType reports a serialization or deserialization of an
	// event type that was never registered.
	ErrUnknownEventType = errors.New("event type is not registered")

	// ErrConflictingRegistration reports a second registration that binds
	// an already used name or type to something different.
	ErrConflictingRegistration = errors.New("conflicting event type registration")
)
View Source 
var (
	// ErrConflictingDeliveryMode reports an attachment whose delivery mode
	// disagrees with the mode another node already registered for the same
	// listener. Without this arbitration two nodes could silently run one
	// listener as competing and broadcast at the same time, processing every
	// event twice.
	ErrConflictingDeliveryMode = errors.New("listener is already registered under a different delivery mode")

	// ErrConflictingOrdering reports an attachment whose ordering disagrees
	// with the ordering another node already registered for the same
	// listener. Without this arbitration an unordered node would claim
	// deliveries out of order behind the back of the FIFO nodes.
	ErrConflictingOrdering = errors.New("listener is already registered under a different ordering")
)

Functions

func AttachBroadcastListener 

func AttachBroadcastListener[T any](
	ctx context.Context,
	bridge *Bridge,
	id eventbus.ListenerID,
	handle func(ctx context.Context, event T) error,
	options ...AttachOption,
) error

AttachBroadcastListener subscribes the listener on the local bus and registers this node as one broadcast consumer of every event of type T: each event is processed once per hosting node, which suits node-local concerns such as cache invalidation. T must already be registered on the serializer.

func AttachCompetingListener 

func AttachCompetingListener[T any](
	ctx context.Context,
	bridge *Bridge,
	id eventbus.ListenerID,
	handle func(ctx context.Context, event T) error,
	options ...AttachOption,
) error

AttachCompetingListener subscribes the listener on the local bus and registers it as the competing consumer of every event of type T: each event is delivered exactly once cluster-wide, by whichever hosting node claims it first. T must already be registered on the serializer.

A freshly attached listener starts at the attachment time minus the materialization grace, so publications still in flight are never missed; it does not replay older history.

func RegisterEventType 

func RegisterEventType[T any](serializer *JSONSerializer, name string) error

RegisterEventType maps the event type T to a stable persistent name. Registering the same pair again is allowed; binding a name or a type that is already bound differently is rejected, because it would silently deserialize stored messages into the wrong type.

func Schema 

func Schema() fs.FS

Schema returns the embedded sqlbus migration scripts as a read-only file system rooted at the migration directory, so its entries are "V1__create_event_message_tables.sql" and any successors, not "migration/V1__...". Both the native Migrator and any external adapter read these exact scripts, so the schema has a single source.

The returned value is an immutable view over the package embedded files; callers cannot mutate them.

Types

type AttachOption 

type AttachOption func(*attachmentConfiguration)

AttachOption customizes one listener at attachment time.

func WithListenerMaximumAttempts 

func WithListenerMaximumAttempts(attempts int) AttachOption

WithListenerMaximumAttempts overrides, for this listener, how many delivery attempts a delivery may consume before it becomes exhausted. Configure the same value on every node hosting the listener.

func WithListenerRetryDelay 

func WithListenerRetryDelay(delay func(attempt int) time.Duration) AttachOption

WithListenerRetryDelay overrides, for this listener, the backoff schedule of failed deliveries. Configure the same schedule on every node hosting the listener.

func WithOrderedDelivery 

func WithOrderedDelivery() AttachOption

WithOrderedDelivery attaches the listener as a FIFO consumer: its events are processed strictly in publication order, one at a time, cluster-wide for a competing listener and per node for a broadcast one. Ordered deliveries wait below the materialization frontier, so their latency is at least the materialization grace; every node must attach the listener with the same ordering.

type AttachmentRegistration 

type AttachmentRegistration struct {
	// ID identifies the listener; it must be unique within the bus.
	ID eventbus.ListenerID

	// Matches decides which events the listener receives on the local bus.
	Matches func(event any) bool

	// Probe carries the zero value of the consumed event type, so the bridge
	// resolves the persistent event type name through its serializer without
	// invoking the handler.
	Probe any

	// Handle processes one delivered event.
	Handle eventbus.Handler

	// Mode selects competing or broadcast consumption.
	Mode DeliveryMode

	// Options carry the per-listener delivery behavior.
	Options []AttachOption
}

AttachmentRegistration describes one listener to Attach: the untyped seam the typed attachment functions and the Broker adapter both go through.

type Bridge 

type Bridge struct {
	// contains filtered or unexported fields
}

Bridge connects the local in-process bus to the shared database: it stores published events as messages, pre-creates the delivery rows of the listeners attached on this node, and runs the claim, deliver, and settle state machine both for the Publisher's after-commit path and for the Dispatcher's polling path.

Attach every listener and register every event type before publishing or starting a Dispatcher. A Bridge is safe for concurrent use.

func NewBridge 

func NewBridge(store Store, bus EventBus, serializer Serializer, options ...BridgeOption) *Bridge

NewBridge constructs a Bridge over the store, the local bus, and the serializer. Without WithNodeIdentifier the node identity is generated per process, so a restarted node joins as a fresh broadcast consumer while its previous identity expires through the heartbeat.

func (*Bridge) Attach 

func (b *Bridge) Attach(ctx context.Context, registration AttachmentRegistration) error

Attach subscribes the listener on the local bus and registers it durably under its delivery mode and ordering. It is the untyped seam the typed attachment functions and the Broker adapter go through.

func (*Bridge) Detach 

func (b *Bridge) Detach(ctx context.Context, id eventbus.ListenerID) error

Detach removes the durable registrations of the listener, unpinning its messages from retention. It is the decommissioning step for a listener whose code was removed: call it after the last node hosting the listener stopped, because a node that still hosts it — including this one, when its Dispatcher is still running — re-registers the consumer on its next heartbeat. The subscription on the local in-process bus remains until the process restarts.

func (*Bridge) FindExhausted 

func (b *Bridge) FindExhausted(ctx context.Context, limit int) ([]DueDelivery, error)

FindExhausted returns the dead letters — deliveries that consumed their attempt budget — oldest first, up to the limit. Each carries its message and the last failure cause, so an operator can inspect them and revive one with Resubmit.

func (*Bridge) Node 

func (b *Bridge) Node() NodeID

Node returns the identity this bridge participates in the cluster under.

func (*Bridge) Publish 

func (b *Bridge) Publish(ctx context.Context, querier Querier, event any) (Message, []DeliveryKey, error)

Publish stores the event as one message, together with the pending delivery rows of the listeners attached on this node, through the provided querier — so the rows join the caller's transaction. Deliveries of listeners hosted on other nodes are materialized there by their dispatchers.

Publish is the low-level seam for callers that manage their own transactions; the stored deliveries are picked up by the dispatchers once the transaction commits. Most callers use Publisher instead, which resolves the querier from the active unit of work and dispatches the local deliveries immediately after commit.

func (*Bridge) Resubmit 

func (b *Bridge) Resubmit(ctx context.Context, key DeliveryKey) (bool, error)

Resubmit gives a failed or exhausted delivery a fresh attempt budget, so an operator can revive a dead letter after fixing its listener. It reports false when the delivery is not in a resubmittable state.

type BridgeOption 

type BridgeOption func(*Bridge)

BridgeOption customizes a Bridge at construction time.

func WithLeaseDuration 

func WithLeaseDuration(lease time.Duration) BridgeOption

WithLeaseDuration overrides how long a claim protects a processing delivery before another dispatcher may steal it (default 5 minutes). It must exceed the slowest listener, or a slow delivery is repeated on another node. Configure the same value on every node.

func WithMaterializationGrace 

func WithMaterializationGrace(grace time.Duration) BridgeOption

WithMaterializationGrace overrides the overlap window that protects materialization against publications whose transaction commits after later events became visible (default 10 minutes). It must exceed the longest business transaction that publishes events; a freshly attached listener may receive events published up to this long before its attachment. Configure the same value on every node.

func WithMaximumAttempts 

func WithMaximumAttempts(attempts int) BridgeOption

WithMaximumAttempts overrides how many delivery attempts a delivery may consume before it becomes exhausted (default 5). Configure the same value on every node: the budget is evaluated against the shared attempt counter.

func WithNodeIdentifier 

func WithNodeIdentifier(node NodeID) BridgeOption

WithNodeIdentifier overrides the generated per-process node identifier. A stable identifier makes a restarted node resume its broadcast consumers instead of starting fresh ones; it must be unique per running process.

func WithRetryDelay 

func WithRetryDelay(delay func(attempt int) time.Duration) BridgeOption

WithRetryDelay overrides the backoff schedule of failed deliveries: the function receives the attempt count just spent (1 for the first attempt) and returns how long to wait before the next one. The default doubles from 5 seconds up to 5 minutes.

type Broker 

type Broker struct {
	// contains filtered or unexported fields
}

Broker adapts the sqlbus machinery to the eventbus.Broker contract: the same consumer-facing interface as the in-memory engine, with persistence and cross-node consumption added underneath. Publications go through the transactional Publisher, so they join the caller's unit of work; consumers attach durably through the Bridge, competing cluster-wide by default and broadcast per node on request.

Delivery settles exactly once per consumer, but a crash or an expired claim lease re-executes the handler, so handlers must be idempotent — the documented cost of durability. The Workers consumer option is not interpreted: the concurrency of this engine is governed by the Dispatcher batch size and by how many nodes host the listener.

func NewBroker 

func NewBroker(bridge *Bridge, publisher *Publisher) *Broker

NewBroker constructs a Broker over the bridge and the publisher. The caller keeps running a Dispatcher per node, exactly as with the lower-level API.

func (*Broker) FindExhausted 

func (b *Broker) FindExhausted(ctx context.Context, limit int) ([]eventbus.DeadLetter, error)

FindExhausted returns the dead letters, oldest first, up to the limit. The event of a dead letter is restored from its serialized form when the type is registered on this node's serializer.

func (*Broker) Publish 

func (b *Broker) Publish(ctx context.Context, event any) error

Publish stores the event through the current unit of work and schedules its deliveries. It reports only serialization and persistence failures; handler outcomes settle asynchronously and surface as dead letters.

func (*Broker) Resubmit 

func (b *Broker) Resubmit(ctx context.Context, reference string) (bool, error)

Resubmit gives the referenced dead letter a fresh attempt budget. It reports false when the reference is unknown or was already resubmitted.

func (*Broker) Subscribe 

func (b *Broker) Subscribe(ctx context.Context, registration eventbus.ConsumerRegistration) error

Subscribe attaches the consumer durably. The consumed event type must be registered on the serializer of the bridge.

type Consumer 

type Consumer struct {
	ListenerID       eventbus.ListenerID
	Instance         string
	EventType        string
	DeliveryMode     DeliveryMode
	StartBoundary    time.Time
	Frontier         time.Time
	RegistrationDate time.Time
	HeartbeatDate    time.Time
}

Consumer is one durable registration: the declaration that a listener, under a delivery mode, consumes one event type starting at a boundary. Instance follows the DeliveryKey convention. Frontier is the advancing publication-date floor below which every visible message already has its delivery row, which bounds the materialization scan; StartBoundary is the fixed registration-time floor used to decide whether the consumer covers a message at all.

func (Consumer) Key 

func (c Consumer) Key() ConsumerKey

Key returns the identifying part of the consumer.

type ConsumerKey 

type ConsumerKey struct {
	ListenerID eventbus.ListenerID
	Instance   string
	EventType  string
}

ConsumerKey identifies one durable consumer registration.

type Delivery 

type Delivery struct {
	Key             DeliveryKey
	Status          Status
	Attempts        int
	ClaimToken      string
	ClaimDate       *time.Time
	NextAttemptDate *time.Time
	CompletionDate  *time.Time
	LastError       string
}

Delivery is one delivery row: the unit the claim and settlement state machine runs on.

type DeliveryKey 

type DeliveryKey struct {
	MessageID  uuid.UUID
	ListenerID eventbus.ListenerID
	Instance   string
}

DeliveryKey identifies one delivery: the processing of one message by one consumer of one listener. Instance is empty for a competing listener, whose single cluster-wide consumer is the listener itself, and holds the NodeID of the consuming node for a broadcast listener.

type DeliveryMode 

type DeliveryMode string

DeliveryMode selects how the cluster shares the work of one listener. 

const (
	// DeliveryModeCompeting delivers each event to exactly one node hosting
	// the listener, so homogeneous replicas share the work instead of
	// repeating it. It is the safe default for scaled deployments.
	DeliveryModeCompeting DeliveryMode = "COMPETING"

	// DeliveryModeBroadcast delivers each event to every node hosting the
	// listener, once per node, for node-local concerns such as invalidating
	// an in-memory cache.
	DeliveryModeBroadcast DeliveryMode = "BROADCAST"
)

type Dialect 

type Dialect string

Dialect identifies the target database without leaking any third-party type into the sqlbus core. It is a closed string enum: the core only ever produces DialectSQLite and DialectPostgres, set by the respective store constructors. A Migrator adapter switches on it to choose its own dialect representation. 

const (
	// DialectSQLite identifies a SQLite database.
	DialectSQLite Dialect = "sqlite"
	// DialectPostgres identifies a PostgreSQL database.
	DialectPostgres Dialect = "postgres"
)

type Dispatcher 

type Dispatcher struct {
	// contains filtered or unexported fields
}

Dispatcher is the per-node delivery loop: it materializes the delivery rows of the listeners attached on its bridge, claims the due ones, delivers them through the local in-process bus, and settles the outcome. It also runs the shared maintenance duties — consumer heartbeats, broadcast expiry, retention, and orphan cleanup — on a slower cadence, so no separately managed component guards the tables against unbounded growth.

Every node that hosts listeners must run one Dispatcher. Polling is the correctness mechanism; a wake signal only shortens the latency of a pass.

func NewDispatcher 

func NewDispatcher(bridge *Bridge, options ...DispatcherOption) *Dispatcher

NewDispatcher constructs a Dispatcher over the bridge.

func (*Dispatcher) Start 

func (d *Dispatcher) Start() (stop func())

Start launches the background loops: one delivery pass immediately, which picks up the backlog of a previous run, then one pass per jittered interval or wake signal, with the consumer heartbeats on their own cadence so a long delivery pass cannot starve liveness. The returned stop function cancels the loops and waits for an in-flight pass to finish; a delivery in flight at that moment still settles its outcome.

type DispatcherOption 

type DispatcherOption func(*Dispatcher)

DispatcherOption customizes a Dispatcher at construction time.

func WithBatchSize 

func WithBatchSize(size int) DispatcherOption

WithBatchSize overrides how many due deliveries one pass claims per attached listener before looking again (default 100).

func WithConsumerExpiry 

func WithConsumerExpiry(expiry time.Duration) DispatcherOption

WithConsumerExpiry overrides how long a broadcast consumer may miss its heartbeats before it is considered gone and reaped (default 15 minutes). It must dwarf the worst scheduling or database stall a live node can suffer, or a paused node is wrongly reaped and re-registers with a fresh boundary. Configure the same value on every node.

func WithMaintenanceInterval 

func WithMaintenanceInterval(interval time.Duration) DispatcherOption

WithMaintenanceInterval overrides how often the maintenance duties run (default 1 minute).

func WithMaximumMessageAge 

func WithMaximumMessageAge(age time.Duration) DispatcherOption

WithMaximumMessageAge overrides the hard age cap after which a message is removed even when unsettled deliveries still pin it (default 30 days). The cap bounds table growth when a dead letter or an abandoned consumer would otherwise pin messages forever; every forced removal is reported loudly. Configure the same value on every node.

func WithPollInterval 

func WithPollInterval(interval time.Duration) DispatcherOption

WithPollInterval overrides how often the dispatcher runs a delivery pass (default 1 second). The interval is jittered by ±10 percent so the nodes of a cluster spread their load instead of polling in step.

func WithSettledRetention 

func WithSettledRetention(retention time.Duration) DispatcherOption

WithSettledRetention overrides how long a fully settled message is kept before retention removes it (default 72 hours). Configure the same value on every node.

func WithWakeSignal 

func WithWakeSignal(wake <-chan struct{}) DispatcherOption

WithWakeSignal makes a receive on the channel trigger an immediate pass, so a caller can wire a latency hint such as PostgreSQL LISTEN/NOTIFY. The signal is strictly best-effort: a lost wake-up costs at most one poll interval, never an event.

type DueDelivery 

type DueDelivery struct {
	Delivery Delivery
	Message  Message
}

DueDelivery pairs a claimable delivery with the message it delivers, so a dispatcher restores the event without a second read.

type EventBus 

type EventBus interface {
	Subscribe(id eventbus.ListenerID, matches func(event any) bool, handle eventbus.Handler) error
	Deliver(ctx context.Context, id eventbus.ListenerID, event any) error
}

EventBus is the slice of the in-process bus the bridge relies on. It is satisfied by *eventbus.Bus.

type JSONSerializer 

type JSONSerializer struct {
	// contains filtered or unexported fields
}

JSONSerializer is a Serializer based on encoding/json. Event types must be registered with RegisterEventType under a stable name, which decouples the persisted representation from Go type names across refactorings.

func NewJSONSerializer 

func NewJSONSerializer() *JSONSerializer

NewJSONSerializer constructs a JSONSerializer with no registered types.

func (*JSONSerializer) Deserialize 

func (s *JSONSerializer) Deserialize(eventType string, payload string) (any, error)

Deserialize reconstructs the event registered under the given name.

func (*JSONSerializer) Serialize 

func (s *JSONSerializer) Serialize(event any) (string, string, error)

Serialize renders the event as JSON under its registered name.

type Message 

type Message struct {
	ID              uuid.UUID
	EventType       string
	SerializedEvent string
	PublisherNode   NodeID
	PublicationDate time.Time
}

Message is one published event as stored in the shared database: the payload every delivery of that event refers to.

type Migrator 

type Migrator interface {
	// Migrate applies schema to db for the given dialect.
	Migrate(ctx context.Context, db *sql.DB, dialect Dialect, schema fs.FS) error
}

Migrator brings the sqlbus schema up to date against an open database. It is the single seam the store uses for schema initialization, so the choice of migration engine stays out of the query code. The store calls Migrate once, from Store.Initialize.

Implementations receive the open *sql.DB (a concrete handle, because some engines require one rather than the narrower Querier), the target Dialect, and the embedded schema scripts as an fs.FS rooted at the directory that holds the SQL files, so an adapter never duplicates the SQL.

Implementations must be idempotent: Initialize may run on every start, so calling Migrate against an already-migrated database must be a no-op.

The core ships a native database/sql implementation (NativeMigrator) as the zero-configuration default, so the common case needs no Migrator and the core go.mod carries no migration-engine dependency. A goway-backed implementation lives in its own module (github.com/cgardev/gokeel/sqlbus/gowaymigrator) so that only clients who opt in pull goway into their build.

type MigratorFunc 

type MigratorFunc func(ctx context.Context, db *sql.DB, dialect Dialect, schema fs.FS) error

MigratorFunc adapts an ordinary function to the Migrator interface, so a caller can supply a one-off migration strategy without declaring a type.

func (MigratorFunc) Migrate 

func (f MigratorFunc) Migrate(ctx context.Context, db *sql.DB, dialect Dialect, schema fs.FS) error

Migrate calls f.

type NativeMigrator 

type NativeMigrator struct{}

NativeMigrator applies the sqlbus schema using only database/sql. It is the default Migrator wired by the store constructors, so the common case pulls in no third-party migration engine.

It executes each embedded *.sql script in ascending file-name order, recording the applied script names so a later run skips them. A database migrated before the record table existed re-runs the historical scripts, whose IF NOT EXISTS DDL is a no-op, and an ALTER TABLE ADD COLUMN that finds its column already present is tolerated, so every upgrade path converges. The dialect is accepted for interface symmetry but is not needed here, because the DDL is portable across SQLite and PostgreSQL.

func (NativeMigrator) Migrate 

func (NativeMigrator) Migrate(ctx context.Context, db *sql.DB, _ Dialect, schema fs.FS) error

Migrate reads every *.sql script from schema in name order, splits each into individual statements, and executes the not-yet-applied ones. It is idempotent.

type NodeID 

type NodeID string

NodeID identifies one application process in the cluster. It is the consumer instance of every broadcast delivery the node handles.

type Option 

type Option func(*sqlStore)

Option customizes a store at construction time.

func WithMigrator 

func WithMigrator(m Migrator) Option

WithMigrator overrides the default native Migrator the store uses in Initialize. The default keeps the sqlbus core free of any migration-engine dependency; pass gowaymigrator.New() from github.com/cgardev/gokeel/sqlbus/gowaymigrator to opt in to goway.

type Ordering 

type Ordering string

Ordering selects how the deliveries of one listener are worked through. 

const (
	// OrderingUnordered processes the deliveries of the listener as they
	// become due, with no ordering guarantee: a failing delivery retries on
	// its own schedule without delaying the others.
	OrderingUnordered Ordering = "UNORDERED"

	// OrderingFIFO processes the deliveries of the listener strictly in
	// publication order, one at a time per consumer, cluster-wide for a
	// competing listener and per node for a broadcast one. A failing delivery
	// blocks its successors while it retries; once it exhausts its attempt
	// budget it parks as a dead letter and the queue continues. Ordered
	// deliveries wait below the materialization frontier, so their latency is
	// at least the materialization grace.
	OrderingFIFO Ordering = "FIFO"
)

type PostgresStore 

type PostgresStore struct {
	// contains filtered or unexported fields
}

PostgresStore is a Store backed by a PostgreSQL database.

func NewPostgresStore 

func NewPostgresStore(database *sql.DB, options ...Option) *PostgresStore

NewPostgresStore constructs a PostgresStore on top of an open PostgreSQL database. The schema and queries are the same as the SQLite store; only the rendering dialect and the migration dialect differ.

func (PostgresStore) AdvanceFrontier 

func (s PostgresStore) AdvanceFrontier(ctx context.Context, key ConsumerKey, frontier time.Time) error

AdvanceFrontier raises the materialization floor of the consumer. The monotonic guard makes concurrent advances from several nodes converge on the highest value.

func (PostgresStore) ClaimDelivery 

func (s PostgresStore) ClaimDelivery(
	ctx context.Context, key DeliveryKey, token string,
	now time.Time, leaseCutoff time.Time, attempts int,
) (bool, error)

ClaimDelivery atomically transitions the delivery into processing under the given token, counting the attempt. The eligibility and attempt guards are re-evaluated by the update itself, so exactly one of several concurrent claimants wins; the losers observe zero affected rows.

func (PostgresStore) ClaimDeliveryInOrder 

func (s PostgresStore) ClaimDeliveryInOrder(
	ctx context.Context, key DeliveryKey, token string,
	now time.Time, leaseCutoff time.Time, attempts int, publicationDate time.Time,
) (bool, error)

ClaimDeliveryInOrder claims like ClaimDelivery but re-verifies inside the update that no earlier incomplete delivery of the same consumer exists — between the find and the claim a predecessor may have been resubmitted, and its revived delivery must run first — and that no other delivery of the consumer is processing under a live lease, so FIFO execution stays serial even when a revived predecessor and its running successor race. The claimed row never excludes itself, because nothing is earlier than itself in the total order.

func (PostgresStore) CompleteDelivery 

func (s PostgresStore) CompleteDelivery(
	ctx context.Context, key DeliveryKey, token string, completionDate time.Time,
) (bool, error)

CompleteDelivery settles a successful delivery; it reports whether this caller's settlement won.

func (PostgresStore) CreateDeliveries 

func (s PostgresStore) CreateDeliveries(ctx context.Context, querier Querier, keys []DeliveryKey) error

CreateDeliveries writes one pending delivery row per key through the provided querier. The conflict-tolerant insert lets a publisher-created row and a dispatcher-materialized row for the same key converge instead of failing.

func (PostgresStore) CreateMessage 

func (s PostgresStore) CreateMessage(ctx context.Context, querier Querier, message Message) error

CreateMessage writes the message through the provided querier, so it joins the transaction of the business change that produced the event.

func (PostgresStore) DeleteMessagesOlderThan 

func (s PostgresStore) DeleteMessagesOlderThan(ctx context.Context, cutoff time.Time) (int64, error)

DeleteMessagesOlderThan unconditionally removes messages older than the cutoff. It is the hard age cap that bounds table growth when unsettled deliveries would otherwise pin messages forever; the caller reports every non-zero count loudly.

func (PostgresStore) DeleteOrphanDeliveries 

func (s PostgresStore) DeleteOrphanDeliveries(ctx context.Context) (int64, error)

DeleteOrphanDeliveries removes deliveries whose message was deleted and deliveries whose consumer registration no longer exists. It must run after message deletion, never before: completed delivery rows are the memory of the materialization anti-join, so removing them while their message survives would resurrect the message as a fresh delivery.

func (PostgresStore) DeleteSettledMessages 

func (s PostgresStore) DeleteSettledMessages(ctx context.Context, olderThan time.Time) (int64, error)

DeleteSettledMessages removes messages older than the reference that every covering registered consumer has completed. Exhausted deliveries do not count as settled: a dead letter pins its message, and so its payload, until Bridge.Resubmit revives it or the hard age cap removes it loudly.

func (PostgresStore) ExpireBroadcastConsumers 

func (s PostgresStore) ExpireBroadcastConsumers(ctx context.Context, cutoff time.Time) (int64, error)

ExpireBroadcastConsumers removes broadcast registrations whose heartbeat is older than the cutoff, so consumers of nodes that left the cluster stop pinning messages. Competing registrations are durable and never expire.

func (PostgresStore) FailDelivery 

func (s PostgresStore) FailDelivery(
	ctx context.Context,
	key DeliveryKey,
	token string,
	cause string,
	nextAttemptDate time.Time,
	attempts int,
	maximumAttempts int,
) (bool, error)

FailDelivery settles a failed delivery: it records the cause and the next attempt date, and moves the delivery to failed, or to exhausted once the attempt budget is spent. The token fence guarantees the attempts value set by this dispatcher's claim is still current, so the exhaustion decision is computed from it without re-reading the row.

func (PostgresStore) FindDueDeliveries 

func (s PostgresStore) FindDueDeliveries(
	ctx context.Context,
	key ConsumerKey,
	ordering Ordering,
	now time.Time,
	leaseCutoff time.Time,
	limit int,
) ([]DueDelivery, error)

FindDueDeliveries returns the claimable deliveries of the consumer in the total order (publication_date, message_id). For a FIFO consumer only the head of the queue is claimable, and only below the materialization frontier: below that watermark every visible message already has its delivery row, so no late-committing publication can slot in front of the head.

func (PostgresStore) FindExhaustedDeliveries 

func (s PostgresStore) FindExhaustedDeliveries(ctx context.Context, limit int) ([]DueDelivery, error)

FindExhaustedDeliveries returns the dead letters, oldest publication first, up to the limit.

func (PostgresStore) Heartbeat 

func (s PostgresStore) Heartbeat(ctx context.Context, key ConsumerKey, at time.Time) (bool, error)

Heartbeat refreshes the liveness timestamp of the consumer. It reports false when the registration row no longer exists (for example, after a broadcast expiry reaped it), so the caller can re-register with a freshly computed boundary instead of resurrecting a stale one.

func (PostgresStore) Initialize 

func (s PostgresStore) Initialize(ctx context.Context) error

Initialize brings the database schema up to date by applying the embedded migration scripts through the configured Migrator. The default native Migrator uses database/sql only; a goway-backed Migrator can be supplied with WithMigrator.

func (PostgresStore) MaterializeDeliveries 

func (s PostgresStore) MaterializeDeliveries(ctx context.Context, key ConsumerKey) (int64, error)

MaterializeDeliveries inserts one pending delivery row for every message the consumer covers but has no row for yet, scanning from the consumer's durable frontier. The frontier is read from the consumer row inside the statement, so the scan floor is always the durable one; a missing consumer row (for example, one reaped by broadcast expiry) yields a NULL frontier, whose comparison matches no rows. The composite primary key makes the insert idempotent, so concurrent materializations converge.

func (PostgresStore) RegisterConsumer 

func (s PostgresStore) RegisterConsumer(ctx context.Context, consumer Consumer) error

RegisterConsumer records the durable consumer registration. Registering an existing consumer again keeps the stored boundary and frontier, so a durable group resumes where it left off, and refreshes only the heartbeat: a node re-attaching under a stable identity must rejoin with fresh liveness, not with the staleness it accumulated while it was down.

func (PostgresStore) RegisterListenerMode 

func (s PostgresStore) RegisterListenerMode(
	ctx context.Context, id eventbus.ListenerID, mode DeliveryMode, ordering Ordering,
) (DeliveryMode, Ordering, error)

RegisterListenerMode records the delivery mode and the ordering of the listener with first-registration-wins semantics and returns the pair that won, so the caller can detect a conflicting registration made by another node.

func (PostgresStore) RemoveListener 

func (s PostgresStore) RemoveListener(ctx context.Context, id eventbus.ListenerID) error

RemoveListener deletes every consumer registration and the delivery-mode row of the listener, unpinning its messages from retention.

func (PostgresStore) ResubmitDelivery 

func (s PostgresStore) ResubmitDelivery(ctx context.Context, key DeliveryKey) (bool, error)

ResubmitDelivery gives a failed or exhausted delivery a fresh attempt budget, clearing the backoff of the failed attempts so the fresh budget starts immediately. It reports false when the delivery is not in a resubmittable state.

type Publisher 

type Publisher struct {
	// contains filtered or unexported fields
}

Publisher is the bridge between a business write and the shared bus: it stores the message inside the current unit of work and delivers it to the locally attached listeners only after that unit commits. Listeners on other nodes receive the event through their own Dispatcher.

When no unit of work is active, the originating write has already auto-committed, so the local deliveries are dispatched immediately instead. A Publisher is immutable after construction and safe for concurrent use.

func NewPublisher 

func NewPublisher(bridge *Bridge, querier QuerierSource) *Publisher

NewPublisher constructs a Publisher that writes through the querier source and dispatches the local deliveries synchronously after commit.

func (*Publisher) Publish 

func (p *Publisher) Publish(ctx context.Context, event any) error

Publish stores the event as one message through the current querier, pre-creates the delivery rows of the listeners attached on this node, and schedules their delivery for after the outermost transaction commits. When no unit of work is active the rows are already durable, so local delivery happens at once. Delivery failures do not fail the call: the affected deliveries stay incomplete and are recovered by a Dispatcher.

The message is stored even when no listener is attached locally, because listeners on other nodes are unknown at publish time; a message no registered consumer covers is removed by retention.

func (*Publisher) WithAsynchronousDispatch 

func (p *Publisher) WithAsynchronousDispatch() *Publisher

WithAsynchronousDispatch returns a Publisher that hands committed local deliveries to a background goroutine, so callers in a request path do not wait for slow listeners. The at-least-once guarantee is unchanged: deliveries settle only after their listener succeeds, and incomplete ones are recovered by a Dispatcher.

type Querier 

type Querier interface {
	QueryContext(ctx context.Context, query string, args ...any) (*sql.Rows, error)
	ExecContext(ctx context.Context, query string, args ...any) (sql.Result, error)
}

Querier is the minimal execution surface the store runs its statements against. It is satisfied by *sql.DB, *sql.Tx, and *sql.Conn, and so by the querier transaction resolves from the context.

type QuerierSource 

type QuerierSource interface {
	Querier(ctx context.Context) transaction.Querier
}

QuerierSource resolves the querier the message rows are written through. It is satisfied by *transaction.Manager, whose Querier returns the active transaction when a unit of work is in progress.

type SQLiteStore 

type SQLiteStore struct {
	// contains filtered or unexported fields
}

SQLiteStore is a Store backed by a SQLite database.

func NewSQLiteStore 

func NewSQLiteStore(database *sql.DB, options ...Option) *SQLiteStore

NewSQLiteStore constructs a SQLiteStore on top of an open SQLite database.

func (SQLiteStore) AdvanceFrontier 

func (s SQLiteStore) AdvanceFrontier(ctx context.Context, key ConsumerKey, frontier time.Time) error

AdvanceFrontier raises the materialization floor of the consumer. The monotonic guard makes concurrent advances from several nodes converge on the highest value.

func (SQLiteStore) ClaimDelivery 

func (s SQLiteStore) ClaimDelivery(
	ctx context.Context, key DeliveryKey, token string,
	now time.Time, leaseCutoff time.Time, attempts int,
) (bool, error)

ClaimDelivery atomically transitions the delivery into processing under the given token, counting the attempt. The eligibility and attempt guards are re-evaluated by the update itself, so exactly one of several concurrent claimants wins; the losers observe zero affected rows.

func (SQLiteStore) ClaimDeliveryInOrder 

func (s SQLiteStore) ClaimDeliveryInOrder(
	ctx context.Context, key DeliveryKey, token string,
	now time.Time, leaseCutoff time.Time, attempts int, publicationDate time.Time,
) (bool, error)

ClaimDeliveryInOrder claims like ClaimDelivery but re-verifies inside the update that no earlier incomplete delivery of the same consumer exists — between the find and the claim a predecessor may have been resubmitted, and its revived delivery must run first — and that no other delivery of the consumer is processing under a live lease, so FIFO execution stays serial even when a revived predecessor and its running successor race. The claimed row never excludes itself, because nothing is earlier than itself in the total order.

func (SQLiteStore) CompleteDelivery 

func (s SQLiteStore) CompleteDelivery(
	ctx context.Context, key DeliveryKey, token string, completionDate time.Time,
) (bool, error)

CompleteDelivery settles a successful delivery; it reports whether this caller's settlement won.

func (SQLiteStore) CreateDeliveries 

func (s SQLiteStore) CreateDeliveries(ctx context.Context, querier Querier, keys []DeliveryKey) error

CreateDeliveries writes one pending delivery row per key through the provided querier. The conflict-tolerant insert lets a publisher-created row and a dispatcher-materialized row for the same key converge instead of failing.

func (SQLiteStore) CreateMessage 

func (s SQLiteStore) CreateMessage(ctx context.Context, querier Querier, message Message) error

CreateMessage writes the message through the provided querier, so it joins the transaction of the business change that produced the event.

func (SQLiteStore) DeleteMessagesOlderThan 

func (s SQLiteStore) DeleteMessagesOlderThan(ctx context.Context, cutoff time.Time) (int64, error)

DeleteMessagesOlderThan unconditionally removes messages older than the cutoff. It is the hard age cap that bounds table growth when unsettled deliveries would otherwise pin messages forever; the caller reports every non-zero count loudly.

func (SQLiteStore) DeleteOrphanDeliveries 

func (s SQLiteStore) DeleteOrphanDeliveries(ctx context.Context) (int64, error)

DeleteOrphanDeliveries removes deliveries whose message was deleted and deliveries whose consumer registration no longer exists. It must run after message deletion, never before: completed delivery rows are the memory of the materialization anti-join, so removing them while their message survives would resurrect the message as a fresh delivery.

func (SQLiteStore) DeleteSettledMessages 

func (s SQLiteStore) DeleteSettledMessages(ctx context.Context, olderThan time.Time) (int64, error)

DeleteSettledMessages removes messages older than the reference that every covering registered consumer has completed. Exhausted deliveries do not count as settled: a dead letter pins its message, and so its payload, until Bridge.Resubmit revives it or the hard age cap removes it loudly.

func (SQLiteStore) ExpireBroadcastConsumers 

func (s SQLiteStore) ExpireBroadcastConsumers(ctx context.Context, cutoff time.Time) (int64, error)

ExpireBroadcastConsumers removes broadcast registrations whose heartbeat is older than the cutoff, so consumers of nodes that left the cluster stop pinning messages. Competing registrations are durable and never expire.

func (SQLiteStore) FailDelivery 

func (s SQLiteStore) FailDelivery(
	ctx context.Context,
	key DeliveryKey,
	token string,
	cause string,
	nextAttemptDate time.Time,
	attempts int,
	maximumAttempts int,
) (bool, error)

FailDelivery settles a failed delivery: it records the cause and the next attempt date, and moves the delivery to failed, or to exhausted once the attempt budget is spent. The token fence guarantees the attempts value set by this dispatcher's claim is still current, so the exhaustion decision is computed from it without re-reading the row.

func (SQLiteStore) FindDueDeliveries 

func (s SQLiteStore) FindDueDeliveries(
	ctx context.Context,
	key ConsumerKey,
	ordering Ordering,
	now time.Time,
	leaseCutoff time.Time,
	limit int,
) ([]DueDelivery, error)

FindDueDeliveries returns the claimable deliveries of the consumer in the total order (publication_date, message_id). For a FIFO consumer only the head of the queue is claimable, and only below the materialization frontier: below that watermark every visible message already has its delivery row, so no late-committing publication can slot in front of the head.

func (SQLiteStore) FindExhaustedDeliveries 

func (s SQLiteStore) FindExhaustedDeliveries(ctx context.Context, limit int) ([]DueDelivery, error)

FindExhaustedDeliveries returns the dead letters, oldest publication first, up to the limit.

func (SQLiteStore) Heartbeat 

func (s SQLiteStore) Heartbeat(ctx context.Context, key ConsumerKey, at time.Time) (bool, error)

Heartbeat refreshes the liveness timestamp of the consumer. It reports false when the registration row no longer exists (for example, after a broadcast expiry reaped it), so the caller can re-register with a freshly computed boundary instead of resurrecting a stale one.

func (SQLiteStore) Initialize 

func (s SQLiteStore) Initialize(ctx context.Context) error

Initialize brings the database schema up to date by applying the embedded migration scripts through the configured Migrator. The default native Migrator uses database/sql only; a goway-backed Migrator can be supplied with WithMigrator.

func (SQLiteStore) MaterializeDeliveries 

func (s SQLiteStore) MaterializeDeliveries(ctx context.Context, key ConsumerKey) (int64, error)

MaterializeDeliveries inserts one pending delivery row for every message the consumer covers but has no row for yet, scanning from the consumer's durable frontier. The frontier is read from the consumer row inside the statement, so the scan floor is always the durable one; a missing consumer row (for example, one reaped by broadcast expiry) yields a NULL frontier, whose comparison matches no rows. The composite primary key makes the insert idempotent, so concurrent materializations converge.

func (SQLiteStore) RegisterConsumer 

func (s SQLiteStore) RegisterConsumer(ctx context.Context, consumer Consumer) error

RegisterConsumer records the durable consumer registration. Registering an existing consumer again keeps the stored boundary and frontier, so a durable group resumes where it left off, and refreshes only the heartbeat: a node re-attaching under a stable identity must rejoin with fresh liveness, not with the staleness it accumulated while it was down.

func (SQLiteStore) RegisterListenerMode 

func (s SQLiteStore) RegisterListenerMode(
	ctx context.Context, id eventbus.ListenerID, mode DeliveryMode, ordering Ordering,
) (DeliveryMode, Ordering, error)

RegisterListenerMode records the delivery mode and the ordering of the listener with first-registration-wins semantics and returns the pair that won, so the caller can detect a conflicting registration made by another node.

func (SQLiteStore) RemoveListener 

func (s SQLiteStore) RemoveListener(ctx context.Context, id eventbus.ListenerID) error

RemoveListener deletes every consumer registration and the delivery-mode row of the listener, unpinning its messages from retention.

func (SQLiteStore) ResubmitDelivery 

func (s SQLiteStore) ResubmitDelivery(ctx context.Context, key DeliveryKey) (bool, error)

ResubmitDelivery gives a failed or exhausted delivery a fresh attempt budget, clearing the backoff of the failed attempts so the fresh budget starts immediately. It reports false when the delivery is not in a resubmittable state.

type Serializer 

type Serializer interface {
	Serialize(event any) (eventType string, payload string, err error)
	Deserialize(eventType string, payload string) (event any, err error)
}

Serializer converts events to and from their persistent representation. The interface is structurally identical to the outbox Serializer, so one serializer instance can serve both modules when an application uses them side by side. The implementation is deliberately duplicated instead of imported: the modules are versioned independently and neither may depend on the other for its core contract.

type Status 

type Status string

Status models the lifecycle of one delivery. 

const (
	// StatusPending marks a delivery that awaits its first claim.
	StatusPending Status = "PENDING"

	// StatusProcessing marks a delivery claimed by a dispatcher.
	StatusProcessing Status = "PROCESSING"

	// StatusCompleted marks a delivery whose listener succeeded.
	StatusCompleted Status = "COMPLETED"

	// StatusFailed marks a delivery whose listener failed and that awaits
	// its next attempt after a backoff delay.
	StatusFailed Status = "FAILED"

	// StatusExhausted marks a delivery that consumed every configured
	// attempt. It is terminal until Bridge.Resubmit gives it a fresh budget.
	StatusExhausted Status = "EXHAUSTED"
)

type Store 

type Store interface {
	Initialize(ctx context.Context) error

	CreateMessage(ctx context.Context, querier Querier, message Message) error
	CreateDeliveries(ctx context.Context, querier Querier, keys []DeliveryKey) error

	RegisterListenerMode(ctx context.Context, id eventbus.ListenerID,
		mode DeliveryMode, ordering Ordering) (DeliveryMode, Ordering, error)
	RegisterConsumer(ctx context.Context, consumer Consumer) error
	Heartbeat(ctx context.Context, key ConsumerKey, at time.Time) (bool, error)
	RemoveListener(ctx context.Context, id eventbus.ListenerID) error

	MaterializeDeliveries(ctx context.Context, key ConsumerKey) (int64, error)
	AdvanceFrontier(ctx context.Context, key ConsumerKey, frontier time.Time) error

	// FindDueDeliveries returns the claimable deliveries of the consumer,
	// oldest publication first. For a FIFO consumer only the head of the
	// queue is claimable, and only below the materialization frontier, where
	// the publication order is known to be complete; the total order is
	// (publication_date, message_id), deterministic across the cluster.
	FindDueDeliveries(ctx context.Context, key ConsumerKey, ordering Ordering,
		now time.Time, leaseCutoff time.Time, limit int) ([]DueDelivery, error)
	FindExhaustedDeliveries(ctx context.Context, limit int) ([]DueDelivery, error)

	// ClaimDelivery atomically transitions a due delivery into processing
	// under the token, counting the attempt. The attempts value the claimant
	// observed is re-checked by the update, so it doubles as a fencing
	// generation: a claim based on a stale read affects zero rows.
	ClaimDelivery(ctx context.Context, key DeliveryKey, token string,
		now time.Time, leaseCutoff time.Time, attempts int) (bool, error)

	// ClaimDeliveryInOrder claims like ClaimDelivery but re-verifies inside
	// the update that no earlier incomplete delivery of the same consumer
	// exists, so a claim races neither a resubmitted predecessor nor another
	// node claiming ahead; publicationDate is the position of the claimed
	// message in the total order.
	ClaimDeliveryInOrder(ctx context.Context, key DeliveryKey, token string,
		now time.Time, leaseCutoff time.Time, attempts int, publicationDate time.Time) (bool, error)
	CompleteDelivery(ctx context.Context, key DeliveryKey, token string, completionDate time.Time) (bool, error)

	// FailDelivery settles a failed delivery under the token fence; attempts
	// is the attempt count this dispatcher's claim recorded, from which the
	// exhaustion decision is computed against maximumAttempts.
	FailDelivery(ctx context.Context, key DeliveryKey, token string, cause string,
		nextAttemptDate time.Time, attempts int, maximumAttempts int) (bool, error)
	ResubmitDelivery(ctx context.Context, key DeliveryKey) (bool, error)

	ExpireBroadcastConsumers(ctx context.Context, cutoff time.Time) (int64, error)
	DeleteSettledMessages(ctx context.Context, olderThan time.Time) (int64, error)
	DeleteMessagesOlderThan(ctx context.Context, cutoff time.Time) (int64, error)
	DeleteOrphanDeliveries(ctx context.Context) (int64, error)
}

Store defines the outbound port for persisting messages, consumer registrations, and deliveries.

CreateMessage and CreateDeliveries receive the querier of the caller, so the rows are written inside the same transaction as the business change that produced the event. Every other method runs on its own connection, because claims, settlements, and maintenance must be settled independently of any business transaction.

ClaimDelivery reports whether the caller obtained the delivery: it returns false when another dispatcher already holds or settled it, which is the arbitration that keeps competing consumption at exactly one node. The settlement methods are fenced by the claim token, so a dispatcher whose claim lease expired and was stolen affects zero rows.

Source Files

View all Source files

Directories

Path Synopsis

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.