subscriptions

package
v1.0.0-beta.3 Latest Latest
Warning

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

 Go to latest Published: Jul 25, 2026 License: MIT Imports: 13 Imported by: 0

Documentation

Overview

Package subscriptions provides an opt-in runtime for running event handlers as crash-safe background subscribers. A Subscriber treats the event store's global ordered feed (EventStore.ReadAfter) as a durable queue: it remembers exactly one number — its checkpoint — so crash recovery and normal startup are the same code path.

Synchronous in-commit dispatch via the UnitOfWork's EventDispatcher is unaffected; this package is for consumers that need resumable, transactional background processing.

Index

Constants

View Source 
const (
	// DefaultBatchSize is the number of events read per cycle when
	// WithBatchSize is not supplied.
	DefaultBatchSize = 100

	// DefaultPollInterval is how long an idle subscriber waits before
	// checking the feed again when WithPollInterval is not supplied.
	DefaultPollInterval = time.Second

	// DefaultMaxRetries is how many times a failing handler is retried per
	// event (after the initial attempt) before the event is parked, when a
	// ParkingLot is configured and WithMaxRetries is not supplied.
	DefaultMaxRetries = 5

	// DefaultRetryBackoff is the first retry delay; it doubles per attempt.
	DefaultRetryBackoff = 100 * time.Millisecond

	// DefaultMaxRetryBackoff caps the doubling retry delay.
	DefaultMaxRetryBackoff = 5 * time.Second
)
View Source 
const DefaultListenerReconnectDelay = 5 * time.Second

DefaultListenerReconnectDelay is how long PostgresListener waits before re-establishing a failed LISTEN connection.

Variables

View Source 
var ErrEventNotParked = errors.New("event is not parked")

ErrEventNotParked is returned by Replay when the event is not parked for the subscriber.

Functions

func TxFromContext 

func TxFromContext(ctx context.Context) *gorm.DB

TxFromContext returns the batch transaction attached to a handler's context by a GormCheckpointStore batch, or nil when the handler is not running inside a database-backed batch. Handlers that write projections to the same database should write through this transaction: those writes then commit atomically with the checkpoint advance (exactly-once), instead of relying on at-least-once redelivery.

Types

type Batch 

type Batch interface {
	// Position is the committed checkpoint at acquisition time; the cycle
	// processes events after it.
	Position() int64

	// HandlerContext derives the context handlers run with. Database-backed
	// implementations attach the batch transaction so handlers can join it
	// via TxFromContext.
	HandlerContext(ctx context.Context) context.Context

	// Commit advances the checkpoint to position and commits the batch
	// (including any handler writes made through the batch transaction).
	Commit(ctx context.Context, position int64) error

	// Rollback abandons the cycle: the checkpoint stays at Position() and
	// any handler writes made through the batch transaction are discarded.
	Rollback() error

	// Savepoint marks a rollback point inside the batch and
	// RollbackToSavepoint discards everything written through the batch
	// transaction after the mark. The subscriber brackets each handler
	// attempt with one so a failed attempt's partial writes are discarded
	// without abandoning the whole batch. Implementations without a
	// transaction treat both as no-ops.
	Savepoint(ctx context.Context, name string) error
	RollbackToSavepoint(ctx context.Context, name string) error
}

Batch is one acquired processing cycle for a subscriber. Exactly one of Commit or Rollback must be called.

type CheckpointStore 

type CheckpointStore interface {
	// Acquire begins a processing cycle for the named subscriber, creating
	// the checkpoint at position 0 if it does not exist. It returns
	// acquired=false (and no Batch) when another process currently holds the
	// subscriber's checkpoint; the caller should skip the cycle.
	Acquire(ctx context.Context, subscriber string) (batch Batch, acquired bool, err error)

	// Position returns the subscriber's committed checkpoint (0 if the
	// subscriber is unknown).
	Position(ctx context.Context, subscriber string) (int64, error)

	// Reset sets the subscriber's committed checkpoint, creating it if
	// needed. Resetting to 0 makes the subscriber replay all history,
	// incrementally and resumably, on its next cycles. Implementations
	// serialize Reset against in-flight batches: either Reset waits for the
	// batch (or fails), or the batch's Commit detects the moved checkpoint
	// and aborts — a reset is never silently overwritten.
	Reset(ctx context.Context, subscriber string, position int64) error
}

CheckpointStore persists the position each named subscriber has processed up to. Implementations coordinate concurrent access: at most one Batch per subscriber name is active at a time.

type GormCheckpointModel 

