member-console/internal/retryloop/retryloop_test.go

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

package retryloop

import (
	"context"
	"errors"
	"fmt"
	"log/slog"
	"math/rand/v2"
	"sync"
	"testing"
	"testing/synctest"
	"time"
)

// deadline bounds every wait a test makes on the loop's goroutine; a test
// that reaches it fails naming what it waited for.
const deadline = 5 * time.Second

// fakeClock is the loop's clock. In auto mode a timer fires at once and moves
// the clock by its duration, so a run is a sequence of recorded waits. In
// manual mode a timer is announced on timers and fires when the test says so.
type fakeClock struct {
	mu     sync.Mutex
	now    time.Time
	auto   bool
	waits  []time.Duration
	timers chan *fakeTimer
}

type fakeTimer struct {
	clock *fakeClock
	c     chan time.Time
	d     time.Duration
}

func newClock(auto bool) *fakeClock {
	return &fakeClock{
		now:    time.Date(2026, 10, 1, 0, 0, 0, 0, time.UTC),
		auto:   auto,
		timers: make(chan *fakeTimer, 64),
	}
}

func (f *fakeClock) Now() time.Time {
	f.mu.Lock()
	defer f.mu.Unlock()
	return f.now
}

func (f *fakeClock) advance(d time.Duration) {
	f.mu.Lock()
	defer f.mu.Unlock()
	f.now = f.now.Add(d)
}

func (f *fakeClock) NewTimer(d time.Duration) Timer {
	t := &fakeTimer{clock: f, c: make(chan time.Time, 1), d: d}
	f.mu.Lock()
	f.waits = append(f.waits, d)
	auto := f.auto
	f.mu.Unlock()
	if auto {
		t.fire()
		return t
	}
	f.timers <- t
	return t
}

func (t *fakeTimer) fire() {
	t.clock.advance(t.d)
	t.c <- t.clock.Now()
}

func (t *fakeTimer) C() <-chan time.Time { return t.c }
func (t *fakeTimer) Stop() bool          { return true }

// seeded returns two sources that yield the same numbers, so a test can
// compute the jitter the loop drew.
func seeded() (*rand.Rand, *rand.Rand) {
	return rand.New(rand.NewPCG(7, 11)), rand.New(rand.NewPCG(7, 11))
}

// noJitter draws zero every time, so a wait is exactly its base.
type noJitter struct{}

func (noJitter) Float64() float64 { return 0 }

// recordingHandler keeps what the loop logged.
type recordingHandler struct {
	mu      sync.Mutex
	records []slog.Record
}

func (h *recordingHandler) Enabled(context.Context, slog.Level) bool { return true }
func (h *recordingHandler) Handle(_ context.Context, r slog.Record) error {
	h.mu.Lock()
	defer h.mu.Unlock()
	h.records = append(h.records, r)
	return nil
}
func (h *recordingHandler) WithAttrs([]slog.Attr) slog.Handler { return h }
func (h *recordingHandler) WithGroup(string) slog.Handler      { return h }

func (h *recordingHandler) lines() []logLine {
	h.mu.Lock()
	defer h.mu.Unlock()
	out := make([]logLine, 0, len(h.records))
	for _, r := range h.records {
		line := logLine{level: r.Level, message: r.Message, attrs: map[string]slog.Value{}}
		r.Attrs(func(a slog.Attr) bool {
			line.attrs[a.Key] = a.Value
			return true
		})
		out = append(out, line)
	}
	return out
}

type logLine struct {
	level   slog.Level
	message string
	attrs   map[string]slog.Value
}

func (l logLine) elapsed() time.Duration { return l.attrs["elapsed"].Duration() }

// started is a Run in its own goroutine.
type started struct {
	result chan error
}

func start(ctx context.Context, p Policy, attempt func(context.Context) error) *started {
	s := &started{result: make(chan error, 1)}
	go func() { s.result <- Run(ctx, p, attempt) }()
	return s
}

func (s *started) await(t *testing.T, what string) error {
	t.Helper()
	select {
	case err := <-s.result:
		return err
	case <-time.After(deadline):
		t.Fatalf("waited %v for %s", deadline, what)
		return nil
	}
}

