From 247b79114d83e13cdfb2136333a949ff4dbe385b Mon Sep 17 00:00:00 2001 From: Paul Buetow Date: Thu, 11 Jun 2026 00:04:11 +0300 Subject: Add retry and circuit-breaker resilience around LLM HTTP calls (ak0) Wrap the shared provider HTTP choke point (doJSONRequest in llm/util.go, used by openai, openrouter, anthropic and ollama) with two resilience patterns implemented with the standard library only: - Retry with exponential backoff + jitter (resilience.go): 3 attempts by default, retrying transient network errors and retryable HTTP statuses (429 and 5xx). Client errors (4xx) and successes are returned immediately and never retried. Backoff waits are context-aware so cancellation and deadlines are respected; retried response bodies are drained and closed for connection reuse. - Circuit breaker (circuitbreaker.go, own file as it has >3 methods): classic closed/open/half-open breaker that trips after 5 consecutive transient failures and stays open for a 30s cooldown, then allows a single trial probe. Only transient failures count; 4xx never trips it. A nil breaker is a valid no-op. doJSONRequest now delegates to doJSONRequestResilient; the single-attempt primitive is preserved as doJSONRequestOnce. Adds httptest-based unit tests for retry/backoff/breaker logic with no real network calls; new code coverage is >80%. Co-Authored-By: Claude Opus 4.8 --- internal/llm/circuitbreaker.go | 139 ++++++++++++++++++++ internal/llm/circuitbreaker_test.go | 103 +++++++++++++++ internal/llm/resilience.go | 185 +++++++++++++++++++++++++++ internal/llm/resilience_test.go | 246 ++++++++++++++++++++++++++++++++++++ internal/llm/util.go | 11 ++ 5 files changed, 684 insertions(+) create mode 100644 internal/llm/circuitbreaker.go create mode 100644 internal/llm/circuitbreaker_test.go create mode 100644 internal/llm/resilience.go create mode 100644 internal/llm/resilience_test.go diff --git a/internal/llm/circuitbreaker.go b/internal/llm/circuitbreaker.go new file mode 100644 index 0000000..f7014de --- /dev/null +++ b/internal/llm/circuitbreaker.go @@ -0,0 +1,139 @@ +package llm + +import ( + "sync" + "time" +) + +// circuitState models the three states of the classic circuit-breaker pattern. +// +// - closed: normal operation, requests flow through. +// - open: too many consecutive failures occurred; requests are rejected +// immediately for a cooldown window to give the upstream time to recover. +// - halfOpen: the cooldown elapsed; a single trial request is allowed through +// to probe whether the upstream has recovered. +type circuitState int + +const ( + circuitClosed circuitState = iota + circuitOpen + circuitHalfOpen +) + +// circuitBreaker is a small, dependency-free circuit breaker used to protect +// the LLM HTTP call path. It trips open after a configurable number of +// consecutive failures and stays open for a cooldown window, after which it +// allows a single trial ("half-open") request. A success in any state resets +// the breaker to closed. +// +// The breaker only counts failures that the retry layer deemed transient +// (network errors, 5xx, 429). Client errors (4xx) are not failures from the +// breaker's perspective: they indicate a bad request, not an unhealthy +// upstream, so they must not trip the circuit. +// +// A nil *circuitBreaker is a valid no-op breaker: Allow always returns true and +// the record* methods do nothing. This lets callers disable the breaker simply +// by passing nil. +type circuitBreaker struct { + mu sync.Mutex + + // threshold is the number of consecutive failures that trips the breaker. + threshold int + // cooldown is how long the breaker stays open before allowing a trial. + cooldown time.Duration + // now is injectable for deterministic tests; defaults to time.Now. + now func() time.Time + + state circuitState + failures int + openedAt time.Time + probeRunning bool +} + +// newCircuitBreaker returns a breaker that trips after `threshold` consecutive +// transient failures and stays open for `cooldown`. A threshold <= 0 disables +// tripping (the breaker stays closed forever), which callers can use to turn +// the breaker off without special-casing nil. +func newCircuitBreaker(threshold int, cooldown time.Duration) *circuitBreaker { + return &circuitBreaker{ + threshold: threshold, + cooldown: cooldown, + now: time.Now, + state: circuitClosed, + } +} + +// Allow reports whether a request may proceed under the current breaker state. +// When the breaker is open and the cooldown has elapsed it transitions to +// half-open and permits exactly one trial request; concurrent callers during +// half-open are rejected until the trial resolves via recordSuccess/recordFailure. +func (cb *circuitBreaker) Allow() bool { + if cb == nil { + return true + } + cb.mu.Lock() + defer cb.mu.Unlock() + + switch cb.state { + case circuitClosed: + return true + case circuitOpen: + if cb.now().Sub(cb.openedAt) < cb.cooldown { + return false + } + // Cooldown elapsed: move to half-open and allow a single probe. + cb.state = circuitHalfOpen + cb.probeRunning = true + return true + case circuitHalfOpen: + // Only one probe at a time while half-open. + if cb.probeRunning { + return false + } + cb.probeRunning = true + return true + default: + return true + } +} + +// recordSuccess resets the breaker to its healthy (closed) state. A success +// from a half-open probe means the upstream recovered. +func (cb *circuitBreaker) recordSuccess() { + if cb == nil { + return + } + cb.mu.Lock() + defer cb.mu.Unlock() + cb.state = circuitClosed + cb.failures = 0 + cb.probeRunning = false +} + +// recordFailure registers a transient failure. While half-open it re-opens the +// breaker immediately (the probe failed). While closed it trips the breaker +// once the consecutive-failure count reaches the threshold. +func (cb *circuitBreaker) recordFailure() { + if cb == nil { + return + } + cb.mu.Lock() + defer cb.mu.Unlock() + cb.probeRunning = false + + if cb.state == circuitHalfOpen { + cb.trip() + return + } + cb.failures++ + if cb.threshold > 0 && cb.failures >= cb.threshold { + cb.trip() + } +} + +// trip moves the breaker to the open state and stamps the open time. Callers +// must hold cb.mu. +func (cb *circuitBreaker) trip() { + cb.state = circuitOpen + cb.openedAt = cb.now() +} diff --git a/internal/llm/circuitbreaker_test.go b/internal/llm/circuitbreaker_test.go new file mode 100644 index 0000000..17510b3 --- /dev/null +++ b/internal/llm/circuitbreaker_test.go @@ -0,0 +1,103 @@ +package llm + +import ( + "testing" + "time" +) + +func TestCircuitBreaker_NilIsNoOp(t *testing.T) { + var cb *circuitBreaker + if !cb.Allow() { + t.Fatal("nil breaker must allow") + } + // These must not panic on a nil receiver. + cb.recordFailure() + cb.recordSuccess() +} + +func TestCircuitBreaker_TripsAfterThreshold(t *testing.T) { + cb := newCircuitBreaker(3, time.Minute) + for i := 0; i < 2; i++ { + cb.recordFailure() + if !cb.Allow() { + t.Fatalf("breaker should stay closed before threshold (i=%d)", i) + } + } + cb.recordFailure() // third failure trips it + if cb.Allow() { + t.Fatal("breaker should be open after threshold failures") + } +} + +func TestCircuitBreaker_SuccessResets(t *testing.T) { + cb := newCircuitBreaker(2, time.Minute) + cb.recordFailure() + cb.recordSuccess() + cb.recordFailure() // only one failure since reset; should not trip + if !cb.Allow() { + t.Fatal("breaker should remain closed after success reset") + } +} + +func TestCircuitBreaker_HalfOpenAllowsSingleProbe(t *testing.T) { + now := time.Unix(0, 0) + cb := newCircuitBreaker(1, 10*time.Second) + cb.now = func() time.Time { return now } + + cb.recordFailure() // trips (threshold 1) + if cb.Allow() { + t.Fatal("should be open immediately after trip") + } + + // Advance past cooldown: first Allow transitions to half-open and permits one probe. + now = now.Add(11 * time.Second) + if !cb.Allow() { + t.Fatal("expected half-open probe to be allowed") + } + // A second concurrent probe is rejected. + if cb.Allow() { + t.Fatal("second probe must be rejected while half-open") + } +} + +func TestCircuitBreaker_HalfOpenProbeSuccessCloses(t *testing.T) { + now := time.Unix(0, 0) + cb := newCircuitBreaker(1, time.Second) + cb.now = func() time.Time { return now } + + cb.recordFailure() + now = now.Add(2 * time.Second) + if !cb.Allow() { + t.Fatal("expected probe allowed") + } + cb.recordSuccess() + if !cb.Allow() { + t.Fatal("breaker should be closed after successful probe") + } +} + +func TestCircuitBreaker_HalfOpenProbeFailureReopens(t *testing.T) { + now := time.Unix(0, 0) + cb := newCircuitBreaker(1, time.Second) + cb.now = func() time.Time { return now } + + cb.recordFailure() + now = now.Add(2 * time.Second) + if !cb.Allow() { + t.Fatal("expected probe allowed") + } + cb.recordFailure() // probe failed: reopen + if cb.Allow() { + t.Fatal("breaker should reopen after failed probe") + } +} + +func TestCircuitBreaker_ZeroThresholdNeverTrips(t *testing.T) { + cb := newCircuitBreaker(0, time.Minute) + for i := 0; i < 100; i++ { + cb.recordFailure() + } + if !cb.Allow() { + t.Fatal("threshold<=0 should never trip the breaker") + } +} diff --git a/internal/llm/resilience.go b/internal/llm/resilience.go new file mode 100644 index 0000000..4d82153 --- /dev/null +++ b/internal/llm/resilience.go @@ -0,0 +1,185 @@ +package llm + +import ( + "context" + "errors" + "fmt" + "io" + "math/rand" + "net/http" + "time" + + "codeberg.org/snonux/hexai/internal/logging" +) + +// errCircuitOpen is returned when the shared circuit breaker is open and is +// rejecting requests during its cooldown window. It is a sentinel so callers +// and tests can detect breaker-induced failures with errors.Is. +var errCircuitOpen = errors.New("llm: circuit breaker open") + +// retryPolicy describes how transient LLM HTTP failures are retried. The +// defaults are intentionally conservative: a handful of attempts with +// exponential backoff plus jitter, so that brief upstream blips (network +// resets, 5xx, 429 rate limits) are smoothed over without hammering the API or +// stalling interactive use for long. +// +// Backoff for attempt n (0-indexed) is baseDelay * 2^n, capped at maxDelay, +// with up to +/- jitterFraction random jitter to avoid thundering-herd retries. +type retryPolicy struct { + maxAttempts int // total tries including the first; <=1 disables retries + baseDelay time.Duration // delay before the first retry + maxDelay time.Duration // upper bound on any single backoff + jitterFraction float64 // 0..1 fraction of random jitter applied to each delay + + // sleep and rand are injectable seams for deterministic tests. In + // production they default to a context-aware sleep and package rng. + sleep func(ctx context.Context, d time.Duration) error + randFloat func() float64 +} + +// defaultRetryPolicy returns the policy used for all LLM HTTP calls. Three +// total attempts (two retries) with 200ms base backoff keeps recovery fast for +// transient errors while staying well within typical request timeouts. +func defaultRetryPolicy() retryPolicy { + return retryPolicy{ + maxAttempts: 3, + baseDelay: 200 * time.Millisecond, + maxDelay: 2 * time.Second, + jitterFraction: 0.2, + sleep: sleepWithContext, + randFloat: rand.Float64, + } +} + +// sharedBreaker guards the whole LLM HTTP path. It trips after several +// consecutive transient failures and stays open briefly so a single unhealthy +// upstream does not cause every caller to wait out full retry+timeout cycles. +var sharedBreaker = newCircuitBreaker(5, 30*time.Second) + +// sleepWithContext sleeps for d unless ctx is cancelled first, in which case it +// returns ctx.Err(). This keeps backoff waits responsive to cancellation and +// deadlines instead of blocking blindly. +func sleepWithContext(ctx context.Context, d time.Duration) error { + if d <= 0 { + return ctx.Err() + } + t := time.NewTimer(d) + defer t.Stop() + select { + case <-ctx.Done(): + return ctx.Err() + case <-t.C: + return nil + } +} + +// backoffFor computes the (possibly jittered) delay before the given retry +// attempt index (0 = first retry). It applies exponential growth capped at +// maxDelay, then symmetric jitter of +/- jitterFraction. +func (p retryPolicy) backoffFor(attempt int) time.Duration { + d := p.baseDelay + for i := 0; i < attempt && d < p.maxDelay; i++ { + d *= 2 + } + if d > p.maxDelay { + d = p.maxDelay + } + if p.jitterFraction > 0 && p.randFloat != nil { + // Map randFloat()'s [0,1) into [-jitterFraction, +jitterFraction). + delta := (p.randFloat()*2 - 1) * p.jitterFraction + d += time.Duration(float64(d) * delta) + } + if d < 0 { + d = 0 + } + return d +} + +// shouldRetryStatus reports whether an HTTP status code represents a transient +// server-side condition worth retrying. We retry 429 (rate limited) and 5xx +// (server errors). We deliberately do NOT retry other 4xx codes: those are +// client errors (bad request, auth, not found) that will fail identically on +// retry and must surface immediately. +func shouldRetryStatus(status int) bool { + return status == http.StatusTooManyRequests || status >= 500 +} + +// doJSONRequestResilient wraps doJSONRequestOnce with retry-with-backoff and a +// shared circuit breaker. It retries transient network errors and retryable +// HTTP statuses (429/5xx) per the supplied policy, while respecting context +// cancellation/deadlines throughout. Non-retryable responses (including 4xx and +// any success) are returned to the caller unread so providers can parse the +// body as before. +func doJSONRequestResilient(ctx context.Context, httpClient *http.Client, url string, body []byte, headers map[string]string, accept string, policy retryPolicy, breaker *circuitBreaker) (*http.Response, error) { + if !breaker.Allow() { + logging.Logf("llm/resilience ", "%scircuit open: rejecting request to %s%s", logging.AnsiRed, url, logging.AnsiBase) + return nil, errCircuitOpen + } + + attempts := policy.maxAttempts + if attempts < 1 { + attempts = 1 + } + + var lastErr error + for attempt := 0; attempt < attempts; attempt++ { + if err := ctx.Err(); err != nil { + return nil, err + } + resp, retryable, err := attemptJSONRequest(ctx, httpClient, url, body, headers, accept) + if err == nil && !retryable { + breaker.recordSuccess() + return resp, nil + } + lastErr = err + // Sleep before the next attempt unless this was the last one. + if attempt < attempts-1 { + if werr := waitBeforeRetry(ctx, policy, attempt, url, err); werr != nil { + breaker.recordFailure() + return nil, werr + } + } + } + + // All attempts exhausted on transient failures: count it against the breaker. + breaker.recordFailure() + if lastErr == nil { + lastErr = fmt.Errorf("llm: request to %s failed after %d attempts", url, attempts) + } + return nil, lastErr +} + +// attemptJSONRequest performs a single HTTP attempt. It returns retryable=true +// when the caller should retry: either a network/transport error, or a +// retryable status (429/5xx) whose body is drained and closed so the connection +// can be reused. A non-retryable response is returned with its body intact. +func attemptJSONRequest(ctx context.Context, httpClient *http.Client, url string, body []byte, headers map[string]string, accept string) (resp *http.Response, retryable bool, err error) { + resp, err = doJSONRequestOnce(ctx, httpClient, url, body, headers, accept) + if err != nil { + // Context cancellation is not a transient condition: do not retry. + if ctx.Err() != nil { + return nil, false, err + } + return nil, true, err + } + if shouldRetryStatus(resp.StatusCode) { + // Drain and close so the keep-alive connection can be reused, then + // signal a retry via a synthetic error for logging/diagnostics. + _, _ = io.Copy(io.Discard, resp.Body) + _ = resp.Body.Close() + return nil, true, fmt.Errorf("llm: retryable status %d from %s", resp.StatusCode, url) + } + return resp, false, nil +} + +// waitBeforeRetry logs the impending retry and sleeps for the policy backoff, +// returning early if the context is cancelled during the wait. +func waitBeforeRetry(ctx context.Context, policy retryPolicy, attempt int, url string, cause error) error { + delay := policy.backoffFor(attempt) + logging.Logf("llm/resilience ", "retry %d for %s after %s (cause: %v)", attempt+1, url, delay, cause) + sleep := policy.sleep + if sleep == nil { + sleep = sleepWithContext + } + return sleep(ctx, delay) +} diff --git a/internal/llm/resilience_test.go b/internal/llm/resilience_test.go new file mode 100644 index 0000000..4a3efc2 --- /dev/null +++ b/internal/llm/resilience_test.go @@ -0,0 +1,246 @@ +package llm + +import ( + "context" + "errors" + "io" + "net/http" + "net/http/httptest" + "sync/atomic" + "testing" + "time" +) + +// testPolicy returns a retry policy with deterministic, instant backoff and no +// jitter so tests are fast and reproducible. +func testPolicy(maxAttempts int) retryPolicy { + return retryPolicy{ + maxAttempts: maxAttempts, + baseDelay: time.Millisecond, + maxDelay: time.Millisecond, + jitterFraction: 0, + sleep: func(ctx context.Context, d time.Duration) error { return ctx.Err() }, + randFloat: func() float64 { return 0.5 }, + } +} + +func TestShouldRetryStatus(t *testing.T) { + cases := map[int]bool{ + 200: false, 201: false, + 400: false, 401: false, 404: false, + 429: true, + 500: true, 502: true, 503: true, + } + for status, want := range cases { + if got := shouldRetryStatus(status); got != want { + t.Errorf("shouldRetryStatus(%d)=%v want %v", status, got, want) + } + } +} + +func TestResilient_RetriesThenSucceeds(t *testing.T) { + var calls int32 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if atomic.AddInt32(&calls, 1) < 3 { + w.WriteHeader(http.StatusInternalServerError) + _, _ = io.WriteString(w, "boom") + return + } + w.WriteHeader(http.StatusOK) + _, _ = io.WriteString(w, "ok") + })) + defer srv.Close() + + resp, err := doJSONRequestResilient(context.Background(), srv.Client(), srv.URL, []byte("{}"), nil, "", testPolicy(3), nil) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + t.Fatalf("status=%d want 200", resp.StatusCode) + } + if got := atomic.LoadInt32(&calls); got != 3 { + t.Fatalf("expected 3 calls, got %d", got) + } +} + +func TestResilient_NoRetryOn4xx(t *testing.T) { + var calls int32 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + atomic.AddInt32(&calls, 1) + w.WriteHeader(http.StatusBadRequest) + })) + defer srv.Close() + + resp, err := doJSONRequestResilient(context.Background(), srv.Client(), srv.URL, []byte("{}"), nil, "", testPolicy(3), nil) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusBadRequest { + t.Fatalf("status=%d want 400", resp.StatusCode) + } + if got := atomic.LoadInt32(&calls); got != 1 { + t.Fatalf("4xx must not be retried: got %d calls", got) + } +} + +func TestResilient_ExhaustsRetries(t *testing.T) { + var calls int32 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + atomic.AddInt32(&calls, 1) + w.WriteHeader(http.StatusServiceUnavailable) + })) + defer srv.Close() + + _, err := doJSONRequestResilient(context.Background(), srv.Client(), srv.URL, []byte("{}"), nil, "", testPolicy(3), nil) + if err == nil { + t.Fatal("expected error after exhausting retries") + } + if got := atomic.LoadInt32(&calls); got != 3 { + t.Fatalf("expected 3 attempts, got %d", got) + } +} + +func TestResilient_NetworkErrorRetries(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {})) + url := srv.URL + srv.Close() // closed server: connection attempts fail at the transport layer + + _, err := doJSONRequestResilient(context.Background(), http.DefaultClient, url, []byte("{}"), nil, "", testPolicy(2), nil) + if err == nil { + t.Fatal("expected network error") + } +} + +func TestResilient_ContextCancelStopsRetries(t *testing.T) { + var calls int32 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + atomic.AddInt32(&calls, 1) + w.WriteHeader(http.StatusInternalServerError) + })) + defer srv.Close() + + ctx, cancel := context.WithCancel(context.Background()) + // Cancel during the backoff sleep after the first failed attempt. + policy := testPolicy(5) + policy.sleep = func(c context.Context, d time.Duration) error { + cancel() + return c.Err() + } + + _, err := doJSONRequestResilient(ctx, srv.Client(), srv.URL, []byte("{}"), nil, "", policy, nil) + if !errors.Is(err, context.Canceled) { + t.Fatalf("expected context.Canceled, got %v", err) + } + if got := atomic.LoadInt32(&calls); got != 1 { + t.Fatalf("expected 1 attempt before cancel, got %d", got) + } +} + +func TestResilient_AlreadyCancelledContext(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {})) + defer srv.Close() + ctx, cancel := context.WithCancel(context.Background()) + cancel() + if _, err := doJSONRequestResilient(ctx, srv.Client(), srv.URL, []byte("{}"), nil, "", testPolicy(3), nil); !errors.Is(err, context.Canceled) { + t.Fatalf("expected context.Canceled, got %v", err) + } +} + +func TestResilient_CircuitOpenRejects(t *testing.T) { + cb := newCircuitBreaker(1, time.Hour) + cb.recordFailure() // trips immediately (threshold 1) + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + t.Fatal("server should not be reached while circuit is open") + })) + defer srv.Close() + + _, err := doJSONRequestResilient(context.Background(), srv.Client(), srv.URL, []byte("{}"), nil, "", testPolicy(3), cb) + if !errors.Is(err, errCircuitOpen) { + t.Fatalf("expected errCircuitOpen, got %v", err) + } +} + +func TestResilient_BreakerTripsAfterExhaustedRetries(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusInternalServerError) + })) + defer srv.Close() + + cb := newCircuitBreaker(1, time.Hour) // a single exhausted run trips it + if _, err := doJSONRequestResilient(context.Background(), srv.Client(), srv.URL, []byte("{}"), nil, "", testPolicy(2), cb); err == nil { + t.Fatal("expected error") + } + // Next call must be rejected by the now-open breaker. + if _, err := doJSONRequestResilient(context.Background(), srv.Client(), srv.URL, []byte("{}"), nil, "", testPolicy(2), cb); !errors.Is(err, errCircuitOpen) { + t.Fatalf("expected errCircuitOpen on second call, got %v", err) + } +} + +func TestBackoffFor_ExponentialCappedNoJitter(t *testing.T) { + p := retryPolicy{baseDelay: 100 * time.Millisecond, maxDelay: 400 * time.Millisecond} + want := []time.Duration{ + 100 * time.Millisecond, + 200 * time.Millisecond, + 400 * time.Millisecond, + 400 * time.Millisecond, // capped + } + for i, w := range want { + if got := p.backoffFor(i); got != w { + t.Errorf("backoffFor(%d)=%s want %s", i, got, w) + } + } +} + +func TestBackoffFor_JitterWithinBounds(t *testing.T) { + p := retryPolicy{ + baseDelay: 100 * time.Millisecond, + maxDelay: time.Second, + jitterFraction: 0.5, + randFloat: func() float64 { return 1.0 }, // max positive jitter + } + got := p.backoffFor(0) + // 100ms + 50% = 150ms. + if got != 150*time.Millisecond { + t.Fatalf("got %s want 150ms", got) + } + p.randFloat = func() float64 { return 0.0 } // max negative jitter + if got := p.backoffFor(0); got != 50*time.Millisecond { + t.Fatalf("got %s want 50ms", got) + } +} + +func TestSleepWithContext_ReturnsOnDeadline(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + cancel() + if err := sleepWithContext(ctx, time.Hour); !errors.Is(err, context.Canceled) { + t.Fatalf("expected canceled, got %v", err) + } + // Zero delay returns immediately with ctx.Err() (nil here). + if err := sleepWithContext(context.Background(), 0); err != nil { + t.Fatalf("expected nil for zero delay, got %v", err) + } + // Positive delay elapses normally. + if err := sleepWithContext(context.Background(), time.Millisecond); err != nil { + t.Fatalf("unexpected error: %v", err) + } +} + +func TestResilient_DefaultPolicyDisabledRetriesWhenSingleAttempt(t *testing.T) { + var calls int32 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + atomic.AddInt32(&calls, 1) + w.WriteHeader(http.StatusInternalServerError) + })) + defer srv.Close() + + // maxAttempts <= 1 means no retries: exactly one attempt. + if _, err := doJSONRequestResilient(context.Background(), srv.Client(), srv.URL, []byte("{}"), nil, "", testPolicy(0), nil); err == nil { + t.Fatal("expected error") + } + if got := atomic.LoadInt32(&calls); got != 1 { + t.Fatalf("expected exactly 1 attempt, got %d", got) + } +} diff --git a/internal/llm/util.go b/internal/llm/util.go index b6e2adc..fe99467 100644 --- a/internal/llm/util.go +++ b/internal/llm/util.go @@ -13,7 +13,18 @@ import ( // small helper to keep return type consistent func nilStringErr(msg string) (string, error) { return "", errors.New(msg) } +// doJSONRequest is the shared entry point for provider HTTP calls. It applies +// the default retry-with-backoff policy and the shared circuit breaker so that +// transient upstream failures (network resets, 429, 5xx) are smoothed over, +// while client errors (4xx) and successes are returned immediately. The body +// bytes are passed by value so each retry can rebuild a fresh request. func doJSONRequest(ctx context.Context, httpClient *http.Client, url string, body []byte, headers map[string]string, accept string) (*http.Response, error) { + return doJSONRequestResilient(ctx, httpClient, url, body, headers, accept, defaultRetryPolicy(), sharedBreaker) +} + +// doJSONRequestOnce performs exactly one HTTP POST with the given JSON body and +// headers. It is the single-attempt primitive used by the resilience layer. +func doJSONRequestOnce(ctx context.Context, httpClient *http.Client, url string, body []byte, headers map[string]string, accept string) (*http.Response, error) { req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(body)) if err != nil { return nil, err -- cgit v1.2.3