type GormCheckpointModel struct {
	Subscriber string    `gorm:"primaryKey;column:subscriber"`
	Position   int64     `gorm:"column:position;not null"`
	UpdatedAt  time.Time `gorm:"column:updated_at"`
}

GormCheckpointModel is the GORM model for subscriber checkpoints. The table is owned and auto-migrated by pericarp.

func (GormCheckpointModel) TableName 

func (GormCheckpointModel) TableName() string

TableName returns the table name for the checkpoint model.

type GormCheckpointStore 

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

GormCheckpointStore is a database-backed CheckpointStore. Each batch runs inside one database transaction: handlers that write to the same database through the batch transaction (TxFromContext) commit atomically with the checkpoint advance — exactly-once processing for same-database handlers.

On Postgres the checkpoint row is locked with FOR UPDATE SKIP LOCKED, so N processes running the same subscriber coordinate as active/passive replicas: one wins each cycle, the others skip. SQLite relies on its serialized writers (single process).

func NewGormCheckpointStore 

func NewGormCheckpointStore(db *gorm.DB) (*GormCheckpointStore, error)

NewGormCheckpointStore creates a checkpoint store and auto-migrates the subscriber_checkpoints table. Like the event feed itself, it supports Postgres and SQLite: replica coordination relies on row locking (Postgres) or serialized writers (SQLite), and other engines would silently double-process events.

func (*GormCheckpointStore) Acquire 

func (g *GormCheckpointStore) Acquire(ctx context.Context, subscriber string) (Batch, bool, error)

Acquire begins a batch transaction holding the subscriber's checkpoint row. On Postgres, acquired is false when another process holds the row.

func (*GormCheckpointStore) Position 

func (g *GormCheckpointStore) Position(ctx context.Context, subscriber string) (int64, error)

Position returns the subscriber's committed checkpoint (0 if unknown).

func (*GormCheckpointStore) Reset 

func (g *GormCheckpointStore) Reset(ctx context.Context, subscriber string, position int64) error

Reset sets the subscriber's checkpoint, creating the row if needed. On Postgres it blocks until any in-flight batch for the subscriber finishes (the batch transaction holds the row). On SQLite a concurrent in-flight batch is not blocked; instead its Commit detects that the checkpoint moved and aborts, so the reset always wins and processing resumes from the reset position.

type GormParkedEventModel 

type GormParkedEventModel struct {
	Subscriber string    `gorm:"primaryKey;column:subscriber"`
	EventID    string    `gorm:"primaryKey;column:event_id"`
	EventType  string    `gorm:"column:event_type"`
	Position   int64     `gorm:"column:position"`
	Error      string    `gorm:"column:error"`
	Attempts   int       `gorm:"column:attempts"`
	ParkedAt   time.Time `gorm:"column:parked_at"`
}

GormParkedEventModel is the GORM model for parked events. The table is owned and auto-migrated by pericarp.

func (GormParkedEventModel) TableName 

func (GormParkedEventModel) TableName() string

TableName returns the table name for the parked event model.

type GormParkingLot 

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

GormParkingLot is a database-backed ParkingLot. When parking happens inside a subscriber batch, the row is written through the batch transaction so the parking and the checkpoint advance commit atomically — a crash between the two cannot lose a poison event or park it twice.

func NewGormParkingLot 

func NewGormParkingLot(db *gorm.DB) (*GormParkingLot, error)

NewGormParkingLot creates a parking lot and auto-migrates the parked_events table. Construct it with the same *gorm.DB as the GormCheckpointStore: that is what lets Park join the batch transaction and commit atomically with the checkpoint advance. A lot built on a different handle still works, but parks non-transactionally through its own connection.

func (*GormParkingLot) List 

func (g *GormParkingLot) List(ctx context.Context, subscriber string) ([]ParkedEvent, error)

List returns the subscriber's parked events ordered by position.

func (*GormParkingLot) Park 

func (g *GormParkingLot) Park(ctx context.Context, parked ParkedEvent) error

Park records a poison event, writing through the batch transaction when ctx carries one that belongs to this lot's database. A batch transaction from a different *gorm.DB is ignored — writing parked_events through a foreign connection could land the row in a different database, where List/Replay would never find it. Re-parking an already-parked event updates its error, attempt count, and timestamp (the event may be reprocessed after a checkpoint reset).

func (*GormParkingLot) Replay 

func (g *GormParkingLot) Replay(ctx context.Context, subscriber, eventID string, event domain.EventEnvelope[any], handler Handler) error

Replay re-runs the handler and clears the parked row in one transaction. The handler receives the transaction via TxFromContext, so a failed replay rolls back both the handler's writes and the row deletion. Concurrent replays of the same event execute the handler at most once: the row is locked on Postgres, and the delete is verified so a transaction that lost the race rolls back its duplicate handler writes instead of committing.

