Documentation
¶
Overview ¶
Package outbox is the transactional-outbox relay: it leases due rows from an interfaces.OutboxRepository, delivers them through an interfaces.OutboxSink in bounded-concurrency chunks, and finalizes each row by its own outcome — sent, retried with capped exponential backoff and full jitter, or parked once its attempts are exhausted. With a drain budget a run keeps claiming while claims come back full, so a burst drains in one run instead of one batch per tick. It reports per-lane outcome counters, passes per run, and the lane's depth and lag. The repository and sink are injected (and decorated) by the caller: build the repository with repos/outbox.New over the store.
Each pass is a saga, so the relay is a workflow: the claim commits in its own transaction, delivery runs with no transaction open (a remote call never holds one), and each row's finalize commits in its own transaction, all opened through the injected transaction manager.
Index ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Option ¶
type Option func(*Relay)
Option customizes a Relay beyond its config (the deterministic seams tests use).
type Relay ¶
type Relay struct {
// contains filtered or unexported fields
}
Relay moves outbox rows from a store to a sink. It is safe for concurrent RunOnce calls (the store's claim keeps them disjoint).
func NewRelay ¶
func NewRelay( cfg *RelayConfig, transactionManager interfaces.TransactionManager, store interfaces.OutboxRepository, sink interfaces.OutboxSink, logger interfaces.Logger, metrics interfaces.Metrics, tracer interfaces.Tracer, opts ...Option, ) *Relay
NewRelay builds a relay over store and sink whose writes commit through tx. logger, metrics and tracer may be nil. An invalid cfg or a nil tx, store or sink is a wiring bug and panics; build cfg with LoadRelayConfig (or call cfg.Validate first) to get a coded error instead. cfg must not be nil.
func (*Relay) RunOnce ¶
RunOnce claims a batch, delivers it and finalizes every row; with a DrainBudget it repeats while claims come back full and the budget lasts. It then records the passes and refreshes the lane gauges. Each claim is called once (it is side-effecting, so it is never retried here) and commits before delivery starts; each finalize commits on its own, so one row's failure never rolls back another's. A finalize failure does not stop the other rows of its batch — the row's lease expires and it is re-claimed — but it ends the drain, and the first such error is returned. A failed claim ends the run with its error. A Stats failure is only logged; with metrics off Stats is skipped.
type RelayConfig ¶
type RelayConfig struct {
// Lane is the outbox lane drained.
Lane string `yaml:"lane" mapstructure:"lane"`
// BatchSize is the most rows one RunOnce claims.
BatchSize int `yaml:"batch_size" mapstructure:"batch_size"`
// SendBatch is the most records handed to one Sink.Send call.
SendBatch int `yaml:"send_batch" mapstructure:"send_batch"`
// Concurrency bounds how many Send calls run at once.
Concurrency int `yaml:"concurrency" mapstructure:"concurrency"`
// MaxAttempts is the attempt on which a still-failing record is parked.
MaxAttempts int `yaml:"max_attempts" mapstructure:"max_attempts"`
// BaseBackoff is the backoff unit: a record that failed attempt n waits up to
// BaseBackoff·2^n before its next attempt.
BaseBackoff time.Duration `yaml:"base_backoff" mapstructure:"base_backoff"`
// MaxBackoff caps the backoff before jitter.
MaxBackoff time.Duration `yaml:"max_backoff" mapstructure:"max_backoff"`
// ParkOnPermanent parks a record on its first permanent failure (one
// apperr.IsPermanent reports, e.g. a malformed payload the destination
// rejects) instead of retrying it until MaxAttempts.
ParkOnPermanent bool `yaml:"park_on_permanent" mapstructure:"park_on_permanent"`
// DrainBudget bounds how long one RunOnce keeps claiming: while a claim returns a
// full BatchSize, the run claims again until the budget has elapsed (the batch in
// flight when it runs out still finishes). Zero, the default, claims once per run.
DrainBudget time.Duration `yaml:"drain_budget" mapstructure:"drain_budget"`
// Partitions is how many partitions the lane is split into (the store's
// partitions); zero means 1, today's single relay per lane.
Partitions int `yaml:"partitions" mapstructure:"partitions"`
// Partition is the partition this relay drains, in [0, Partitions). Run one relay
// per partition, each with its own leader election, so N relays drain N
// partitions in parallel.
Partition int `yaml:"partition" mapstructure:"partition"`
}
RelayConfig tunes one relay; one relay drains one lane.
func DefaultRelayConfig ¶
func DefaultRelayConfig(lane string) RelayConfig
DefaultRelayConfig returns a relay config for lane: claim 100 rows, send 10 at a time with 4 concurrent sends, park on the 10th attempt, back off from 1s up to 5m.
func LoadRelayConfig ¶
func LoadRelayConfig( loader interfaces.ConfigLoader, key, lane string, ) (RelayConfig, error)
LoadRelayConfig reads the relay config for lane from loader's key section, overlaid on DefaultRelayConfig(lane), and validates it. An absent section yields the defaults; an unreadable or invalid one is CodeInvalidInput. Load through it so a bad section stops boot with a coded error, never NewRelay's panic.
func (*RelayConfig) Validate ¶
func (c *RelayConfig) Validate() error
Validate reports a config the relay cannot run with as CodeInvalidInput.
Source Files