Documentation
¶
Overview ¶
Package consumer is the journal-as-bus push seam (ADR-0009): one paging reader over queue.Store bounded reads fans every fact out to exact subscribers, in Seq order, at-least-once.
Delivery contract (ADR-0009 D1/D2): exact consumers are NEVER skipped — a handler error pauses that subscriber's drain until its next tick (backpressure), while other subscribers continue. Subscribers own their persistence (the watermarks table); this dispatcher's cursors are per-process and resume from whatever `since` the subscriber passes. Signal consumers (the webui hub) stay on the hub — the dispatcher serves the exact class only.
Index ¶
Constants ¶
const DefaultPageSize = 500
DefaultPageSize bounds one Facts page per subscriber drain.
const DefaultPollInterval = time.Second
DefaultPollInterval is how often the dispatcher pages the journal when no notify hook exists (the v1 wake strategy; a notify-after-commit hook is the documented v2 path and must fire outside the mutation transaction — ADR-0009 D4).
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Config ¶
type Config struct {
// PollInterval is the paging cadence. Default 1s.
PollInterval time.Duration
// PageSize bounds one Facts page per drain. Default 500.
PageSize int
// Logger receives drain diagnostics. Default slog.Default().
Logger *slog.Logger
// Wake is the optional drain-wake channel (a Waker store's Notify):
// a fact-landing commit fires it and Run drains immediately instead
// of waiting out PollInterval. The ticker stays the degraded fallback
// — a nil Wake keeps the pure poll cadence, and journal appends the
// wake does not cover (a non-Waker writer, another process) still
// land on the next tick.
Wake <-chan struct{}
}
Config controls a Dispatcher. Zero values select the defaults.
type Dispatcher ¶
type Dispatcher struct {
// contains filtered or unexported fields
}
Dispatcher fans journal facts out to registered subscribers.
func New ¶
func New(src Source, cfg Config) *Dispatcher
New builds a Dispatcher over src. Call Subscribe before Run.
func (*Dispatcher) Cursor ¶
func (d *Dispatcher) Cursor(name string) (int64, bool)
Cursor reports one subscriber's current delivery cursor (0 = registered but nothing delivered yet). ok=false when the name is unknown.
func (*Dispatcher) Lag ¶
Lag reports per-subscriber lag (HeadSeq − cursor, floored at 0): how far behind the journal head each subscriber sits — the ops surface of ADR-0009 D4.
func (*Dispatcher) Run ¶
func (d *Dispatcher) Run(ctx context.Context) error
Run pages the journal for every subscriber until ctx is cancelled. Each tick drains each subscriber independently in bounded pages: a handler error stops that subscriber's drain (the failing fact redelivers next tick) and the others continue — per-class backpressure, ADR-0009 D2. Lag is logged once a minute per subscriber that is behind the head.
func (*Dispatcher) Subscribe ¶
func (d *Dispatcher) Subscribe(name string, since int64, handler Handler) (unsubscribe func())
Subscribe registers an exact consumer delivering facts with Seq > since. It returns the subscriber's unsubscribe function. Subscribing while Run is active is safe; unsubscribing stops delivery before the next handler invocation — a handler already running when unsubscribe is called may still complete.
Source Files