type Handler 

type Handler func(ctx context.Context, event domain.EventEnvelope[any]) error

Handler processes a single event from the feed. EventDispatcher.Dispatch satisfies this signature, so a dispatcher with registered pattern handlers can be wired directly as a Subscriber's handler.

When the subscriber's CheckpointStore is database-backed, the context passed to the handler carries the batch's transaction (see TxFromContext): writes made through it commit atomically with the checkpoint advance, giving exactly-once processing for same-database handlers. Handlers with side effects outside that transaction must tolerate at-least-once delivery.

type InProcessNotifier 

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

InProcessNotifier fans a commit signal out to subscribers in the same process — the SQLite / single-process counterpart of Postgres LISTEN/NOTIFY. Wire it by wrapping the event store in a NotifyingEventStore and passing each subscriber a Subscribe() channel via WithWakeSignal.

Signals are best-effort wake-ups, never load-bearing: sends never block (a subscriber that is busy keeps at most one pending signal), and a missed signal costs at most one poll interval.

func NewInProcessNotifier 

func NewInProcessNotifier() *InProcessNotifier

NewInProcessNotifier creates a notifier with no subscribers.

func (*InProcessNotifier) Notify 

func (n *InProcessNotifier) Notify()

Notify wakes all subscribers without blocking.

func (*InProcessNotifier) Subscribe 

func (n *InProcessNotifier) Subscribe() <-chan struct{}

Subscribe returns a channel that receives after each Notify. Pass it to a Subscriber via WithWakeSignal.

type MemoryCheckpointStore 

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

MemoryCheckpointStore is an in-memory CheckpointStore for tests and single-process development setups. It offers no transactional coupling with handler writes — handlers get at-least-once delivery.

func NewMemoryCheckpointStore 

func NewMemoryCheckpointStore() *MemoryCheckpointStore

NewMemoryCheckpointStore creates an empty in-memory checkpoint store.

func (*MemoryCheckpointStore) Acquire 

func (m *MemoryCheckpointStore) Acquire(ctx context.Context, subscriber string) (Batch, bool, error)

Acquire begins a processing cycle; acquired is false while another batch for the same subscriber is active.

func (*MemoryCheckpointStore) Position 

func (m *MemoryCheckpointStore) Position(ctx context.Context, subscriber string) (int64, error)

Position returns the committed checkpoint for the subscriber.

func (*MemoryCheckpointStore) Reset 

func (m *MemoryCheckpointStore) Reset(ctx context.Context, subscriber string, position int64) error

Reset sets the committed checkpoint. It fails while a batch is in flight (an in-memory caller can simply stop the subscriber first).

type MemoryParkingLot 

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

MemoryParkingLot is an in-memory ParkingLot for tests and single-process development setups. Parking is not transactional with checkpoint advances, and parked records do not survive a restart — pairing it with a durable checkpoint store means a poison event whose park record is lost on restart has already been skipped by the durable checkpoint, permanently.

func NewMemoryParkingLot 

func NewMemoryParkingLot() *MemoryParkingLot

NewMemoryParkingLot creates an empty in-memory parking lot.

func (*MemoryParkingLot) List 

func (m *MemoryParkingLot) List(ctx context.Context, subscriber string) ([]ParkedEvent, error)

List returns the subscriber's parked events ordered by position.

func (*MemoryParkingLot) Park 

func (m *MemoryParkingLot) Park(ctx context.Context, parked ParkedEvent) error

Park records a poison event.

func (*MemoryParkingLot) Replay 

func (m *MemoryParkingLot) Replay(ctx context.Context, subscriber, eventID string, event domain.EventEnvelope[any], handler Handler) error

Replay re-runs the handler and clears the parked entry on success. The handler runs at most once per entry: concurrent replays of the same event conflict instead of double-executing, and an entry re-parked while the replay was running (a fresh failure) is preserved rather than cleared.

type NotifyingEventStore 

type NotifyingEventStore struct {
	domain.EventStore
	// contains filtered or unexported fields
}

NotifyingEventStore decorates an EventStore so every successful Append fires a callback — typically InProcessNotifier.Notify — after the events are durably stored. Wrap the store handed to the UnitOfWork (or used directly) so background subscribers wake on new commits instead of waiting out their poll interval.

Postgres deployments don't need this: the GORM event store NOTIFYs on commit by itself (see infrastructure.PostgresNotifyChannel) and PostgresListener picks it up across processes.

func NewNotifyingEventStore 

func NewNotifyingEventStore(store domain.EventStore, notify func()) *NotifyingEventStore

