broad-announce/internal/retry/retry.go

// Package retry is the M8 in-process retry helper used by
// deliverd-fcm and deliverd-telegram. It implements
// bounded exponential backoff with a per-attempt cap
// and a total time budget so a single delivery can't
// tie up a NATS consumer for minutes.
//
// Why in-process retry, not JetStream redelivery?
//   - We want explicit, config-driven backoff (SPEC §9:
//     1s, 2s, 4s, … up to 10 attempts). JetStream's
//     redelivery timer is fixed at the consumer level
//     and doesn't express per-attempt exp backoff.
//   - We want a single deliveries table row per
//     attempt for the audit trail. JetStream redelivery
//     would re-process the same envelope; we'd have
//     to dedupe in the worker anyway.
//   - We want a hard cap (RetryBudget) so a stuck
//     downstream (e.g. fakefcmd with --fail-rate=1.0)
//     can never block a consumer for the full
//     1+2+4+…+512 = 1023s the SPEC literally calls
//     for. In practice we ship defaults that
//     terminate in ~10s.
package retry

import (
	"context"
	"errors"
	"time"
)

// Config is the retry policy. All durations are wall
// clock; the helper never sleeps past the ctx deadline.
type Config struct {
	// MaxAttempts is the total number of attempts
	// (including the first). Default 10. After
	// MaxAttempts failures, Run returns the last
	// error from fn.
	MaxAttempts int
	// BaseDelay is the wait before the SECOND attempt;
	// it doubles each subsequent attempt. Default 100ms.
	BaseDelay time.Duration
	// MaxDelay caps the per-attempt wait. Default 2s.
	// With BaseDelay=100ms, MaxDelay=2s, MaxAttempts=10,
	// the per-attempt waits are 100, 200, 400, 800, 1600,
	// 2000, 2000, 2000, 2000 ms (9 waits), ~12s total.
	MaxDelay time.Duration
	// Budget is the wall-clock cap across all attempts.
	// Run returns ctx.DeadlineExceeded if the budget
	// is hit before MaxAttempts completes. Default 30s.
	Budget time.Duration
}

// Default returns the M8 spec defaults.
func Default() Config {
	return Config{
		MaxAttempts: 10,
		BaseDelay:  100 * time.Millisecond,
		MaxDelay:   2 * time.Second,
		Budget:     30 * time.Second,
	}
}

// PermanentError signals "don't retry this". The helper
// returns it to the caller as-is after wrapping the
// attempt count. Use it for HTTP 4xx (except 408/429),
// parse errors, and other "retry won't help" cases.
type PermanentError struct {
	Err error
}

func (e *PermanentError) Error() string { return e.Err.Error() }
func (e *PermanentError) Unwrap() error { return e.Err }

// IsPermanent reports whether err is a PermanentError.
func IsPermanent(err error) bool {
	var p *PermanentError
	return errors.As(err, &p)
}

// Result is the outcome of Run.
type Result struct {
	// Attempts is the number of fn invocations that ran
	// (1 = succeeded on first try, MaxAttempts = gave up).
	Attempts int
	// LastError is the error from the final attempt, or
	// nil on success.
	LastError error
}

// Run calls fn up to cfg.MaxAttempts times, sleeping
// exp-backoff between failed attempts. It returns a
// Result with the attempt count and last error.
//
// The sleep respects ctx cancellation. If the budget
// is hit before all attempts complete, fn is not called
// again and Result.LastError is the original fn error
// (not ctx.DeadlineExceeded, so the caller can decide
// whether to DLQ the message).
//
// PermanentError short-circuits the loop: if fn
// returns &PermanentError{…}, Run returns immediately
// with Attempts set to the current count and LastError
// = the wrapped error.
func Run(ctx context.Context, cfg Config, fn func(ctx context.Context, attempt int) error) Result {
	if cfg.MaxAttempts <= 0 {
		cfg.MaxAttempts = 10
	}
	if cfg.BaseDelay <= 0 {
		cfg.BaseDelay = 100 * time.Millisecond
	}
	if cfg.MaxDelay <= 0 {
		cfg.MaxDelay = 2 * time.Second
	}
	if cfg.Budget <= 0 {
		cfg.Budget = 30 * time.Second
	}
	deadline := time.Now().Add(cfg.Budget)
	res := Result{}
	for attempt := 1; attempt <= cfg.MaxAttempts; attempt++ {
		res.Attempts = attempt
		err := fn(ctx, attempt)
		if err == nil {
			res.LastError = nil // clear any prior failure on success
			return res
		}
		res.LastError = err
		if IsPermanent(err) {
			return res
		}
		if attempt == cfg.MaxAttempts {
			break
		}
		// Compute this attempt's wait, then sleep
		// with ctx + budget awareness.
		wait := backoff(cfg.BaseDelay, cfg.MaxDelay, attempt)
		// If the budget would be exceeded by this wait,
		// bail out early with the last error.
		if time.Now().Add(wait).After(deadline) {
			break
		}
		t := time.NewTimer(wait)
		select {
		case <-ctx.Done():
			t.Stop()
			return res
		case <-t.C:
		}
	}
	return res
}

// backoff returns the wait for the Nth retry. attempt=1
// is the FIRST attempt (no wait); attempt=2 is the wait
// before the second attempt. So we use attempt-1 to
// compute the exponent.
func backoff(base, max time.Duration, attempt int) time.Duration {
	if attempt < 2 {
		return 0
	}
	d := base
	for i := 2; i < attempt; i++ {
		d *= 2
		if d > max {
			return max
		}
	}
	if d > max {
		return max
	}
	return d
}