// nextTimer waits for the loop to announce its next wait.
func nextTimer(t *testing.T, c *fakeClock, what string) *fakeTimer {
	t.Helper()
	select {
	case timer := <-c.timers:
		return timer
	case <-time.After(deadline):
		t.Fatalf("waited %v for %s", deadline, what)
		return nil
	}
}

// failing returns an attempt that fails with err each time and signals the
// clock reading at each start.
func failing(c *fakeClock, err error, began chan<- time.Time) func(context.Context) error {
	return func(context.Context) error {
		began <- c.Now()
		return err
	}
}

func receive[T any](t *testing.T, ch <-chan T, what string) T {
	t.Helper()
	select {
	case v := <-ch:
		return v
	case <-time.After(deadline):
		t.Fatalf("waited %v for %s", deadline, what)
		var zero T
		return zero
	}
}

func TestRunWaitsFollowThePolicy(t *testing.T) {
	clock := newClock(true)
	drawn, expected := seeded()
	attempts := 0
	p := Policy{Clock: clock, Rand: drawn, Logger: slog.New(&recordingHandler{})}
	err := Run(t.Context(), p, func(context.Context) error {
		attempts++
		if attempts <= 8 {
			return errors.New("refused")
		}
		return nil
	})
	if err != nil {
		t.Fatalf("Run = %v, want nil", err)
	}
	bases := []time.Duration{500 * time.Millisecond, time.Second, 2 * time.Second, 4 * time.Second, 8 * time.Second, 15 * time.Second, 15 * time.Second, 15 * time.Second}
	if len(clock.waits) != len(bases) {
		t.Fatalf("waits = %v, want %d of them", clock.waits, len(bases))
	}
	for i, base := range bases {
		want := base + time.Duration(expected.Float64()*float64(base)/2)
		if clock.waits[i] != want {
			t.Errorf("wait %d = %v, want %v (base %v plus the seeded fraction of half of it)", i+1, clock.waits[i], want, base)
		}
	}
}

// wakeRow is one attempt that takes took, then a wake in the wait after it.
type wakeRow struct {
	name string
	took time.Duration
	// gap is the time from the start of the first attempt to the start of the
	// second.
	gap time.Duration
}

func runWakeRow(t *testing.T, row wakeRow) {
	clock := newClock(false)
	wake := make(chan struct{}, 1)
	began := make(chan time.Time, 4)
	calls := 0
	p := Policy{FirstWait: 15 * time.Second, Cap: 15 * time.Second, Clock: clock, Rand: noJitter{}, Wake: wake, Logger: slog.New(&recordingHandler{})}
	run := start(t.Context(), p, func(context.Context) error {
		began <- clock.Now()
		calls++
		if calls == 1 {
			clock.advance(row.took)
			return errors.New("refused")
		}
		return nil
	})
	first := receive(t, began, "the first attempt")
	wait := nextTimer(t, clock, "the fifteen-second wait")
	if wait.d != 15*time.Second {
		t.Fatalf("first wait = %v, want 15s", wait.d)
	}
	wake <- struct{}{}
	if remaining := row.gap - row.took; remaining > 0 {
		short := nextTimer(t, clock, "the wait that a wake shortens to the one-second gap")
		if short.d != remaining {
			t.Fatalf("a wake %v into the attempt left a wait of %v, want %v", row.took, short.d, remaining)
		}
		short.fire()
	}
	second := receive(t, began, "the attempt a wake starts")
	if got := second.Sub(first); got != row.gap {
		t.Errorf("the second attempt began %v after the first, want %v", got, row.gap)
	}
	if err := run.await(t, "Run to end after the second attempt"); err != nil {
		t.Errorf("Run = %v, want nil", err)
	}
}

func TestRunWakeEndsTheWait(t *testing.T) {
	rows := []wakeRow{
		{name: "a wake right after the attempt waits for the second", took: 0, gap: time.Second},
		{name: "a wake within the second waits for the rest of it", took: 300 * time.Millisecond, gap: time.Second},
		{name: "a wake after the second starts the attempt at once", took: 4 * time.Second, gap: 4 * time.Second},
	}
	for _, row := range rows {
		t.Run(row.name, func(t *testing.T) { runWakeRow(t, row) })
	}
}