NewNotifyingEventStore wraps store; notify runs after every successful Append.

func (*NotifyingEventStore) Append 

func (s *NotifyingEventStore) Append(ctx context.Context, aggregateID string, expectedVersion int, events ...domain.EventEnvelope[any]) error

Append delegates to the wrapped store and signals on success.

type ParkedEvent 

type ParkedEvent struct {
	Subscriber string    `json:"subscriber"`
	EventID    string    `json:"event_id"`
	EventType  string    `json:"event_type"`
	Position   int64     `json:"position"`
	Error      string    `json:"error"`
	Attempts   int       `json:"attempts"`
	ParkedAt   time.Time `json:"parked_at"`
}

ParkedEvent records an event a subscriber could not process after exhausting its retries.

type ParkingLot 

type ParkingLot interface {
	// Park records a poison event. When ctx carries a batch transaction
	// (TxFromContext), the row is written through it so the parking and the
	// checkpoint advance past the event commit atomically.
	Park(ctx context.Context, parked ParkedEvent) error

	// List returns the subscriber's parked events ordered by position.
	List(ctx context.Context, subscriber string) ([]ParkedEvent, error)

	// Replay re-runs handler for a parked event and clears the row on
	// success. Database-backed implementations run both in one transaction
	// (with the transaction exposed to the handler via TxFromContext), so a
	// failed replay leaves the row — and any handler writes — untouched.
	// Returns ErrEventNotParked when the event is not parked.
	Replay(ctx context.Context, subscriber, eventID string, event domain.EventEnvelope[any], handler Handler) error
}

ParkingLot stores poison events so one unprocessable event never blocks the events behind it: the subscriber parks it and advances the checkpoint past it. CLI or HTTP surfaces over this API are the consumer's job.

type PostgresListener 

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

PostgresListener holds a dedicated Postgres connection LISTENing for the event store's commit notifications and fans them out as subscriber wake signals: every Subscribe() channel receives each notification, so one listener serves any number of subscribers in the process. Notifications are never load-bearing: while the connection is down — or if a NOTIFY is missed entirely — subscribers still make progress through their poll interval.

func NewPostgresListener 

func NewPostgresListener(dsn string, opts ...PostgresListenerOption) (*PostgresListener, error)

NewPostgresListener creates a listener for the given Postgres DSN. Call Run to start it and pass each subscriber its own Subscribe() channel via WithWakeSignal.

func (*PostgresListener) Run 

func (l *PostgresListener) Run(ctx context.Context) error

Run maintains the LISTEN connection until ctx is cancelled, reconnecting with a delay on any failure. It returns nil on cancellation; connection errors are logged, never fatal — polling subscribers keep making progress while the listener is down.

func (*PostgresListener) Subscribe 

func (l *PostgresListener) Subscribe() <-chan struct{}

Subscribe returns a channel that receives after each notification. Every subscriber needs its own channel (a Go channel receive is point-to-point, not broadcast — sharing one channel would wake only one subscriber per notification). Channels are buffered and sends never block.

type PostgresListenerOption 

type PostgresListenerOption func(*PostgresListener)

PostgresListenerOption configures a PostgresListener.

func WithListenerChannel 

func WithListenerChannel(channel string) PostgresListenerOption

WithListenerChannel overrides the NOTIFY channel (default infrastructure.PostgresNotifyChannel, "pericarp_events").

func WithListenerLogger 

func WithListenerLogger(logger *slog.Logger) PostgresListenerOption

WithListenerLogger sets the logger for connection failures (default slog.Default(); nil falls back to the default).

func WithListenerReconnectDelay 

func WithListenerReconnectDelay(d time.Duration) PostgresListenerOption

WithListenerReconnectDelay sets the wait between reconnect attempts (default DefaultListenerReconnectDelay).

type Subscriber 

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

Subscriber runs a Handler as a crash-safe background worker over the event store's global ordered feed. Each cycle acquires the subscriber's checkpoint, reads a batch of events past it via EventStore.ReadAfter, invokes the handler for each event, and advances the checkpoint. Because the checkpoint is the only state, crash recovery and normal startup are the same code path.

A handler error abandons the whole batch — the checkpoint stays put, any handler writes made through the batch transaction roll back, and the batch is retried after the poll interval. Events are therefore delivered at-least-once; handlers writing through TxFromContext get exactly-once.

func NewSubscriber 

func NewSubscriber(name string, events domain.EventStore, checkpoints CheckpointStore, handler Handler, opts ...SubscriberOption) (*Subscriber, error)

