member-console/internal/retryloop/retryloop.go

// SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Commercial
// SPDX-FileCopyrightText: 2025-2026 Christian Galo

// Package retryloop runs one attempt function until it succeeds or its
// context ends, with a capped, jittered backoff and a log that decays.
//
// It serves the two dependencies the console waits for in the background, the
// identity provider's discovery document and the Temporal connection, so that
// both follow one policy and the bound on log lines is proved once.
//
// The waits. The first wait is FirstWait (0.5 seconds) and doubles after each
// failed attempt up to Cap (15 seconds). Each wait is the base wait plus a
// uniform random fraction of up to half of it, so a wait at the cap is between
// 15 and 22.5 seconds. A failure marked with AtCap makes the base of the next
// wait the cap, whatever the attempt count; the jitter still applies.
//
// The wake. A send on Policy.Wake ends the wait, but not sooner than MinWakeGap
// (1 second) after the last attempt began, so a burst of wakes makes at most one
// attempt a second. A wake does not end a wait that follows an AtCap failure.
//
// The log. Every line goes through Run; an attempt logs nothing itself. A
// failed attempt is written when it is the first failure, when it is the first
// line written after Policy.Escalation has elapsed, when its error text differs
// from the last line's and 30 seconds have passed since that line, and
// otherwise once 5 minutes have passed since the last line. A line is Warn until
// Policy.Escalation has elapsed since Run began and Error after, and a
// Failure's Level is a floor under that. The success is one Info line.
// Every failure line carries the attributes attempts (the attempts made so
// far), elapsed, retry_in and error, after Policy.Attrs; the success line
// carries attempts and elapsed.
//
// The bound. With a steady error, an hour of retries writes at most 14 lines:
// the first, the first Error line, and one per 5 minutes. With an error whose
// text changes on every attempt it writes at most 120 more, one per 30 seconds.
package retryloop

import (
	"context"
	"errors"
	"log/slog"
	"math/rand/v2"
	"time"
)

// The defaults a zero Policy field takes.
const (
	// DefaultFirstWait is the wait after the first failed attempt.
	DefaultFirstWait = 500 * time.Millisecond
	// DefaultCap is the longest base wait.
	DefaultCap = 15 * time.Second
	// DefaultMinWakeGap is how long after an attempt began a wake counts.
	DefaultMinWakeGap = time.Second
	// DefaultEscalation is the time after which the lines are Error.
	DefaultEscalation = 90 * time.Second

	// changedTextGap is the least time between two lines whose error text
	// differs, and repeatGap the least time between two lines that repeat.
	changedTextGap = 30 * time.Second
	repeatGap      = 5 * time.Minute
)

// Clock is the time the loop reads and waits on.
type Clock interface {
	Now() time.Time
	NewTimer(d time.Duration) Timer
}

// Timer is a one-shot timer, as *time.Timer is.
type Timer interface {
	C() <-chan time.Time
	Stop() bool
}

// Source is the random source of the jitter: Float64 returns a number in
// [0, 1), as *math/rand/v2.Rand does.
type Source interface {
	Float64() float64
}

// Policy is how Run waits and what it logs. Its zero value is usable: the
// zero FirstWait, Cap, MinWakeGap and Escalation take the defaults above, the
// Logger is slog's default, and the clock and the random source are the real
// ones.
type Policy struct {
	// FirstWait is the wait after the first failure, doubled after each
	// failed attempt until Cap.
	FirstWait time.Duration
	// Cap is the longest base wait, and the base of the wait after an AtCap failure.
	Cap time.Duration
	// Escalation is the time after Run began from which the lines are Error.
	Escalation time.Duration
	// Wake, when set, ends a wait that follows an ordinary failure.
	Wake <-chan struct{}
	// MinWakeGap is the least time between the start of an attempt and the
	// start of the next one that a wake causes.
	MinWakeGap time.Duration
	// FailureMessage is the message of a failed attempt that returns a plain
	// error, or a Failure with no Message.
	FailureMessage string
	// SuccessMessage is the message of the Info line written on success.
	SuccessMessage string
	// Attrs are carried by every line, before the loop's own.
	Attrs []slog.Attr
	// Logger writes the lines. Nil means slog.Default().
	Logger *slog.Logger
	// Clock is the time source. Nil means the real clock.
	Clock Clock
	// Rand is the jitter's source. Nil means the package-level source of
	// math/rand/v2.
	Rand Source
}

// Failure is a failed attempt that carries its own log line: the message to
// write in place of the policy's, a level floor, and the error that failed.
// Return it by value or by pointer. A Level at or below Warn changes nothing; Error writes the line at Error
// from the first.
type Failure struct {
	Message string
	Level   slog.Level
	Err     error
}

// Error returns the text of the wrapped error.
func (f Failure) Error() string {
	if f.Err == nil {
		return f.Message
	}
	return f.Err.Error()
}

// Unwrap returns the wrapped error.
func (f Failure) Unwrap() error { return f.Err }

// atCap marks an error after which the base of the next wait is the cap.
type atCap struct{ err error }

func (a atCap) Error() string { return a.err.Error() }
func (a atCap) Unwrap() error { return a.err }

// AtCap marks err so that the base of the wait after it is the cap and a wake
// does not end that wait. A nil err stays nil.
func AtCap(err error) error {
	if err == nil {
		return nil
	}
	return atCap{err}
}

