Documentation
¶
Overview ¶
Package dispatch is the dispatcher's core loop: claim a batch, publish it, write the outcome back.
One pipeline runs per stream. Processing every stream in one batch and publishing serially would mean a broker that is down does not merely delay its own messages: it holds up everything claimed behind them.
Index ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type DeadLetter ¶
type DeadLetter struct {
// contains filtered or unexported fields
}
DeadLetter forwards messages that stopped being retried to a configured destination, so a consumer can react to them instead of an operator having to notice a row in a table.
The row stays in the outbox either way. The dead-letter topic is a signal, not the record: a failure to publish it must never make the record itself harder to find, which is why nothing here can change a message's status.
It runs as an asynchronous subscriber on the bus. That is the difference between it and the metrics observer, which is synchronous: a counter must not be dropped under backpressure, whereas forwarding involves a database read and a broker round trip and has no business happening inside the publisher's own loop.
func NewDeadLetter ¶
func NewDeadLetter( fetch Fetcher, router Router, emitter DeadLetterEmitter, cfg config.Config, log *slog.Logger, ) *DeadLetter
NewDeadLetter builds the forwarder.
type DeadLetterEmitter ¶
type DeadLetterEmitter interface {
DeadLetter(ctx context.Context, ev events.DeadLetter)
}
DeadLetterEmitter reports each forwarding attempt.
type Emitter ¶
type Emitter interface {
Iteration(ctx context.Context, ev events.Iteration)
Breaker(ctx context.Context, ev events.Breaker)
}
Emitter publishes a domain event. It exists so a pipeline can be tested without a bus.
type Housekeeper ¶
type Housekeeper interface {
Reclaim(ctx context.Context, limit int) ([]store.Reclaimed, error)
Stats(ctx context.Context) (store.Stats, error)
Purge(ctx context.Context, retention time.Duration, limit int) (int64, error)
TryLock(ctx context.Context, class, key int32) (release func(), ok bool, err error)
// The partition side. Only reached on a range-partitioned table, which is
// something the janitor asks about rather than being told.
IsPartitioned(ctx context.Context) (bool, error)
EnsurePartitions(ctx context.Context, ahead int) ([]string, error)
DropExpiredPartitions(ctx context.Context, retention time.Duration) ([]string, error)
DefaultPartitionRows(ctx context.Context) (int64, error)
}
Housekeeper is the database side of the janitor.
type Janitor ¶
type Janitor struct {
// contains filtered or unexported fields
}
Janitor runs the periodic work that must happen exactly once across the whole deployment rather than once per replica: returning expired leases, sampling the gauges, and removing delivered rows.
Each cycle takes a PostgreSQL advisory lock and skips itself if another replica holds it. Every instance tries, one wins, the rest move on without waiting — the right shape for periodic work nobody is blocked on.
func NewJanitor ¶
func NewJanitor( st Housekeeper, emitter JanitorEmitter, cfg config.Config, log *slog.Logger, ) *Janitor
NewJanitor builds a janitor. reclaimLimit bounds how many leases one cycle returns, so a mass failure does not become one enormous statement.
func (*Janitor) ReclaimExpired ¶
ReclaimExpired returns leases whose owner never released them.
func (*Janitor) SampleStats ¶
SampleStats refreshes the backlog gauges.
Unlike the other two cycles this runs on every replica, without the advisory lock. A gauge that only the lock holder refreshes leaves every other replica exporting its zero value forever, so a dashboard reading one series — or averaging them — reports an empty backlog while the queue grows. The query is four index-only counts on a slow tick; paying for it per replica is cheaper than a metric that is wrong on all but one of them.
type JanitorEmitter ¶
type JanitorEmitter interface {
Reclaimed(ctx context.Context, ev events.Reclaimed)
Stats(ctx context.Context, ev events.Stats)
Retention(ctx context.Context, ev events.Retention)
Partitions(ctx context.Context, ev events.Partitions)
}
JanitorEmitter publishes what the janitor observes.
type Listener ¶
type Listener struct {
// contains filtered or unexported fields
}
Listener turns PostgreSQL notifications into pipeline wakeups, so a message is picked up milliseconds after it is written instead of on the next poll tick.
It is a latency optimisation and nothing more. NOTIFY is best-effort and is lost when the listening connection drops, so the poll loop stays on as reconciliation. Losing a notification costs one poll interval; it never costs a message.
func NewListener ¶
func NewListener(pool *pgxpool.Pool, cfg config.DispatchConfig, pipelines []Waker, log *slog.Logger) *Listener
NewListener wires a listener to the pipelines it can wake.
type Option ¶ added in v1.5.0
type Option func(*Pipeline)
Option adjusts a pipeline at construction. Options exist so that wiring a dependency the dispatcher can run without — tracing is the only one so far — does not become a parameter every caller has to pass nil for.
func WithTracer ¶ added in v1.5.0
WithTracer records a span per published message. Without it the pipeline holds a tracer that does nothing, which is also what a deployment with no collector configured gets.
type Pipeline ¶
type Pipeline struct {
// contains filtered or unexported fields
}
Pipeline dispatches one stream.
func New ¶
func New( stream string, st Store, router Router, emitter Emitter, cfg config.Config, log *slog.Logger, opts ...Option, ) *Pipeline
New builds a pipeline for one stream.
func (*Pipeline) Run ¶
Run drives the pipeline until ctx is canceled.
The loop is adaptive: a full batch means there is a backlog, so the next iteration starts at once rather than sleeping out the poll interval. Always waiting would cap throughput at batch size divided by poll interval — a fixed number of messages per tick, whatever the hardware underneath.
func (*Pipeline) RunOnce ¶
RunOnce performs one claim-publish-write-back cycle and reports what it did.
It always tries: holding claims back is the run loop's decision, not this one's, which is what lets a caller — a test, or a trial after a pause — ask the broker directly.
type Result ¶ added in v1.3.0
type Result struct {
// Claimed is the batch size. A batch that keeps arriving full means the
// dispatcher is not keeping up.
Claimed int
// Delivered counts messages a broker accepted.
Delivered int
// Deferred counts messages held back because the broker could not be
// reached.
Deferred int
}
Result summarises one cycle in the terms the run loop needs: whether more work is waiting, and whether the broker answered at all.
type Router ¶
type Router interface {
Publish(ctx context.Context, stream string, msgs []core.Message) []error
DriverFor(stream string) (string, bool)
}
Router publishes a batch belonging to one stream.
type Store ¶
type Store interface {
Claim(ctx context.Context, stream string, limit int, lease core.Lease) ([]core.Message, error)
Ack(ctx context.Context, ids []string, token string) (store.AckResult, error)
Nack(ctx context.Context, outcomes []core.Outcome, token string, limits core.RetryLimits) (store.NackResult, error)
ReleaseLease(ctx context.Context, ids []string, token string) (int, error)
}
Store is the database side of a pipeline.
Source Files