NewSubscriber creates a subscriber. name identifies the checkpoint — processes using the same name share one position. handler is invoked for every event in feed order; EventDispatcher.Dispatch satisfies the Handler signature for pattern-based routing.

func (*Subscriber) Lag 

func (s *Subscriber) Lag(ctx context.Context) (int64, error)

Lag returns how far the subscriber's committed checkpoint trails the feed head (0 when caught up). Consumers log or meter it; pericarp does not.

A checkpoint ahead of the head is reported as an error rather than clamped to 0: it means the events table was truncated or the checkpoint was reset past the head, and the subscriber would otherwise look caught up while silently skipping everything until positions catch back up.

func (*Subscriber) ListParked 

func (s *Subscriber) ListParked(ctx context.Context) ([]ParkedEvent, error)

ListParked returns this subscriber's parked events ordered by position. Requires a ParkingLot (WithParkingLot).

func (*Subscriber) Name 

func (s *Subscriber) Name() string

Name returns the subscriber's checkpoint name.

func (*Subscriber) ReplayParked 

func (s *Subscriber) ReplayParked(ctx context.Context, eventID string) error

ReplayParked re-runs the handler for one parked event and clears it on success (a failed replay leaves the event parked). The event is loaded fresh from the event store; database-backed parking lots run the handler and the row deletion in one transaction, exposed via TxFromContext exactly like a live batch. Requires a ParkingLot (WithParkingLot).

func (*Subscriber) ResetCheckpoint 

func (s *Subscriber) ResetCheckpoint(ctx context.Context, position int64) error

ResetCheckpoint sets the subscriber's checkpoint. Resetting to 0 replays all history — incrementally and resumably, since replay uses the same batch/checkpoint cycle as live processing. It is safe while the subscriber runs: the checkpoint store serializes with in-flight batches (see CheckpointStore.Reset), so the reset is never silently overwritten.

func (*Subscriber) Run 

func (s *Subscriber) Run(ctx context.Context) error

Run processes the feed until ctx is cancelled. A batch that is already in flight when cancellation arrives is drained — handlers finish and the checkpoint advances — before Run returns nil.

Transient errors (database hiccups, handler failures) are logged and retried after the poll interval; they never stop the subscriber. The only error Run returns is fatal misconfiguration: an event store without a global ordered feed (ErrGlobalOrderingNotSupported).

type SubscriberOption 

type SubscriberOption func(*Subscriber)

SubscriberOption configures a Subscriber.

func WithBatchSize 

func WithBatchSize(n int) SubscriberOption

WithBatchSize sets how many events are read per cycle (default DefaultBatchSize).

func WithLogger 

func WithLogger(logger *slog.Logger) SubscriberOption

WithLogger sets the logger for batch failures and lifecycle events. The default is slog.Default() — a permanently failing subscriber must be visible somewhere out of the box.

func WithMaxRetries 

func WithMaxRetries(n int) SubscriberOption

WithMaxRetries sets how many times a failing handler is retried per event (after the initial attempt) before the event is parked (default DefaultMaxRetries). Only meaningful with WithParkingLot.

func WithParkingLot 

func WithParkingLot(lot ParkingLot) SubscriberOption

WithParkingLot enables poison-event handling: a handler that keeps failing on one event has the event parked (with its error) after bounded retries, and the checkpoint advances past it so the events behind it keep flowing. Without a parking lot, a failing event abandons the batch and is retried every poll interval forever.

Failed attempts' writes are discarded only when the batch has a real transaction (GormCheckpointStore) and the handler writes through it; with MemoryCheckpointStore or out-of-transaction side effects, each retry re-applies whatever the handler did before failing.

func WithPollInterval 

func WithPollInterval(d time.Duration) SubscriberOption

WithPollInterval sets how long an idle subscriber waits before checking the feed again (default DefaultPollInterval). It is also the retry delay after a failed batch.

func WithRetryBackoff 

func WithRetryBackoff(initial, maximum time.Duration) SubscriberOption

WithRetryBackoff sets the first retry delay and its cap; the delay doubles per attempt up to the cap (defaults DefaultRetryBackoff and DefaultMaxRetryBackoff).

func WithWakeSignal 

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

WithWakeSignal lets the subscriber wake on new commits instead of waiting out the poll interval: pass InProcessNotifier.Subscribe() (single-process / SQLite) or PostgresListener.Subscribe() (cross-process via LISTEN/NOTIFY). Each subscriber needs its own channel — channel receives are point-to-point, so a shared channel would wake only one subscriber per signal. Polling continues as the fallback, so notifications are never load-bearing. A closed channel is detected and the subscriber falls back to pure polling.

Source Files

View all Source files

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.