Something went wrong. Try again.
wip outbox library design
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286package outpost
import ( "context" "errors" "fmt" "math/rand/v2" "sync" "time")
const ( defaultBatchSize = 16 defaultWorkers = 4 defaultPollInterval = time.Second defaultDeliveryTimeout = 30 * time.Second defaultMaxAttempts = 8
// storeOpTimeout bounds the Complete and Fail writes. They deliberately // ignore the caller's cancellation so a delivery that already succeeded is // still recorded during shutdown. storeOpTimeout = 5 * time.Second
// maxDerivedLease rejects configurations whose derived lease would be // unreasonable, nearly always a batch size or worker mistake, and bounds // the multiplication in NewRelay. maxDerivedLease = 30 * time.Minute)
// Relay claims messages from a Store and hands them to a Producer.type Relay struct { store Store producer Producer hooks Hooks
batch int workers int timeout time.Duration maxAttempts int backoff Backoff lease time.Duration}
// RelayOption configures a Relay.type RelayOption func(*Relay)
// WithBatchSize sets how many messages one claim asks for. Default 16.func WithBatchSize(n int) RelayOption { return func(r *Relay) { r.batch = n } }
// WithWorkers sets how many deliveries run concurrently. Default 4.func WithWorkers(n int) RelayOption { return func(r *Relay) { r.workers = n } }
// WithDeliveryTimeout bounds a single delivery. Default 30s.func WithDeliveryTimeout(d time.Duration) RelayOption { return func(r *Relay) { r.timeout = d } }
// WithMaxAttempts sets how many deliveries a message gets before it is// abandoned. Default 8.func WithMaxAttempts(n int) RelayOption { return func(r *Relay) { r.maxAttempts = n } }
// WithBackoff sets the retry delay policy. It is a lower bound, not an exact// schedule: the relay retries no earlier than the computed delay and spreads it// by up to 50% so independent workers do not retry in lockstep. Pass a// deterministic backoff; the relay applies the spread. The default is// ExponentialBackoff(1s, 10m).func WithBackoff(b Backoff) RelayOption { return func(r *Relay) { r.backoff = b } }
// WithHooks sets the observability hooks.func WithHooks(h Hooks) RelayOption { return func(r *Relay) { r.hooks = h } }
// NewRelay validates the configuration and derives the lease duration:// (ceil(batch/workers)+1) * deliveryTimeout, so a claimed batch cannot outlive// its lease even when every delivery runs to its deadline. The arithmetic is// done in int64 and bounded by maxDerivedLease before it is multiplied, so an// absurd batch size fails loudly instead of wrapping into a tiny lease.func NewRelay(store Store, producer Producer, opts ...RelayOption) (*Relay, error) { if store == nil { return nil, errors.New("outpost: nil store") } if producer == nil { return nil, errors.New("outpost: nil producer") } r := &Relay{ store: store, producer: producer, batch: defaultBatchSize, workers: defaultWorkers, timeout: defaultDeliveryTimeout, maxAttempts: defaultMaxAttempts, backoff: ExponentialBackoff(time.Second, 10*time.Minute), } for _, opt := range opts { opt(r) } switch { case r.batch < 1: return nil, fmt.Errorf("outpost: batch size must be >= 1, got %d", r.batch) case r.workers < 1: return nil, fmt.Errorf("outpost: workers must be >= 1, got %d", r.workers) case r.timeout <= 0: return nil, fmt.Errorf("outpost: delivery timeout must be > 0, got %s", r.timeout) case r.maxAttempts < 1: return nil, fmt.Errorf("outpost: max attempts must be >= 1, got %d", r.maxAttempts) case r.backoff == nil: return nil, errors.New("outpost: nil backoff") } batches := int64(r.batch) / int64(r.workers) if int64(r.batch)%int64(r.workers) != 0 { batches++ } if limit := int64(maxDerivedLease / r.timeout); batches >= limit { return nil, fmt.Errorf("outpost: derived lease would exceed %s; lower the batch size or raise the worker count", maxDerivedLease) } r.lease = time.Duration(batches+1) * r.timeout return r, nil}
// Run polls until ctx is done, one batch at a time. Store errors are reported// through Hooks.OnError and retried after a pause; Run does not return until// ctx is cancelled.func (r *Relay) Run(ctx context.Context) { for ctx.Err() == nil { n, err := r.cycle(ctx) if err != nil { r.hooks.err(err) if !sleep(ctx, defaultPollInterval) { return } continue } if n == 0 && !sleep(ctx, defaultPollInterval) { return } }}
// RunOnce performs a single claim-and-process cycle and reports how many// messages were claimed. Loop it to drain, or call it from a cron job.func (r *Relay) RunOnce(ctx context.Context) (int, error) { return r.cycle(ctx)}
// cycle claims one batch, delivers it, then records the outcomes.func (r *Relay) cycle(ctx context.Context) (int, error) { claims, err := r.store.Claim(ctx, r.batch, r.lease) if err != nil { return 0, err } if len(claims) == 0 { return 0, nil }
var ( mu sync.Mutex done []ClaimRef failures []failure wg sync.WaitGroup ) sem := make(chan struct{}, r.workers) for _, c := range claims { if ctx.Err() != nil { // Shutting down: remaining leases expire and are redelivered. break } sem <- struct{}{} wg.Add(1) go func() { defer wg.Done() defer func() { <-sem }() ref, f := r.deliver(ctx, c) switch { case ref != nil: mu.Lock() done = append(done, *ref) mu.Unlock() case f != nil: mu.Lock() failures = append(failures, *f) mu.Unlock() } }() } wg.Wait()
if len(done) > 0 { r.record(ctx, func(ctx context.Context) error { return r.store.Complete(ctx, done...) }) } for _, f := range failures { r.record(ctx, func(ctx context.Context) error { return r.store.Fail(ctx, f.ref, f.cause, f.retryAt) }) } return len(claims), nil}
type failure struct { ref ClaimRef cause error retryAt *time.Time}
// deliver runs one delivery. It returns the ref on success, a failure when one// must be recorded, and neither while the relay is shutting down.func (r *Relay) deliver(ctx context.Context, c Claim) (*ClaimRef, *failure) { info := Info{ID: c.ID, Name: c.Name, Key: c.Key, Attempt: c.Attempts + 1, Headers: c.Headers}
dctx, cancel := context.WithTimeout(ctx, r.timeout) defer cancel() dctx = context.WithValue(dctx, infoKey{}, info)
err := r.producer.Produce(dctx, c.Message) switch { case err == nil: ref := ClaimRef{ID: c.ID, Token: c.Token} return &ref, nil case ctx.Err() != nil: return nil, nil }
retryAt, abandon := r.verdict(c, err) if abandon { r.hooks.deadLetter(info, err) } else { r.hooks.retry(info, err, *retryAt) } ref := ClaimRef{ID: c.ID, Token: c.Token} return nil, &failure{ref: ref, cause: err, retryAt: retryAt}}
// verdict maps a producer error to a retry time, or abandons the message.// Precedence: Permanent, then exhausted attempts, then RetryAfter, then the// configured backoff.func (r *Relay) verdict(c Claim, err error) (*time.Time, bool) { if isPermanent(err) || c.Attempts+1 >= r.maxAttempts { return nil, true } delay := r.backoff(c.Attempts + 1) if d, ok := retryDelay(err); ok { delay = d } t := time.Now().Add(spread(delay)) return &t, false}
// spread returns d plus up to half of it, so independent workers do not retry// at the same instant. It never returns less than d: a configured backoff and a// provider's Retry-After are both floors.//// A half is the largest fraction that keeps attempt windows disjoint under a// doubling backoff, which is why it is a constant and not a setting.//// math/rand/v2's top-level functions are safe for concurrent use, so this needs// no state or mutex even though every delivery goroutine calls it.func spread(d time.Duration) time.Duration { if d <= 0 { return d } const maxDuration = time.Duration(1<<63 - 1) if d > maxDuration-d/2 { // Guard the addition: an absurd delay would wrap into the past. return d } return d + time.Duration(rand.Int64N(int64(d/2)+1))}
// record writes an outcome to the store, ignoring cancellation and stale// claims.func (r *Relay) record(ctx context.Context, fn func(context.Context) error) { wctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), storeOpTimeout) defer cancel() if err := fn(wctx); err != nil && !errors.Is(err, ErrStaleClaim) { r.hooks.err(err) }}
func sleep(ctx context.Context, d time.Duration) bool { if d <= 0 { return ctx.Err() == nil } t := time.NewTimer(d) defer t.Stop() select { case <-ctx.Done(): return false case <-t.C: return true }}