func TestRunAtCapWaitsTheCap(t *testing.T) {
	synctest.Test(t, func(t *testing.T) {
		clock := newClock(false)
		// The wake channel is buffered as the discovery loop's is. A loop that
		// listened for a wake during an AtCap wait would take it as soon as it is
		// sent and start a one-second timer; this one leaves it unread.
		wake := make(chan struct{}, 1)
		began := make(chan time.Time, 4)
		calls := 0
		drawn, expected := seeded()
		p := Policy{Clock: clock, Rand: drawn, Wake: wake, Logger: slog.New(&recordingHandler{})}
		run := start(t.Context(), p, func(context.Context) error {
			began <- clock.Now()
			calls++
			if calls == 1 {
				return AtCap(errors.New("unusable"))
			}
			return nil
		})
		first := receive(t, began, "the first attempt")
		wait := nextTimer(t, clock, "the wait after an AtCap failure")
		wantWait := DefaultCap + time.Duration(expected.Float64()*float64(DefaultCap)/2)
		if wait.d != wantWait {
			t.Fatalf("wait after an AtCap failure = %v, want the cap plus the seeded jitter, %v", wait.d, wantWait)
		}
		synctest.Wait() // the loop is blocked in the wait after the AtCap failure
		wake <- struct{}{}
		synctest.Wait() // a loop that listens for the wake has taken it by now
		if len(wake) != 1 || len(clock.timers) != 0 {
			t.Fatalf("a wake during an AtCap wait was read")
		}
		// The next attempt starts only when the fake clock reaches the cap plus the
		// jitter, which firing the timer does.
		wait.fire()
		second := receive(t, began, "the second attempt")
		if got := second.Sub(first); got != wantWait {
			t.Errorf("the second attempt began %v after the first, want the cap plus the seeded jitter, %v", got, wantWait)
		}
		if err := run.await(t, "Run to end after the second attempt"); err != nil {
			t.Errorf("Run = %v, want nil", err)
		}
		if len(clock.waits) != 1 {
			t.Errorf("waits = %v, want only the cap's wait; a wake must not start a shorter one", clock.waits)
		}
	})
}

func TestRunAtCapPinsOnlyItsOwnWait(t *testing.T) {
	clock := newClock(true)
	calls := 0
	p := Policy{Clock: clock, Rand: noJitter{}, Logger: slog.New(&recordingHandler{})}
	err := Run(t.Context(), p, func(context.Context) error {
		calls++
		switch calls {
		case 1:
			return AtCap(errors.New("unusable"))
		case 2:
			return errors.New("refused")
		}
		return nil
	})
	if err != nil {
		t.Fatalf("Run = %v, want nil", err)
	}
	want := []time.Duration{DefaultCap, 2 * DefaultFirstWait}
	if len(clock.waits) != len(want) || clock.waits[0] != want[0] || clock.waits[1] != want[1] {
		t.Errorf("waits = %v, want %v: the cap after the AtCap failure, then the ordinary sequence", clock.waits, want)
	}
}

func TestRunEndedContextMakesNoAttempt(t *testing.T) {
	ctx, cancel := context.WithCancel(t.Context())
	cancel()
	calls := 0
	err := Run(ctx, Policy{Clock: newClock(true), Logger: slog.New(&recordingHandler{})}, func(context.Context) error {
		calls++
		return nil
	})
	if !errors.Is(err, context.Canceled) {
		t.Errorf("Run = %v, want context.Canceled", err)
	}
	if calls != 0 {
		t.Errorf("a context that had ended made %d attempts, want none", calls)
	}
}

func runFailureForm(t *testing.T, failure error) {
	clock := newClock(true)
	handler := &recordingHandler{}
	p := Policy{Clock: clock, Rand: noJitter{}, Logger: slog.New(handler), FailureMessage: "policy message"}
	calls := 0
	err := Run(t.Context(), p, func(context.Context) error {
		calls++
		if calls == 1 {
			return failure
		}
		return nil
	})
	if err != nil {
		t.Fatalf("Run = %v, want nil", err)
	}
	lines := handler.lines()
	if len(lines) == 0 {
		t.Fatal("nothing was logged")
	}
	if lines[0].level != slog.LevelError || lines[0].message != "unusable" {
		t.Errorf("failure line is %v %q, want Error \"unusable\"", lines[0].level, lines[0].message)
	}
}

