Pace provider traffic and identify genuine foreground bank requests

2026-09-10 17:47:44 +02:00
parent ba3ea6ae5a
commit 4324660888
13 changed files with 817 additions and 32 deletions
@@ -13,18 +13,30 @@ import (
 )

 const (
-	maxRateAttempts = 4
-	maxRateWait     = 2 * time.Minute
+	maxRateAttempts    = 4
+	maxRateWait        = 2 * time.Minute
+	maxLearnedInterval = 30 * time.Second
+	maxBackoff         = 15 * time.Minute
 )

 // Controller serializes provider calls and retains their HTTP 429 cooldown.
 // Its zero value is ready for use. A Controller must not be copied after use.
 // The active caller owns retries; other callers fail fast during known cooldowns.
 type Controller struct {
+	// Configure before first use; these fields must remain immutable afterward.
+	// Zero values preserve unpaced requests and a one-second initial backoff.
+	MinimumInterval time.Duration
+	InitialBackoff  time.Duration
+
 	once   sync.Once
 	active chan struct{}
 	mu     sync.Mutex
 	limit  *RateLimitError
+
+	// Only the acquired caller accesses pacing and consecutive-failure state.
+	lastAttempt     time.Time
+	learnedInterval time.Duration
+	backoff         time.Duration
 }

 // RateLimitError is a safe provider HTTP 429 error. Its message contains only
@@ -35,6 +47,12 @@ type RateLimitError struct {
 	unbounded bool
 }

