backstop_test.go

package imappoller

import (
	"context"
	"errors"
	"sync/atomic"
	"testing"
	"time"
)

func TestBackstop_PollsStaleAccounts(t *testing.T) {
	accounts := newMemAccounts()
	const (
		stale    = int64(1)
		fresh    = int64(2)
		nilAlive = int64(3)
	)
	accounts.seedStale(stale, 15*time.Minute) // last alive 15m ago -> stale
	accounts.seedStale(fresh, 0)              // last alive now -> fresh
	accounts.seedStale(nilAlive, -1)          // never alive -> stale

	var pollCount atomic.Int32
	polled := make(chan int64, 16)

	bs := &Backstop{
		Accounts:   accounts,
		Interval:   100 * time.Millisecond,
		StaleAfter: 10 * time.Minute,
		BatchSize:  100,
		PollAccount: func(_ context.Context, id int64) error {
			pollCount.Add(1)
			polled <- id
			return nil
		},
	}

	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
	defer cancel()
	go bs.Run(ctx)

	if !waitFor(500*time.Millisecond, func() bool { return pollCount.Load() >= 1 }) {
		t.Fatal("backstop did not poll the stale accounts")
	}

	seen := map[int64]bool{}
collect:
	for {
		select {
		case id := <-polled:
			seen[id] = true
		case <-time.After(300 * time.Millisecond):
			break collect
		}
	}

	if !seen[stale] {
		t.Errorf("stale account %d was not polled (seen=%v)", stale, seen)
	}
	if !seen[nilAlive] {
		t.Errorf("nilAlive account %d was not polled (seen=%v)", nilAlive, seen)
	}
	if seen[fresh] {
		t.Errorf("fresh account %d was unexpectedly polled (seen=%v)", fresh, seen)
	}
}

func TestBackstop_MarksAliveOnSuccess(t *testing.T) {
	accounts := newMemAccounts()
	const stale = int64(1)
	accounts.seedStale(stale, 15*time.Minute)

	bs := &Backstop{
		Accounts:    accounts,
		Interval:    100 * time.Millisecond,
		StaleAfter:  10 * time.Minute,
		BatchSize:   100,
		PollAccount: func(context.Context, int64) error { return nil },
	}

	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
	defer cancel()
	go bs.Run(ctx)

	if !waitFor(500*time.Millisecond, func() bool {
		age, ok := accounts.aliveAge(stale)
		return ok && age < 500*time.Millisecond
	}) {
		t.Fatal("backstop did not MarkAlive on the stale account")
	}
}

// TestBackstop_PerAccountTimeoutUnblocksSweep pins the per-account timeout: with
// concurrency forced to 1, a hung poll ahead of a good one must not block it
// forever — the timeout bounds the hung poll so the good one gets its turn.
func TestBackstop_PerAccountTimeoutUnblocksSweep(t *testing.T) {
	origTimeout := backstopAccountTimeout
	backstopAccountTimeout = 20 * time.Millisecond
	t.Cleanup(func() { backstopAccountTimeout = origTimeout })
	origConc := backstopConcurrency
	backstopConcurrency = 1
	t.Cleanup(func() { backstopConcurrency = origConc })

	accounts := newMemAccounts()
	const (
		hangs    = int64(1) // lower id -> dispatched first through the single slot
		succeeds = int64(2)
	)
	accounts.seedStale(hangs, 20*time.Minute)
	accounts.seedStale(succeeds, 15*time.Minute)

	var succeedsPolled atomic.Bool
	bs := &Backstop{
		Accounts:   accounts,
		Interval:   time.Hour,
		StaleAfter: 10 * time.Minute,
		BatchSize:  100,
		PollAccount: func(pctx context.Context, id int64) error {
			if id == hangs {
				<-pctx.Done()
				return pctx.Err()
			}
			succeedsPolled.Store(true)
			return nil
		},
	}

	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
	defer cancel()
	go bs.Run(ctx)

	if !waitFor(2*time.Second, func() bool { return succeedsPolled.Load() }) {
		t.Fatalf("account %d was never polled — the hung account %d blocked it", succeeds, hangs)
	}
}

// TestBackstop_FailedPollAdvancesAttemptNotAlive verifies a failed poll advances
// the retry position (so the account stops monopolising the head of the queue)
// but does not mark it alive, and the retry floor keeps an immediate second
// sweep from re-listing it.
func TestBackstop_FailedPollAdvancesAttemptNotAlive(t *testing.T) {
	accounts := newMemAccounts()
	const poison = int64(1)
	accounts.seedStale(poison, 15*time.Minute)

	var pollCount atomic.Int32
	bs := &Backstop{
		Accounts:   accounts,
		StaleAfter: 10 * time.Minute,
		RetryFloor: time.Hour,
		BatchSize:  100,
		PollAccount: func(context.Context, int64) error {
			pollCount.Add(1)
			return errors.New("always fails")
		},
	}

	ctx := context.Background()
	bs.sweep(ctx) // polls (and fails) poison
	bs.sweep(ctx) // retry floor must exclude it

	if n := pollCount.Load(); n != 1 {
		t.Fatalf("PollAccount called %d times across two immediate sweeps, want exactly 1", n)
	}
	if !accounts.attempted(poison) {
		t.Error("MarkAttempted never advanced on a failed poll")
	}
	if age, _ := accounts.aliveAge(poison); age < 14*time.Minute {
		t.Errorf("last-alive moved on a failed poll (age=%v); MarkAlive must stay success-only", age)
	}
}

// TestBackstop_SweepIsConcurrent verifies the sweep runs accounts concurrently
// under a bound: a 100-account batch of 50ms polls finishes far faster than
// serial, and the in-flight high-water mark lands in (1, backstopConcurrency].
func TestBackstop_SweepIsConcurrent(t *testing.T) {
	accounts := newMemAccounts()
	const n = 100
	for i := int64(1); i <= n; i++ {
		accounts.seedStale(i, 15*time.Minute)
	}

	var polled, current, highWater atomic.Int32
	bs := &Backstop{
		Accounts:   accounts,
		StaleAfter: 10 * time.Minute,
		RetryFloor: time.Hour,
		BatchSize:  n,
		PollAccount: func(context.Context, int64) error {
			c := current.Add(1)
			for {
				hw := highWater.Load()
				if c <= hw || highWater.CompareAndSwap(hw, c) {
					break
				}
			}
			time.Sleep(50 * time.Millisecond)
			current.Add(-1)
			polled.Add(1)
			return nil
		},
	}

	start := time.Now()
	bs.sweep(context.Background())
	elapsed := time.Since(start)

	if int(polled.Load()) != n {
		t.Fatalf("only %d/%d accounts polled", polled.Load(), n)
	}
	if elapsed >= 2*time.Second {
		t.Errorf("sweep of %d accounts (50ms each) took %v, well over a concurrent sweep", n, elapsed)
	}
	if hw := highWater.Load(); hw <= 1 || hw > int32(backstopConcurrency) {
		t.Errorf("in-flight high-water = %d, want 1 < hw <= %d", hw, backstopConcurrency)
	}
}