func TestRunFailureForms(t *testing.T) {
	rows := []struct {
		name    string
		failure error
	}{
		{"by value", Failure{Message: "unusable", Level: slog.LevelError, Err: errors.New("bad")}},
		{"by pointer", &Failure{Message: "unusable", Level: slog.LevelError, Err: errors.New("bad")}},
	}
	for _, row := range rows {
		t.Run(row.name, func(t *testing.T) { runFailureForm(t, row.failure) })
	}
}

func TestRunStopsWhenTheContextEnds(t *testing.T) {
	clock := newClock(false)
	ctx, cancel := context.WithCancel(t.Context())
	defer cancel()
	began := make(chan time.Time, 4)
	run := start(ctx, Policy{Clock: clock, Rand: noJitter{}, Logger: slog.New(&recordingHandler{})},
		failing(clock, errors.New("refused"), began))
	receive(t, began, "the first attempt")
	nextTimer(t, clock, "the wait after the first attempt")
	cancel()
	if err := run.await(t, "Run to return after the context ended"); !errors.Is(err, context.Canceled) {
		t.Errorf("Run = %v, want context.Canceled", err)
	}
	select {
	case <-began:
		t.Error("an attempt began after the context ended")
	default:
	}
}

// hour is how long a decay row runs on the fake clock.
const hour = time.Hour

// runForAnHour calls attempt until an hour has passed on the fake clock and
// returns what was logged.
func runForAnHour(t *testing.T, p Policy, attempt func(n int) error) []logLine {
	t.Helper()
	clock := newClock(true)
	handler := &recordingHandler{}
	p.Clock, p.Logger = clock, slog.New(handler)
	drawn, _ := seeded()
	p.Rand = drawn
	ctx, cancel := context.WithCancel(t.Context())
	defer cancel()
	began := clock.Now()
	n := 0
	err := Run(ctx, p, func(context.Context) error {
		n++
		if clock.Now().Sub(began) >= hour {
			cancel()
		}
		return attempt(n)
	})
	if !errors.Is(err, context.Canceled) {
		t.Fatalf("Run = %v, want context.Canceled at the end of the hour", err)
	}
	return handler.lines()
}

func checkLinesCarry(t *testing.T, lines []logLine) {
	t.Helper()
	for i, line := range lines {
		for _, key := range []string{"attempts", "elapsed", "retry_in", "error"} {
			if _, ok := line.attrs[key]; !ok {
				t.Errorf("line %d (%q) has no %s attribute", i+1, line.message, key)
			}
		}
	}
}

func decaySteadyError(t *testing.T) {
	lines := runForAnHour(t, Policy{FailureMessage: "refused"}, func(int) error { return errors.New("connection refused") })
	checkLinesCarry(t, lines)
	if len(lines) > 14 {
		t.Errorf("an hour of one error wrote %d lines, want at most 14", len(lines))
	}
	if len(lines) == 0 || lines[0].level != slog.LevelWarn {
		t.Fatalf("the first line must be Warn, got %v", lines)
	}
	firstError := -1
	for i, line := range lines {
		wantLevel := slog.LevelWarn
		if line.elapsed() >= DefaultEscalation {
			wantLevel = slog.LevelError
		}
		if line.level != wantLevel {
			t.Errorf("line %d at %v is %v, want %v", i+1, line.elapsed(), line.level, wantLevel)
		}
		if line.level == slog.LevelError && firstError < 0 {
			firstError = i
		}
	}
	if firstError < 0 {
		t.Fatal("no line was written at Error after the escalation time")
	}
	if got := lines[firstError].elapsed(); got >= DefaultEscalation+DefaultCap*2 {
		t.Errorf("the first Error line was written at %v, want when %v passed", got, DefaultEscalation)
	}
}