+// NewError records a trusted provider-specific retry deadline without exposing
+// provider response text. A zero deadline disables automatic retries.
+func NewError(retryAt time.Time) *RateLimitError {
+	return &RateLimitError{next: retryAt, unbounded: retryAt.IsZero()}
+}
+
 func (r *RateLimitError) Error() string {
 	if r.unbounded {
 		return "provider rate limit (HTTP 429): retry time exceeds the supported range; automatic retry disabled"
@@ -128,7 +146,7 @@ func retryLimit(header string, now time.Time, fallback time.Duration) *RateLimit
 // Its per-attempt timeout must not include this controller's retry waiting.
 func (g *Controller) Do(ctx context.Context, attempt func(context.Context) (*http.Response, error), retry bool) (*http.Response, error) {
 	remainingWait := maxRateWait
-	var lastLimit error
+	var lastLimit *RateLimitError
 	canceled := func(err error) error {
 		if lastLimit != nil {
 			return fmt.Errorf("%w: %w", lastLimit, err)
@@ -139,6 +157,31 @@ func (g *Controller) Do(ctx context.Context, attempt func(context.Context) (*htt
 		if err := ctx.Err(); err != nil {
 			return nil, canceled(err)
 		}
+		// Pace all attempts, not just retries, including after successful calls.
+		next := g.lastAttempt.Add(max(g.MinimumInterval, g.learnedInterval))
+		if lastLimit != nil && lastLimit.next.After(next) {
+			next = lastLimit.next
+		}
+		delay := time.Until(next)
+		if lastLimit != nil && (lastLimit.unbounded || delay > remainingWait) {
+			return nil, lastLimit
+		}
+		if delay > 0 {
+			if lastLimit != nil {
+				remainingWait -= delay
+			}
+			timer := time.NewTimer(delay)
+			select {
+			case <-ctx.Done():
+				timer.Stop()
+				return nil, canceled(ctx.Err())
+			case <-timer.C:
+			}
+		}
+		if err := ctx.Err(); err != nil {
+			return nil, canceled(err)
+		}
+		g.lastAttempt = time.Now()
 		resp, err := attempt(ctx)
 		if err != nil {
 			if ctx.Err() != nil {
@@ -150,12 +193,36 @@ func (g *Controller) Do(ctx context.Context, attempt func(context.Context) (*htt
 			return nil, err
 		}
 		if resp.StatusCode != http.StatusTooManyRequests {
+			if resp.StatusCode < http.StatusBadRequest {
+				// Recovery resets consecutive-failure escalation, but never
+				// erases learned spacing: a single success is not a quota reset.
+				g.backoff = 0
+			}
 			return resp, nil
 		}
-		limit := retryLimit(resp.Header.Get("Retry-After"), time.Now(), time.Second<<number)
+		if g.backoff <= 0 {
+			g.backoff = g.InitialBackoff
+			if g.backoff <= 0 {
+				g.backoff = time.Second
+			}
+		} else if g.backoff >= maxBackoff/2 {
+			g.backoff = max(g.backoff, maxBackoff)
+		} else {
+			g.backoff *= 2
+		}
+		fallback := max(g.backoff, g.MinimumInterval, g.learnedInterval)
+		limit := retryLimit(resp.Header.Get("Retry-After"), time.Now(), fallback)
 		g.mu.Lock()
 		g.limit = limit
 		g.mu.Unlock()
+		// Keep the most conservative learned cadence for this controller's
+		// lifetime, capped at 30 seconds. The actual provider deadline is never
+		// capped; persistent failures separately escalate up to 15 minutes.
+		learned := maxLearnedInterval
+		if !limit.unbounded {
+			learned = min(learned, time.Until(limit.next))
+		}
+		g.learnedInterval = max(g.learnedInterval, learned)
 		// Never read or expose provider errors, and release each response before
 		// any sleep or retry. Other responses are processed by the caller.
 		resp.Body.Close()
@@ -163,21 +230,9 @@ func (g *Controller) Do(ctx context.Context, attempt func(context.Context) (*htt
 		if err := ctx.Err(); err != nil {
 			return nil, canceled(err)
 		}
-		delay := time.Until(limit.next)
-		if !retry || number == maxRateAttempts-1 || limit.unbounded || delay > remainingWait {
+		if !retry || number == maxRateAttempts-1 {
 			return nil, limit
 		}
-		if delay <= 0 {
-			continue
-		}
-		remainingWait -= delay
-		timer := time.NewTimer(delay)
-		select {
-		case <-ctx.Done():
-			timer.Stop()
-			return nil, canceled(ctx.Err())
-		case <-timer.C:
-		}
 	}
 	return nil, lastLimit
 }
@@ -0,0 +1,190 @@
+package ratelimit_test
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"net/http"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"finance-duck/internal/ratelimit"
+)
+
+func execute(ctx context.Context, gate *ratelimit.Controller, attempt func(context.Context) (*http.Response, error), retry bool) error {
+	if err := gate.Acquire(ctx); err != nil {
+		return err
+	}
+	defer gate.Release()
+	response, err := gate.Do(ctx, attempt, retry)
+	if response != nil {
+		response.Body.Close()
+	}
+	return err
+}
+
+func response(status int) *http.Response {
+	return &http.Response{StatusCode: status, Header: make(http.Header), Body: io.NopCloser(strings.NewReader(""))}
+}
+
+func TestSuccessfulConcurrentCallsArePaced(t *testing.T) {
+	const interval = 30 * time.Millisecond
+	gate := &ratelimit.Controller{MinimumInterval: interval}
+	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
+	defer cancel()
+	var mu sync.Mutex
+	var arrivals []time.Time
+	attempt := func(context.Context) (*http.Response, error) {
+		mu.Lock()
+		arrivals = append(arrivals, time.Now())
+		mu.Unlock()
+		return response(http.StatusOK), nil
+	}
+	done := make(chan error, 4)
+	for range cap(done) {
+		go func() { done <- execute(ctx, gate, attempt, true) }()
+	}
+	for range cap(done) {
+		if err := <-done; err != nil {
+			t.Fatal(err)
+		}
+	}
+	mu.Lock()
+	defer mu.Unlock()
+	if len(arrivals) != 4 {
+		t.Fatalf("got %d outbound calls, want four", len(arrivals))
+	}
+	for i := 1; i < len(arrivals); i++ {
+		if gap := arrivals[i].Sub(arrivals[i-1]); gap < interval-time.Millisecond {
+			t.Fatalf("successful calls burst: gap %v, minimum %v", gap, interval)
+		}
+	}
+}
+
+func TestRecoveryRetainsLearnedPacing(t *testing.T) {
+	const initial = 20 * time.Millisecond
+	gate := &ratelimit.Controller{InitialBackoff: initial}
+	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
+	defer cancel()
+	var arrivals []time.Time
+	attempt := func(context.Context) (*http.Response, error) {
+		arrivals = append(arrivals, time.Now())
+		if len(arrivals) <= 2 {
+			return response(http.StatusTooManyRequests), nil
+		}
+		return response(http.StatusOK), nil
+	}
+	for range 3 {
+		if err := execute(ctx, gate, attempt, true); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if len(arrivals) != 5 {
+		t.Fatalf("got %d attempts, want two retries and three successes", len(arrivals))
+	}
+	for i := 2; i < len(arrivals); i++ {
+		if gap := arrivals[i].Sub(arrivals[i-1]); gap < 2*initial-time.Millisecond {
+			t.Fatalf("attempt %d discarded learned cadence after recovery: %v", i+1, gap)
+		}
+	}
+}
+
+func TestExhaustionEscalatesAcrossCalls(t *testing.T) {
+	const initial = 10 * time.Millisecond
+	gate := &ratelimit.Controller{InitialBackoff: initial}
+	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
+	defer cancel()
+	var arrivals []time.Time
+	attempt := func(context.Context) (*http.Response, error) {
+		arrivals = append(arrivals, time.Now())
+		return response(http.StatusTooManyRequests), nil
+	}
+	err := execute(ctx, gate, attempt, true)
+	var exhausted *ratelimit.RateLimitError
+	if !errors.As(err, &exhausted) || len(arrivals) != 4 {
+		t.Fatalf("expected exhaustion after four attempts, got %v and %d", err, len(arrivals))
+	}
+	for i := 1; i < len(arrivals); i++ {
+		if gap := arrivals[i].Sub(arrivals[i-1]); gap < (initial<<(i-1))-time.Millisecond {
+			t.Fatalf("retry %d did not escalate: %v", i, gap)
+		}
+	}
+	// A new transaction is eligible only after the fourth attempt's cooldown.
+	time.Sleep(time.Until(exhausted.RetryAt()))
+	err = execute(ctx, gate, attempt, false)
+	var next *ratelimit.RateLimitError
+	if !errors.As(err, &next) || len(arrivals) != 5 {
+		t.Fatalf("eligible successor did not make exactly one attempt: %v, %d", err, len(arrivals))
+	}
+	if delay := next.RetryAt().Sub(arrivals[4]); delay < 16*initial {
+		t.Fatalf("new call reset consecutive-failure backoff: %v, need at least %v", delay, 16*initial)
+	}
+	if err := execute(ctx, gate, attempt, true); !errors.As(err, &next) || len(arrivals) != 5 {
+		t.Fatalf("retained cooldown allowed an early attempt: %v, %d", err, len(arrivals))
+	}
+}
+
+func TestCanceledPacingWaitMakesNoOutboundCall(t *testing.T) {
+	gate := &ratelimit.Controller{MinimumInterval: 100 * time.Millisecond}
+	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
+	defer cancel()
+	calls := 0
+	attempt := func(context.Context) (*http.Response, error) {
+		calls++
+		return response(http.StatusOK), nil
+	}
+	if err := execute(ctx, gate, attempt, true); err != nil {
+		t.Fatal(err)
+	}
+	pacedCtx, pacedCancel := context.WithTimeout(ctx, 10*time.Millisecond)
+	err := execute(pacedCtx, gate, attempt, true)
+	pacedCancel()
+	if !errors.Is(err, context.DeadlineExceeded) || calls != 1 {
+		t.Fatalf("paced cancellation sent a request or lost context identity: %v, calls=%d", err, calls)
+	}
+	if err := execute(ctx, gate, attempt, true); err != nil || calls != 2 {
+		t.Fatalf("cancelled waiter leaked an attempt or blocked its successor: %v, calls=%d", err, calls)
+	}
+}
+
+func TestAutomaticWaitBudgetRetainsLongCooldown(t *testing.T) {
+	gate := &ratelimit.Controller{InitialBackoff: 3 * time.Minute}
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	defer cancel()
+	calls := 0
+	attempt := func(context.Context) (*http.Response, error) {
+		calls++
+		return response(http.StatusTooManyRequests), nil
+	}
+	for range 2 {
+		err := execute(ctx, gate, attempt, true)
+		var limit *ratelimit.RateLimitError
+		if !errors.As(err, &limit) || errors.Is(err, context.DeadlineExceeded) || calls != 1 {
+			t.Fatalf("over-budget cooldown waited or replayed: %v, calls=%d", err, calls)
+		}
+	}
+}
+
+func TestProviderSpecificErrorDoesNotPauseUnrelatedCalls(t *testing.T) {
+	for _, deadline := range []time.Time{time.Now().Add(6 * time.Hour), {}} {
+		gate := &ratelimit.Controller{}
+		err := execute(context.Background(), gate, func(context.Context) (*http.Response, error) {
+			return nil, fmt.Errorf("account quota: %w", ratelimit.NewError(deadline))
+		}, true)
+		var limit *ratelimit.RateLimitError
+		if !errors.As(err, &limit) || !limit.RetryAt().Equal(deadline) {
+			t.Fatalf("provider quota lost its typed retry deadline: %v", err)
+		}
+		calls := 0
+		err = execute(context.Background(), gate, func(context.Context) (*http.Response, error) {
+			calls++
+			return response(http.StatusOK), nil
+		}, true)
+		if err != nil || calls != 1 {
+			t.Fatalf("provider-specific quota poisoned common controller: %v, calls=%d", err, calls)
+		}
+	}
+}