// Run calls attempt until it returns nil, and then returns nil, or until ctx
// ends, and then returns ctx.Err(); a ctx that has already ended returns before
// the first attempt. An error from attempt is retried after the
// wait the policy gives, and logged as the package comment says.
func Run(ctx context.Context, policy Policy, attempt func(context.Context) error) error {
	if err := ctx.Err(); err != nil {
		return err
	}
	l := newLoop(policy)
	l.start = l.clock.Now()
	for {
		began := l.clock.Now()
		l.attempts++
		err := attempt(ctx)
		if err == nil {
			l.logSuccess()
			return nil
		}
		if ctx.Err() != nil {
			return ctx.Err()
		}
		wait, wakeable := l.nextWait(err)
		l.logFailure(err, wait)
		if err := l.sleep(ctx, wait, began, wakeable); err != nil {
			return err
		}
	}
}

// loop is the state of one Run.
type loop struct {
	p      Policy
	clock  Clock
	rand   Source
	logger *slog.Logger

	start    time.Time
	attempts int
	base     time.Duration // the base wait after the next failure

	logged           bool
	escalationLogged bool // a line has been written since Escalation elapsed
	lastAt           time.Time
	lastText         string
}

func newLoop(p Policy) *loop {
	if p.FirstWait <= 0 {
		p.FirstWait = DefaultFirstWait
	}
	if p.Cap <= 0 {
		p.Cap = DefaultCap
	}
	if p.MinWakeGap <= 0 {
		p.MinWakeGap = DefaultMinWakeGap
	}
	if p.Escalation <= 0 {
		p.Escalation = DefaultEscalation
	}
	l := &loop{p: p, clock: p.Clock, rand: p.Rand, logger: p.Logger, base: p.FirstWait}
	if l.clock == nil {
		l.clock = realClock{}
	}
	if l.rand == nil {
		l.rand = globalSource{}
	}
	if l.logger == nil {
		l.logger = slog.Default()
	}
	return l
}

// nextWait returns the wait after the failed attempt err, jitter included,
// and whether a wake may end it.
func (l *loop) nextWait(err error) (time.Duration, bool) {
	base := l.base
	l.base = min(l.base*2, l.p.Cap)
	var pinned atCap
	atTheCap := errors.As(err, &pinned)
	if atTheCap {
		base = l.p.Cap
	}
	jitter := time.Duration(l.rand.Float64() * float64(base) / 2)
	return base + jitter, !atTheCap
}

// sleep waits d, or less when a wake counts, and returns ctx.Err() when ctx
// ends first.
func (l *loop) sleep(ctx context.Context, d time.Duration, began time.Time, wakeable bool) error {
	timer := l.clock.NewTimer(d)
	defer func() { timer.Stop() }()
	ends := l.clock.Now().Add(d)
	var wake <-chan struct{}
	if wakeable {
		wake = l.p.Wake
	}
	for {
		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-timer.C():
			return nil
		case <-wake:
			wake = nil
			earliest := began.Add(l.p.MinWakeGap)
			remaining := earliest.Sub(l.clock.Now())
			if remaining <= 0 {
				return nil
			}
			if earliest.Before(ends) {
				timer.Stop()
				timer = l.clock.NewTimer(remaining)
				ends = earliest
			}
		}
	}
}

// attrs are the attributes every line carries.
func (l *loop) attrs(retryIn time.Duration, err error) []slog.Attr {
	attrs := append([]slog.Attr(nil), l.p.Attrs...)
	attrs = append(attrs,
		slog.Int("attempts", l.attempts),
		slog.Duration("elapsed", l.clock.Now().Sub(l.start)))
	if err != nil {
		attrs = append(attrs,
			slog.Duration("retry_in", retryIn),
			slog.Any("error", err))
	}
	return attrs
}

func (l *loop) logSuccess() {
	l.logger.LogAttrs(context.Background(), slog.LevelInfo, l.p.SuccessMessage, l.attrs(0, nil)...)
}

// logFailure writes the line for a failed attempt when the decay allows it.
func (l *loop) logFailure(err error, retryIn time.Duration) {
	now := l.clock.Now()
	message, level := l.p.FailureMessage, slog.LevelWarn
	escalated := now.Sub(l.start) >= l.p.Escalation
	if escalated {
		level = slog.LevelError
	}
	shown := err
	var failure Failure
	var byPointer *Failure
	found := errors.As(err, &failure)
	if !found && errors.As(err, &byPointer) && byPointer != nil {
		failure, found = *byPointer, true
	}
	if found {
		if failure.Message != "" {
			message = failure.Message
		}
		level = max(level, failure.Level)
		if failure.Err != nil {
			shown = failure.Err
		}
	}
	if !l.due(now, shown.Error()) {
		return
	}
	l.logged, l.lastAt, l.lastText = true, now, shown.Error()
	l.escalationLogged = l.escalationLogged || escalated
	l.logger.LogAttrs(context.Background(), level, message, l.attrs(retryIn, shown)...)
}

// due reports whether a failure with this error text is written now.
func (l *loop) due(now time.Time, text string) bool {
	if !l.logged || (!l.escalationLogged && now.Sub(l.start) >= l.p.Escalation) {
		return true
	}
	since := now.Sub(l.lastAt)
	if text != l.lastText {
		return since >= changedTextGap
	}
	return since >= repeatGap
}

type realClock struct{}

func (realClock) Now() time.Time { return time.Now() }

func (realClock) NewTimer(d time.Duration) Timer { return realTimer{time.NewTimer(d)} }

type realTimer struct{ t *time.Timer }

func (r realTimer) C() <-chan time.Time { return r.t.C }
func (r realTimer) Stop() bool          { return r.t.Stop() }

type globalSource struct{}

func (globalSource) Float64() float64 { return rand.Float64() }