Documentation
¶
Overview ¶
Package outbox is the transactional-outbox relay: it leases due rows from an interfaces.OutboxStore, 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. It reports per-lane outcome counters and the lane's depth and lag. The store and sink are injected (and decorated) by the caller.
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, store interfaces.OutboxStore, sink interfaces.OutboxSink, logger interfaces.Logger, metrics interfaces.Metrics, tracer interfaces.Tracer, opts ...Option, ) *Relay
NewRelay builds a relay over store and sink. logger, metrics and tracer may be nil. An invalid cfg or a nil store or sink is a wiring bug and panics; build cfg with LoadRelayConfig (or call cfg.Validate first) to get a coded error instead.
func (*Relay) RunOnce ¶
RunOnce claims one batch, delivers it and finalizes every row, then refreshes the lane gauges. The claim is called exactly once (it is side-effecting, so it is never retried here). A finalize failure does not stop the other rows — the row's lease expires and it is re-claimed — and the first such error is returned. 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"`
}
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