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. It reports per-lane outcome counters 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.
A 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, tx 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.
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) 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 — 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