func decayChangingError(t *testing.T) {
	lines := runForAnHour(t, Policy{FailureMessage: "refused", Escalation: 24 * hour}, func(n int) error {
		return fmt.Errorf("refused on attempt %d", n)
	})
	checkLinesCarry(t, lines)
	if len(lines) > 121 {
		t.Errorf("an hour of changing errors wrote %d lines, want at most 121", len(lines))
	}
	for i := 1; i < len(lines); i++ {
		if gap := lines[i].elapsed() - lines[i-1].elapsed(); gap < 30*time.Second {
			t.Errorf("lines %d and %d are %v apart, want at least 30s", i, i+1, gap)
		}
	}
}

func decayFailureLevel(t *testing.T) {
	lines := runForAnHour(t, Policy{FailureMessage: "unused"}, func(int) error {
		return Failure{Message: "unusable", Level: slog.LevelError, Err: errors.New("bad document")}
	})
	checkLinesCarry(t, lines)
	if len(lines) > 14 {
		t.Errorf("an hour of an Error failure wrote %d lines, want at most 14", len(lines))
	}
	for i, line := range lines {
		if line.level != slog.LevelError || line.message != "unusable" {
			t.Errorf("line %d is %v %q, want Error \"unusable\"", i+1, line.level, line.message)
		}
	}
	if len(lines) == 0 {
		t.Fatal("nothing was logged")
	}
}

func decayRecovery(t *testing.T) {
	clock := newClock(true)
	handler := &recordingHandler{}
	p := Policy{
		Clock: clock, Rand: noJitter{}, Logger: slog.New(handler),
		FailureMessage: "refused", SuccessMessage: "connected", Attrs: []slog.Attr{slog.String("host", "h")},
	}
	n := 0
	began := clock.Now()
	err := Run(t.Context(), p, func(context.Context) error {
		n++
		clock.advance(40 * time.Second)
		if n <= 3 {
			return fmt.Errorf("refused on attempt %d", n)
		}
		return nil
	})
	if err != nil {
		t.Fatalf("Run = %v, want nil", err)
	}
	lines := handler.lines()
	if len(lines) != 4 {
		t.Fatalf("logged %d lines, want three failures and one success", len(lines))
	}
	checkLinesCarry(t, lines[:3])
	for i, line := range lines[:3] {
		if got := line.attrs["attempts"].Int64(); got != int64(i+1) {
			t.Errorf("failure line %d attempts = %d, want %d", i+1, got, i+1)
		}
	}
	last := lines[3]
	if last.level != slog.LevelInfo || last.message != "connected" {
		t.Errorf("last line is %v %q, want Info \"connected\"", last.level, last.message)
	}
	if got := last.attrs["attempts"].Int64(); got != 4 {
		t.Errorf("success line attempts = %d, want 4", got)
	}
	if got, want := last.elapsed(), clock.Now().Sub(began); got != want {
		t.Errorf("success line elapsed = %v, want %v", got, want)
	}
	if got := last.attrs["host"].String(); got != "h" {
		t.Errorf("success line lost the policy's attribute: host = %q", got)
	}
}

func decayAlternatingLevels(t *testing.T) {
	lines := runForAnHour(t, Policy{FailureMessage: "refused", Escalation: 24 * hour}, func(n int) error {
		if n%2 == 0 {
			return Failure{Level: slog.LevelError, Err: errors.New("unusable")}
		}
		return errors.New("refused")
	})
	if len(lines) < 2 {
		t.Fatalf("wrote %d lines, want at least 2", len(lines))
	}
	for i := 1; i < len(lines); i++ {
		if gap := lines[i].elapsed() - lines[i-1].elapsed(); gap < 30*time.Second {
			t.Errorf("lines %d and %d are %v apart, want at least 30s", i, i+1, gap)
		}
	}
}

func TestRunLogDecay(t *testing.T) {
	rows := []struct {
		name string
		run  func(*testing.T)
	}{
		{"an hour of one error", decaySteadyError},
		{"an hour of an error that changes", decayChangingError},
		{"a failure with level Error", decayFailureLevel},
		{"levels that alternate before the escalation", decayAlternatingLevels},
		{"three failures then a success", decayRecovery},
	}
	for _, row := range rows {
		t.Run(row.name, row.run)
	}
}