From 722e6f3d291fbfa399f7c973a897ed98c60cc083 Mon Sep 17 00:00:00 2001 From: Lukasz Raczylo Date: Thu, 20 Aug 2026 17:34:52 +0100 Subject: [PATCH] fix(retry): retries-after-first semantics, ctx-aware backoff, hardened classification Critical: attempts now counts retries performed AFTER the first try (default 1 -> two tries total), not the total try count. The prior reading made retry:{enabled:true}'s own documented default perform zero retries. (*retryPolicy).totalTries() is the single conversion point, clamping any attempts<1 to 1 so do()'s loop bound can never compute <=0 and skip calling call() entirely (no (nil,nil) return path). Backoff waits are now ctx-aware: sleepFn func(time.Duration) is replaced by waitFn func(context.Context, time.Duration) bool, whose production implementation (defaultWaitFn) selects between time.After(d) and ctx.Done(). A context canceled mid-wait aborts the remaining retries immediately instead of blocking out the full backoff. Also: classify a request-build failure (invalid method/URL) as non-transient via errRequestBuildFailed, since retrying an identically malformed request only wastes attempts; drop the drain cap to a dedicated 64KiB constant instead of the 32MiB response cap; fix a t.Fatal call from a non-test goroutine in the mid-stream-death test; drop the misleading omitempty on Config.Retry (a struct value is never 'empty' to encoding/json); document the Retry-After >2s-falls-back-to- backoff behavior and cover it with a direct table test. README: add the retry config table, a RetryConfig field reference, and a Retry section documenting the semantics. examples/kubernetes.yaml gains a retry block alongside its other optional sections. Gate: go test -race, gofmt, vet, yaegi-check, golangci-lint (incl a fieldalignment fix on two new test structs) all clean. 295 subtests pass, 0 failed. integration/ untouched. --- README.md | 48 ++++ examples/kubernetes.yaml | 4 + llmgateway.go | 38 +++- providers.go | 12 +- retry.go | 167 ++++++++++---- retry_test.go | 468 +++++++++++++++++++++++++++++++-------- 6 files changed, 587 insertions(+), 150 deletions(-) diff --git a/README.md b/README.md index 40f1fc5..e158ec4 100644 --- a/README.md +++ b/README.md @@ -167,6 +167,7 @@ config accepts them as YAML, which decodes to the same JSON shape. | `agents` | `map[string]AgentConfig` | `{}` | A2A agent registry; see [MCP/A2A](#mcp-and-a2a). | | `users` | `UsersConfig` | — | Inline and/or file-backed API-key holders. With none configured, every request is unauthenticated and gets 401. | | `redis` | `RedisConfig` | — | Distributed limit-counter backend. Omitted means in-process counters only (per-replica, approximate across multiple Traefik instances). | +| `retry` | `RetryConfig` | `{}` (disabled) | Same-provider retry for transient upstream failures — see [Retry](#retry). Omitted or `enabled: false` means no retry: every request makes exactly one upstream attempt, byte-identical to a gateway built before this field existed. | | `passthroughUnknown` | `bool` | `false` | `false`: a request matching none of the plugin's routes gets a 404 JSON envelope. `true`: it falls through to the router's own backing service. | Provider, `mcpServers`, and `agents` map **keys** (names) must match @@ -244,6 +245,14 @@ own is governed purely by their group's — see | `db` | `int` | `0` | Must be `≥ 0`. Sent via `SELECT` on every new connection. | | `failOpen` | `*bool` | `true` | `true`: a Redis error switches that operation to the in-process fallback store and logs (rate-limited to once per 30s). `false`: a Redis error returns 503 instead of enforcing against a non-shared fallback. | +### `RetryConfig` + +| Field | Type | Default | Semantics | +|---|---|---|---| +| `enabled` | `bool` | `false` | `false`: no retry, one upstream attempt per request. `true`: retry per [Retry](#retry) below. | +| `attempts` | `int` | `1` | Retries performed **after** the first try, not the total try count. `1` (default) allows one retry — two tries total. The maximum, `3`, allows three retries — four tries total. A value outside `1`-`3` is a construction error. | +| `backoff` | `string` (Go duration) | `250ms` | Base wait before the first retry; doubles on each further retry, capped at 2s per wait. Invalid duration string is a construction error. | + ### `ModelPricing` | Field | Type | Semantics | @@ -374,6 +383,45 @@ kept — a bad edit to the file never breaks already-authenticated traffic. across several, since each replica counts independently. Exact limits across multiple Traefik replicas require the shared Redis. +## Retry + +Same-provider retry for a transient upstream failure — no cross-provider +failover. Off by default (`retry.enabled: false`); every config that +predates this field keeps making exactly one upstream attempt per +request. + +- **Scope**: every adapter upstream call — chat, embeddings, model + listing — across all three provider types. Native passthrough + (`/{provider}/...`) and the MCP/A2A proxy are raw reverse proxies and + are never retried; the client owns retry semantics there. +- **Attempts is retries, not tries**: `attempts` counts retries + performed **after** the first try. The default, `1`, allows one retry + — two tries total, not one. The maximum, `3`, allows three retries — + four tries total. +- **Transient classification**: a connection error (dial/EOF/reset), HTTP + 429, or HTTP 5xx. Never a non-429 4xx, a translation error, a request + that failed to build in the first place (a malformed method or URL + fails identically on every attempt), or a context + cancellation/deadline. +- **Zero-bytes rule**: retry only runs before any response byte reaches + the client. A non-streaming call may retry its whole exchange. A + streaming call retries only the initial request, up to the point the + upstream's headers and status arrive — once the first chunk is + forwarded, a stream that then dies mid-flight is not retried, since the + client already holds partial data. +- **Waiting**: a 429 response carrying a `Retry-After` header that parses + as a non-negative whole-second count no greater than 2s waits exactly + that long. Every other case — no header, an unparseable value (for + example the HTTP-date form), a negative value, a value over 2s, or a + `Retry-After` on a non-429 5xx — falls back to exponential backoff + (`backoff × 2^(attempt-1)`), capped at 2s per wait either way. A + context canceled or timed out during a wait aborts the remaining + retries immediately, rather than blocking out the wait in full. +- **Accounting**: only the final attempt's usage is recorded — a failed + attempt's body is drained and discarded, never parsed for usage, so it + is never double-counted. Request counters still increment once per + client request, unchanged. + ## Unified vs. passthrough | | Unified (`/v1/...`) | Native passthrough (`/{provider}/...`) | diff --git a/examples/kubernetes.yaml b/examples/kubernetes.yaml index 6acc4f7..765cf67 100644 --- a/examples/kubernetes.yaml +++ b/examples/kubernetes.yaml @@ -90,6 +90,10 @@ spec: address: redis.shared-resources.svc:6379 password: "env:REDIS_PASSWORD" failOpen: true + retry: + enabled: true + attempts: 2 # retries after the first try: 3 tries total + backoff: 250ms pricing: gpt-5-mini: {inputPerM: 0.25, outputPerM: 2.0} mcpServers: diff --git a/llmgateway.go b/llmgateway.go index 1495ad1..07a6ec8 100644 --- a/llmgateway.go +++ b/llmgateway.go @@ -15,15 +15,21 @@ import ( // the middleware's YAML/testData. Every limit or list field follows the // convention: zero/empty/omitted means unlimited/all. type Config struct { - Providers map[string]*ProviderConfig `json:"providers,omitempty"` - Groups map[string]*GroupConfig `json:"groups,omitempty"` - Pricing map[string]*ModelPricing `json:"pricing,omitempty"` - MCPServers map[string]*TargetConfig `json:"mcpServers,omitempty"` - Agents map[string]*AgentConfig `json:"agents,omitempty"` - Users *UsersConfig `json:"users,omitempty"` - Redis *RedisConfig `json:"redis,omitempty"` - Retry RetryConfig `json:"retry,omitempty"` - PassthroughUnknown bool `json:"passthroughUnknown,omitempty"` + Providers map[string]*ProviderConfig `json:"providers,omitempty"` + Groups map[string]*GroupConfig `json:"groups,omitempty"` + Pricing map[string]*ModelPricing `json:"pricing,omitempty"` + MCPServers map[string]*TargetConfig `json:"mcpServers,omitempty"` + Agents map[string]*AgentConfig `json:"agents,omitempty"` + Users *UsersConfig `json:"users,omitempty"` + Redis *RedisConfig `json:"redis,omitempty"` + // Retry is a struct value, not a pointer, because its own Enabled + // field is the on/off signal (unlike Redis/Users, where the block's + // mere presence is the signal) — so its tag omits "omitempty": + // encoding/json never treats a struct value as "empty" regardless of + // its fields, so "omitempty" here would be a no-op that misleadingly + // implies otherwise. + Retry RetryConfig `json:"retry"` + PassthroughUnknown bool `json:"passthroughUnknown,omitempty"` } // ProviderConfig describes one upstream LLM provider. @@ -85,9 +91,17 @@ type RedisConfig struct { // Attempts and Backoff are validated (and defaulted, when left zero) by // newRetryPolicy (retry.go) only when Enabled is true. type RetryConfig struct { - Backoff string `json:"backoff,omitempty"` - Attempts int `json:"attempts,omitempty"` - Enabled bool `json:"enabled,omitempty"` + // Backoff is the base wait before the first retry (a Go duration + // string, e.g. "250ms"); it doubles on each further retry, capped at + // 2s per wait. Defaults to "250ms" when Enabled and left empty. + Backoff string `json:"backoff,omitempty"` + // Attempts is the number of retries performed AFTER the first try — + // not the total try count. 1 (the default, applied when Enabled and + // left at 0) allows one retry: two tries total. The maximum, 3, + // allows three retries: four tries total. A value outside 1..3 is a + // construction error. + Attempts int `json:"attempts,omitempty"` + Enabled bool `json:"enabled,omitempty"` } // ModelPricing overrides the built-in per-model price table. diff --git a/providers.go b/providers.go index b78bc37..3873fbf 100644 --- a/providers.go +++ b/providers.go @@ -85,6 +85,16 @@ func newProviderHTTPError(resp *http.Response) *providerHTTPError { // with a non-2xx status. var errUpstream = errors.New("llmgateway: upstream request failed") +// errRequestBuildFailed marks an upstreamJSON failure that happened +// before any network I/O — http.NewRequestWithContext rejected the +// method or URL. It is chained alongside errUpstream (Go's multi-%w), so +// an existing errors.Is(err, errUpstream) check still matches, while +// retryPolicy's isTransient (retry.go) checks specifically for this +// sentinel to treat it as non-transient: retrying an identically +// malformed request produces the identical failure every time, so a +// retry only wastes attempts. +var errRequestBuildFailed = errors.New("llmgateway: build upstream request failed") + // providerAdapter is the gateway's uniform interface over one configured // upstream LLM provider. Task 9 (anthropic) and Task 10 (gemini) implement // it alongside this task's openai-type adapter; Tasks 11-12 wire a built @@ -188,7 +198,7 @@ func upstreamJSON(ctx context.Context, client *http.Client, method, url string, } req, err := http.NewRequestWithContext(ctx, method, url, r) if err != nil { - return nil, fmt.Errorf("%w: build request: %w", errUpstream, err) + return nil, fmt.Errorf("%w: %w: build request: %w", errUpstream, errRequestBuildFailed, err) } for k, vs := range hdr { for _, v := range vs { diff --git a/retry.go b/retry.go index ea4b1cf..d9abdd3 100644 --- a/retry.go +++ b/retry.go @@ -13,13 +13,18 @@ import ( // defaultRetryAttempts and defaultRetryBackoff are the values // newRetryPolicy applies when a Retry config block is Enabled but leaves // Attempts or Backoff at its zero value — spec §1's stated defaults. +// defaultRetryAttempts counts RETRIES performed after the initial try +// (see retryPolicy's doc comment), so the default of 1 means one retry — +// two tries total — not one try total. const ( defaultRetryAttempts = 1 defaultRetryBackoff = "250ms" ) // maxRetryAttempts is the highest RetryConfig.Attempts newRetryPolicy -// accepts — spec §1's "attempts: int (default 1, max 3)". +// accepts — spec §1's "attempts: int (default 1, max 3)". Like +// defaultRetryAttempts, this counts retries after the initial try: the +// maximum makes four tries total, not three. const maxRetryAttempts = 3 // maxRetryWait caps every wait retryPolicy.do performs between attempts, @@ -27,32 +32,91 @@ const maxRetryAttempts = 3 // Retry-After header — spec §1's "capped at 2s per wait". const maxRetryWait = 2 * time.Second -// retryPolicy implements spec §1's same-provider retry: at most -// p.attempts total tries per upstream call, exponential backoff between -// them (capped at maxRetryWait), honoring a qualifying 429 Retry-After -// header outright. A nil *retryPolicy, or one with enabled=false, makes -// do try call exactly once — the same shape every v0.1 adapter -// constructor leaves its adapter's retry field in, since none of them set -// it. sleepFn is a field (not a direct time.Sleep call) so tests can -// inject a non-blocking spy instead of paying real wall-clock time. +// retryDrainLimit caps how much of a retried-away response's body +// drainAndClose reads before closing it. It is generous enough to let +// the underlying connection settle for reuse, but far below +// maxResponseBytes (32MiB): a failed attempt's body is discarded, never +// parsed, so there is no reason to pay for reading a large one in full. +const retryDrainLimit = 64 << 10 + +// retryPolicy implements spec §1's same-provider retry. +// +// attempts counts RETRIES performed after the initial try, not the +// total try count — ruling (controller review, 2026-08-20): the earlier +// "attempts = total tries" reading made the documented default of 1 +// perform zero retries, silently defeating retry:{enabled:true}'s own +// defaults. With this policy, attempts=1 (the default) means one retry — +// two tries total; attempts=3 (the max) means three retries — four +// tries total. totalTries() is the single place that turns attempts +// into the actual try count do's loop uses. +// +// A nil *retryPolicy, or one with enabled=false, makes do try call +// exactly once — the same shape every v0.1 adapter constructor leaves +// its adapter's retry field in, since none of them set it. waitFn is a +// field (not a direct call to a wait implementation) so tests can inject +// a non-blocking spy instead of paying real wall-clock time; the +// production default (defaultWaitFn) is ctx-aware. type retryPolicy struct { - sleepFn func(time.Duration) + waitFn func(context.Context, time.Duration) bool backoff time.Duration attempts int enabled bool } +// totalTries returns how many times do should call call() in total: 1 +// for a nil or disabled policy (no retry — v0.1 behavior), or +// p.attempts+1 for an enabled one, per retryPolicy's own doc comment. A +// malformed attempts value below 1 — reachable only by constructing a +// *retryPolicy directly rather than through newRetryPolicy, which +// already rejects it — is clamped to 1 retry rather than trusted as-is, +// so this can never return less than 1: a totalTries of 0 or less would +// make do's loop skip calling call() entirely and return the disallowed +// (nil, nil) shape. +func (p *retryPolicy) totalTries() int { + if p == nil || !p.enabled { + return 1 + } + retries := p.attempts + if retries < 1 { + retries = 1 + } + return retries + 1 +} + +// defaultWaitFn is the production (*retryPolicy).waitFn: it blocks for +// d, or returns early the moment ctx is done, whichever comes first. A +// context canceled, or a deadline reached, while do is backing off +// aborts the remaining wait promptly instead of blocking out the full +// duration for a retry the caller no longer wants — this also bounds how +// long newGateway's synchronous warmFill discovery fill can be held up +// by a retrying listModels call. Returns true when the wait completed +// normally (do should proceed with the next attempt), false when ctx +// ended it early (do should stop retrying and return the last result). +func defaultWaitFn(ctx context.Context, d time.Duration) bool { + select { + case <-time.After(d): + return true + case <-ctx.Done(): + return false + } +} + // newRetryPolicy validates rc and returns the retryPolicy buildAdapters // attaches to every adapter it constructs. Disabled config (the v0.1 // default) skips validation and defaulting entirely — an unset // Attempts/Backoff on a disabled block is never an error, since neither -// value is ever read. Enabled config defaults an unset Attempts to -// defaultRetryAttempts and an unset Backoff to defaultRetryBackoff before -// validating: Attempts must fall within 1..maxRetryAttempts, and Backoff -// must parse via time.ParseDuration. +// value is ever read. +// +// rc.Attempts is the number of retries performed after the initial try, +// not the total try count (see retryPolicy's doc comment): the default, +// applied when rc.Attempts is left at its zero value, is 1 retry (two +// tries total); the maximum, maxRetryAttempts, is 3 retries (four tries +// total). A value outside 1..maxRetryAttempts is a config error. +// rc.Backoff defaults to defaultRetryBackoff when left empty, and must +// parse via time.ParseDuration either way. func newRetryPolicy(rc RetryConfig) (*retryPolicy, error) { if !rc.Enabled { - return &retryPolicy{enabled: false, sleepFn: time.Sleep}, nil + return &retryPolicy{enabled: false, waitFn: defaultWaitFn}, nil } attempts := rc.Attempts @@ -60,7 +124,7 @@ func newRetryPolicy(rc RetryConfig) (*retryPolicy, error) { attempts = defaultRetryAttempts } if attempts < 1 || attempts > maxRetryAttempts { - return nil, fmt.Errorf("llmgateway: retry.attempts must be between 1 and %d, got %d", maxRetryAttempts, attempts) + return nil, fmt.Errorf("llmgateway: retry.attempts (retries after the first try) must be between 1 and %d, got %d", maxRetryAttempts, attempts) } backoffStr := rc.Backoff @@ -72,19 +136,25 @@ func newRetryPolicy(rc RetryConfig) (*retryPolicy, error) { return nil, fmt.Errorf("llmgateway: retry.backoff: %w", err) } - return &retryPolicy{enabled: true, attempts: attempts, backoff: backoff, sleepFn: time.Sleep}, nil + return &retryPolicy{enabled: true, attempts: attempts, backoff: backoff, waitFn: defaultWaitFn}, nil } // isTransient reports whether resp/err — the result of one upstream // attempt, mutually exclusive per http.Client's own contract (a non-nil // err always means resp is nil) — is a failure retryPolicy.do should -// retry: any error (a dial/EOF/reset failure reaching upstreamJSON's -// client.Do) other than a context cancellation or deadline, or a -// response carrying HTTP 429 or a 5xx status. Every other outcome — -// success, or a non-429 4xx — is not transient, per spec §1's "Never on -// 4xx≠429, translate errors, or context cancellation". +// retry: any error reaching upstreamJSON's client.Do (dial/EOF/reset), +// other than a context cancellation, a deadline, or a failure to build +// the outgoing *http.Request in the first place (errRequestBuildFailed — +// a malformed method or URL fails identically on every attempt, so +// retrying it only wastes attempts) — or a response carrying HTTP 429 or +// a 5xx status. Every other outcome — success, or a non-429 4xx — is not +// transient, per spec §1's "Never on 4xx≠429, translate errors, or +// context cancellation". func isTransient(resp *http.Response, err error) bool { if err != nil { + if errors.Is(err, errRequestBuildFailed) { + return false + } return !errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded) } if resp == nil { @@ -95,11 +165,15 @@ func isTransient(resp *http.Response, err error) bool { // waitBefore returns how long do should wait before the attempt after // attempt (1-based) — the one that just produced resp (nil on a network -// error). A 429 carrying a Retry-After header parseable as whole seconds, -// and no greater than maxRetryWait, wins outright: the upstream said -// exactly how long to back off. Every other transient failure falls back -// to exponential backoff (p.backoff * 2^(attempt-1)), capped at -// maxRetryWait either way. +// error). Only a 429 response carrying a Retry-After header that parses +// as a non-negative whole-second count no greater than maxRetryWait wins +// outright: the upstream said exactly how long to back off. Every other +// case falls back to exponential backoff (p.backoff * 2^(attempt-1)), +// capped at maxRetryWait — that covers a 429 with no Retry-After header, +// one that fails to parse as whole seconds (for example the HTTP-date +// form), one that parses negative, one whose value exceeds maxRetryWait, +// a Retry-After header present on a non-429 5xx (ignored: Retry-After is +// only meaningful on 429 here), and every network-error retry. func (p *retryPolicy) waitBefore(attempt int, resp *http.Response) time.Duration { if resp != nil && resp.StatusCode == http.StatusTooManyRequests { if ra := resp.Header.Get("Retry-After"); ra != "" { @@ -119,47 +193,44 @@ func (p *retryPolicy) waitBefore(attempt int, resp *http.Response) time.Duration } // drainAndClose discards a retried-away response's remaining body -// (capped at maxResponseBytes, defensively) and closes it, so the -// underlying connection is eligible for reuse on the next attempt instead -// of being forced closed by an un-drained body. +// (capped at retryDrainLimit, defensively) and closes it, so the +// underlying connection is eligible for reuse on the next attempt +// instead of being forced closed by an un-drained body. func drainAndClose(resp *http.Response) { if resp == nil || resp.Body == nil { return } - _, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, maxResponseBytes)) + _, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, retryDrainLimit)) _ = resp.Body.Close() } // do executes call, retrying while isTransient(resp, err) reports a -// transient failure, up to p.attempts total tries. A nil p, or one with +// transient failure, up to totalTries() tries in total (see its doc for +// the retries-vs-total-tries distinction). A nil p, or one with // enabled=false, tries call exactly once — the disabled/unwired shape. // Between attempts it drains and closes the failed attempt's response -// body, then waits per waitBefore via p.sleepFn. do never starts a wait -// once ctx is already done: at that point it returns the last attempt's -// (resp, err) as-is, the same value it would return had attempts run out -// naturally. do returns the final attempt's result either way — success, -// a still-transient failure once attempts are exhausted, or a -// non-transient failure on any attempt. +// body, then waits per waitBefore via p.waitFn — ctx-aware, so a context +// canceled or timed out during that wait aborts the remaining attempts +// immediately rather than blocking out the wait in full. do returns the +// last attempt's (resp, err) as-is either way: on success, on a +// still-transient failure once tries are exhausted, on a non-transient +// failure at any attempt, or on a wait ctx aborted early. func (p *retryPolicy) do(ctx context.Context, call func() (*http.Response, error)) (*http.Response, error) { - attempts := 1 - if p != nil && p.enabled { - attempts = p.attempts - } + tries := p.totalTries() var resp *http.Response var err error - for attempt := 1; attempt <= attempts; attempt++ { + for attempt := 1; attempt <= tries; attempt++ { resp, err = call() - if !isTransient(resp, err) || attempt == attempts { - return resp, err - } - if ctx.Err() != nil { + if !isTransient(resp, err) || attempt == tries { return resp, err } wait := p.waitBefore(attempt, resp) drainAndClose(resp) - p.sleepFn(wait) + if !p.waitFn(ctx, wait) { + return resp, err + } } return resp, err } diff --git a/retry_test.go b/retry_test.go index 11e5987..689bea7 100644 --- a/retry_test.go +++ b/retry_test.go @@ -2,6 +2,7 @@ package traefikllmgateway import ( "context" + "encoding/json" "errors" "fmt" "io" @@ -23,10 +24,44 @@ func fakeResp(status int, hdr http.Header) *http.Response { return &http.Response{StatusCode: status, Header: hdr, Body: io.NopCloser(strings.NewReader("body"))} } +// retryAfterResp builds a fakeResp carrying a Retry-After header set to +// the raw value given — deliberately untyped (a caller passes a bogus +// or negative string too) so waitBefore's parsing branches can be +// exercised directly. +func retryAfterResp(status int, retryAfter string) *http.Response { + h := http.Header{} + h.Set("Retry-After", retryAfter) + return fakeResp(status, h) +} + +// spyWait returns a waitFn that records every requested duration +// instead of actually waiting, and always reports the wait completed +// (true), so retry-wait tests run instantly and can assert on the exact +// durations retryPolicy.do computed. +func spyWait(waits *[]time.Duration) func(context.Context, time.Duration) bool { + return func(_ context.Context, d time.Duration) bool { + *waits = append(*waits, d) + return true + } +} + +// callSequence returns a call func for retryPolicy.do that returns +// results[i] on its i-th invocation (0-indexed), and counts invocations +// into calls. +func callSequence(calls *int32, results ...func() (*http.Response, error)) func() (*http.Response, error) { + return func() (*http.Response, error) { + n := int(atomic.AddInt32(calls, 1)) - 1 + if n >= len(results) { + n = len(results) - 1 + } + return results[n]() + } +} + // TestIsTransient covers the transience classification retryPolicy.do // relies on: net errors (wrapping errUpstream) and HTTP 429/5xx are -// transient; a successful or non-429 4xx status, and a context -// cancellation/deadline error, are not. +// transient; a successful or non-429 4xx status, a context +// cancellation/deadline error, and a request-build failure, are not. func TestIsTransient(t *testing.T) { tests := []struct { resp *http.Response @@ -43,6 +78,7 @@ func TestIsTransient(t *testing.T) { {name: "network error", err: fmt.Errorf("%w: dial tcp: connection reset", errUpstream), want: true}, {name: "context canceled", err: fmt.Errorf("%w: %w", errUpstream, context.Canceled), want: false}, {name: "context deadline exceeded", err: fmt.Errorf("%w: %w", errUpstream, context.DeadlineExceeded), want: false}, + {name: "request build failure", err: fmt.Errorf("%w: %w: build request: %w", errUpstream, errRequestBuildFailed, errors.New("invalid method")), want: false}, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { @@ -56,7 +92,10 @@ func TestIsTransient(t *testing.T) { // TestNewRetryPolicy_Disabled asserts the zero-value RetryConfig (the // v0.1-preserving default) produces a policy with enabled=false and no // validation error, even though Attempts/Backoff are left at their zero -// values. +// values, and that it always carries a non-nil waitFn (totalTries() +// never needs one, but do() calls p.waitFn only when a retry is +// warranted, so a nil waitFn would only panic on a path this policy +// shape never takes — still, newRetryPolicy always sets one). func TestNewRetryPolicy_Disabled(t *testing.T) { p, err := newRetryPolicy(RetryConfig{}) if err != nil { @@ -68,10 +107,17 @@ func TestNewRetryPolicy_Disabled(t *testing.T) { if p.enabled { t.Error("p.enabled = true, want false for RetryConfig{}") } + if p.waitFn == nil { + t.Error("p.waitFn is nil, want a non-nil default") + } + if got := p.totalTries(); got != 1 { + t.Errorf("p.totalTries() = %d, want 1 (disabled: no retry)", got) + } } // TestNewRetryPolicy_Defaults asserts Enabled=true with Attempts/Backoff -// left unset applies the spec defaults: 1 attempt, 250ms backoff. +// left unset applies the spec defaults: 1 retry (two tries total, not +// one) and 250ms backoff. func TestNewRetryPolicy_Defaults(t *testing.T) { p, err := newRetryPolicy(RetryConfig{Enabled: true}) if err != nil { @@ -81,14 +127,22 @@ func TestNewRetryPolicy_Defaults(t *testing.T) { t.Error("p.enabled = false, want true") } if p.attempts != 1 { - t.Errorf("p.attempts = %d, want 1", p.attempts) + t.Errorf("p.attempts = %d, want 1 (retries after the first try)", p.attempts) } if p.backoff != 250*time.Millisecond { t.Errorf("p.backoff = %v, want 250ms", p.backoff) } + if p.waitFn == nil { + t.Error("p.waitFn is nil, want a non-nil default") + } + if got := p.totalTries(); got != 2 { + t.Errorf("p.totalTries() = %d, want 2 (1 initial try + 1 default retry)", got) + } } -// TestNewRetryPolicy_ValidatesAttempts covers the 1..3 attempts range. +// TestNewRetryPolicy_ValidatesAttempts covers the 1..3 range — +// Attempts counts retries after the first try, so this range means +// "1..3 retries", not "1..3 tries". func TestNewRetryPolicy_ValidatesAttempts(t *testing.T) { tests := []struct { name string @@ -128,33 +182,115 @@ func TestNewRetryPolicy_ValidatesBackoff(t *testing.T) { } } -// spySleep returns a sleepFn that records every requested duration -// instead of actually sleeping, so retry-wait tests run instantly and can -// assert on the exact durations retryPolicy.do computed. -func spySleep(waits *[]time.Duration) func(time.Duration) { - return func(d time.Duration) { - *waits = append(*waits, d) +// TestRetryPolicy_TotalTries pins totalTries()'s retries-to-tries +// conversion directly, including the defensive clamp on a malformed +// attempts value (0 or negative) reachable only by constructing a +// *retryPolicy by hand rather than through newRetryPolicy. +func TestRetryPolicy_TotalTries(t *testing.T) { + tests := []struct { + policy *retryPolicy + name string + want int + }{ + {nil, "nil policy", 1}, + {&retryPolicy{enabled: false, attempts: 5}, "disabled, attempts ignored", 1}, + {&retryPolicy{enabled: true, attempts: 1}, "enabled, attempts=1 (default)", 2}, + {&retryPolicy{enabled: true, attempts: 2}, "enabled, attempts=2", 3}, + {&retryPolicy{enabled: true, attempts: 3}, "enabled, attempts=3 (max)", 4}, + {&retryPolicy{enabled: true, attempts: 0}, "enabled, attempts=0 (malformed, clamped to 1)", 2}, + {&retryPolicy{enabled: true, attempts: -5}, "enabled, attempts=-5 (malformed, clamped to 1)", 2}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := tt.policy.totalTries(); got != tt.want { + t.Errorf("totalTries() = %d, want %d", got, tt.want) + } + }) } } -// callSequence returns a call func for retryPolicy.do that returns -// results[i] on its i-th invocation (0-indexed), and counts invocations -// into calls. -func callSequence(calls *int32, results ...func() (*http.Response, error)) func() (*http.Response, error) { - return func() (*http.Response, error) { - n := int(atomic.AddInt32(calls, 1)) - 1 - if n >= len(results) { - n = len(results) - 1 - } - return results[n]() +// TestRetryPolicy_Do_MalformedAttempts_NeverReturnsNilNil proves the +// totalTries() clamp keeps do's loop from ever skipping call() entirely: +// a hand-built policy with an out-of-range attempts value (impossible +// through newRetryPolicy, but not impossible through a direct struct +// literal) must still make at least one real attempt, never returning +// the disallowed (nil, nil) shape. +func TestRetryPolicy_Do_MalformedAttempts_NeverReturnsNilNil(t *testing.T) { + var calls int32 + p := &retryPolicy{enabled: true, attempts: -1, backoff: time.Millisecond, waitFn: spyWait(&[]time.Duration{})} + call := callSequence(&calls, func() (*http.Response, error) { return fakeResp(http.StatusServiceUnavailable, nil), nil }) + + resp, err := p.do(context.Background(), call) + if resp == nil && err == nil { + t.Fatal("do returned (nil, nil): call() was never invoked despite a malformed attempts value") + } + if got := atomic.LoadInt32(&calls); got < 1 { + t.Errorf("calls = %d, want at least 1", got) + } +} + +// TestRetryPolicy_Do_EnabledDefaults_ExactlyOneRetry is the ruling's own +// acceptance case: retry:{enabled:true} with every other field left at +// its default must perform exactly one retry (two tries total), not +// zero — the bug the controller's review caught. +func TestRetryPolicy_Do_EnabledDefaults_ExactlyOneRetry(t *testing.T) { + p, err := newRetryPolicy(RetryConfig{Enabled: true}) + if err != nil { + t.Fatalf("newRetryPolicy: %v", err) + } + var waits []time.Duration + p.waitFn = spyWait(&waits) + + var calls int32 + call := callSequence(&calls, + func() (*http.Response, error) { return fakeResp(http.StatusServiceUnavailable, nil), nil }, + func() (*http.Response, error) { return fakeResp(http.StatusOK, nil), nil }, + ) + resp, err := p.do(context.Background(), call) + if err != nil { + t.Fatalf("do: %v", err) + } + if got := atomic.LoadInt32(&calls); got != 2 { + t.Errorf("calls = %d, want 2 (default attempts=1 means one retry: two tries total)", got) + } + if resp.StatusCode != http.StatusOK { + t.Errorf("resp.StatusCode = %d, want 200", resp.StatusCode) + } + if len(waits) != 1 { + t.Errorf("waited %d times, want exactly 1", len(waits)) + } +} + +// TestRetryPolicy_Do_EnabledDefaults_CapsAtTwoTotalTries is the same +// default-config case against an upstream that never succeeds: it must +// stop at 2 total tries (1 initial + 1 default retry), not loop forever +// and not stop at 1. +func TestRetryPolicy_Do_EnabledDefaults_CapsAtTwoTotalTries(t *testing.T) { + p, err := newRetryPolicy(RetryConfig{Enabled: true}) + if err != nil { + t.Fatalf("newRetryPolicy: %v", err) + } + p.waitFn = spyWait(&[]time.Duration{}) + + var calls int32 + call := callSequence(&calls, func() (*http.Response, error) { return fakeResp(http.StatusServiceUnavailable, nil), nil }) + resp, err := p.do(context.Background(), call) + if err != nil { + t.Fatalf("do: %v", err) + } + if got := atomic.LoadInt32(&calls); got != 2 { + t.Errorf("calls = %d, want 2 (1 initial + 1 default retry)", got) + } + if resp.StatusCode != http.StatusServiceUnavailable { + t.Errorf("resp.StatusCode = %d, want 503 (last attempt's result)", resp.StatusCode) } } // TestRetryPolicy_Do_RetriesOnTransientThenSucceeds covers the brief's // three canonical transient-then-success cases: 503, a raw network error // (connection reset), and 429 with a qualifying Retry-After header. Each -// retries exactly once, waits the expected duration, and returns the -// second attempt's success. +// uses attempts:1 (one retry allowed) and retries exactly once, waits +// the expected duration, and returns the second attempt's success. func TestRetryPolicy_Do_RetriesOnTransientThenSucceeds(t *testing.T) { tests := []struct { name string @@ -174,20 +310,16 @@ func TestRetryPolicy_Do_RetriesOnTransientThenSucceeds(t *testing.T) { wantWaits: []time.Duration{100 * time.Millisecond}, }, { - name: "429 with Retry-After then success", - first: func() (*http.Response, error) { - h := http.Header{} - h.Set("Retry-After", "1") - return fakeResp(http.StatusTooManyRequests, h), nil - }, - wantWaits: []time.Duration{1 * time.Second}, + name: "429 with Retry-After then success", + first: func() (*http.Response, error) { return retryAfterResp(http.StatusTooManyRequests, "1"), nil }, + wantWaits: []time.Duration{time.Second}, }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { var calls int32 var waits []time.Duration - p := &retryPolicy{enabled: true, attempts: 2, backoff: 100 * time.Millisecond, sleepFn: spySleep(&waits)} + p := &retryPolicy{enabled: true, attempts: 1, backoff: 100 * time.Millisecond, waitFn: spyWait(&waits)} call := callSequence(&calls, tt.first, func() (*http.Response, error) { return fakeResp(http.StatusOK, nil), nil }) @@ -199,8 +331,8 @@ func TestRetryPolicy_Do_RetriesOnTransientThenSucceeds(t *testing.T) { if resp.StatusCode != http.StatusOK { t.Errorf("resp.StatusCode = %d, want 200", resp.StatusCode) } - if calls != 2 { - t.Errorf("calls = %d, want 2 (one retry)", calls) + if got := atomic.LoadInt32(&calls); got != 2 { + t.Errorf("calls = %d, want 2 (one retry)", got) } if len(waits) != len(tt.wantWaits) || (len(waits) > 0 && waits[0] != tt.wantWaits[0]) { t.Errorf("waits = %v, want %v", waits, tt.wantWaits) @@ -209,13 +341,45 @@ func TestRetryPolicy_Do_RetriesOnTransientThenSucceeds(t *testing.T) { } } +// TestRetryPolicy_WaitBefore_RetryAfterHandling pins spec §1's +// Retry-After handling directly: only a 429 whose Retry-After parses as +// a non-negative whole-second count within the 2s cap wins outright. +// Every other shape — over the cap, the HTTP-date form, negative, no +// header at all, or present on a non-429 status — falls back to +// exponential backoff (spec: ">2s -> exponential backoff", not a flat +// 2s), verified here against attempt=1's backoff*2^0 value. +func TestRetryPolicy_WaitBefore_RetryAfterHandling(t *testing.T) { + p := &retryPolicy{backoff: 100 * time.Millisecond} + backoffFallback := 100 * time.Millisecond // attempt=1: backoff * 2^0 + + tests := []struct { + resp *http.Response + name string + want time.Duration + }{ + {retryAfterResp(http.StatusTooManyRequests, "1"), "429 with qualifying Retry-After (1s, within cap)", time.Second}, + {retryAfterResp(http.StatusTooManyRequests, "30"), "429 with Retry-After over the 2s cap falls back to exponential backoff", backoffFallback}, + {retryAfterResp(http.StatusTooManyRequests, "Wed, 21 Oct 2015 07:28:00 GMT"), "429 with HTTP-date Retry-After falls back to exponential backoff", backoffFallback}, + {retryAfterResp(http.StatusTooManyRequests, "-5"), "429 with negative Retry-After falls back to exponential backoff", backoffFallback}, + {fakeResp(http.StatusTooManyRequests, nil), "429 with no Retry-After header falls back to exponential backoff", backoffFallback}, + {retryAfterResp(http.StatusServiceUnavailable, "1"), "503 with a Retry-After header is ignored (not 429)", backoffFallback}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := p.waitBefore(1, tt.resp); got != tt.want { + t.Errorf("waitBefore(1, ...) = %v, want %v", got, tt.want) + } + }) + } +} + // TestRetryPolicy_Do_NonTransientStatus_NoRetry asserts a 400 response // (not 429, not 5xx) is returned as-is on the first attempt, with no // retry and no wait. func TestRetryPolicy_Do_NonTransientStatus_NoRetry(t *testing.T) { var calls int32 var waits []time.Duration - p := &retryPolicy{enabled: true, attempts: 3, backoff: time.Millisecond, sleepFn: spySleep(&waits)} + p := &retryPolicy{enabled: true, attempts: 3, backoff: time.Millisecond, waitFn: spyWait(&waits)} call := callSequence(&calls, func() (*http.Response, error) { return fakeResp(http.StatusBadRequest, nil), nil }) resp, err := p.do(context.Background(), call) @@ -225,8 +389,8 @@ func TestRetryPolicy_Do_NonTransientStatus_NoRetry(t *testing.T) { if resp.StatusCode != http.StatusBadRequest { t.Errorf("resp.StatusCode = %d, want 400", resp.StatusCode) } - if calls != 1 { - t.Errorf("calls = %d, want 1 (no retry on non-transient status)", calls) + if got := atomic.LoadInt32(&calls); got != 1 { + t.Errorf("calls = %d, want 1 (no retry on non-transient status)", got) } if len(waits) != 0 { t.Errorf("waits = %v, want none", waits) @@ -238,7 +402,10 @@ func TestRetryPolicy_Do_NonTransientStatus_NoRetry(t *testing.T) { // a transient failure. func TestRetryPolicy_Do_DisabledPolicy_NoRetry(t *testing.T) { var calls int32 - p := &retryPolicy{enabled: false, sleepFn: func(time.Duration) { t.Error("sleepFn called on a disabled policy") }} + p := &retryPolicy{enabled: false, waitFn: func(context.Context, time.Duration) bool { + t.Error("waitFn called on a disabled policy") + return true + }} call := callSequence(&calls, func() (*http.Response, error) { return fakeResp(http.StatusServiceUnavailable, nil), nil }) resp, err := p.do(context.Background(), call) @@ -248,8 +415,8 @@ func TestRetryPolicy_Do_DisabledPolicy_NoRetry(t *testing.T) { if resp.StatusCode != http.StatusServiceUnavailable { t.Errorf("resp.StatusCode = %d, want 503", resp.StatusCode) } - if calls != 1 { - t.Errorf("calls = %d, want 1 (retry disabled)", calls) + if got := atomic.LoadInt32(&calls); got != 1 { + t.Errorf("calls = %d, want 1 (retry disabled)", got) } } @@ -269,34 +436,33 @@ func TestRetryPolicy_Do_NilPolicy_NoRetry(t *testing.T) { if resp.StatusCode != http.StatusServiceUnavailable { t.Errorf("resp.StatusCode = %d, want 503", resp.StatusCode) } - if calls != 1 { - t.Errorf("calls = %d, want 1 (nil policy = disabled)", calls) + if got := atomic.LoadInt32(&calls); got != 1 { + t.Errorf("calls = %d, want 1 (nil policy = disabled)", got) } } -// TestRetryPolicy_Do_AttemptsCap asserts a policy that never sees success -// stops after exactly p.attempts tries, waiting p.attempts-1 times with -// exponential backoff, capped at maxRetryWait. +// TestRetryPolicy_Do_AttemptsCap asserts a policy that never sees +// success stops after exactly p.attempts+1 total tries (attempts +// retries after the first), waiting p.attempts times with exponential +// backoff, capped at maxRetryWait. func TestRetryPolicy_Do_AttemptsCap(t *testing.T) { var calls int32 var waits []time.Duration - p := &retryPolicy{enabled: true, attempts: 3, backoff: time.Second, sleepFn: spySleep(&waits)} + p := &retryPolicy{enabled: true, attempts: 3, backoff: time.Second, waitFn: spyWait(&waits)} call := callSequence(&calls, func() (*http.Response, error) { return fakeResp(http.StatusServiceUnavailable, nil), nil }) resp, err := p.do(context.Background(), call) if err != nil { t.Fatalf("do: %v", err) } - if calls != 3 { - t.Errorf("calls = %d, want 3 (attempts cap)", calls) + if got := atomic.LoadInt32(&calls); got != 4 { + t.Errorf("calls = %d, want 4 (1 initial + 3 retries)", got) } if resp.StatusCode != http.StatusServiceUnavailable { t.Errorf("resp.StatusCode = %d, want 503 (last attempt's result)", resp.StatusCode) } - wantWaits := []time.Duration{maxRetryWait, maxRetryWait} // 1s*2^0=1s→wait capped? no: 1s, 2s both <= cap(2s) - _ = wantWaits - if len(waits) != 2 { - t.Fatalf("waits = %v, want 2 entries (between 3 attempts)", waits) + if len(waits) != 3 { + t.Fatalf("waits = %v, want 3 entries (between 4 attempts)", waits) } if waits[0] != time.Second { t.Errorf("waits[0] = %v, want 1s (backoff * 2^0)", waits[0]) @@ -304,6 +470,9 @@ func TestRetryPolicy_Do_AttemptsCap(t *testing.T) { if waits[1] != 2*time.Second { t.Errorf("waits[1] = %v, want 2s (backoff * 2^1, at the cap)", waits[1]) } + if waits[2] != 2*time.Second { + t.Errorf("waits[2] = %v, want 2s (backoff * 2^2, clamped to the cap)", waits[2]) + } } // TestRetryPolicy_Do_BackoffCappedAtMaxWait asserts a backoff whose @@ -311,12 +480,15 @@ func TestRetryPolicy_Do_AttemptsCap(t *testing.T) { func TestRetryPolicy_Do_BackoffCappedAtMaxWait(t *testing.T) { var calls int32 var waits []time.Duration - p := &retryPolicy{enabled: true, attempts: 3, backoff: 2 * time.Second, sleepFn: spySleep(&waits)} + p := &retryPolicy{enabled: true, attempts: 3, backoff: 2 * time.Second, waitFn: spyWait(&waits)} call := callSequence(&calls, func() (*http.Response, error) { return fakeResp(http.StatusServiceUnavailable, nil), nil }) if _, err := p.do(context.Background(), call); err != nil { t.Fatalf("do: %v", err) } + if len(waits) != 3 { + t.Fatalf("waits = %v, want 3 entries", waits) + } for i, w := range waits { if w > maxRetryWait { t.Errorf("waits[%d] = %v, exceeds cap %v", i, w, maxRetryWait) @@ -324,41 +496,97 @@ func TestRetryPolicy_Do_BackoffCappedAtMaxWait(t *testing.T) { } } -// TestRetryPolicy_Do_CtxCanceledAbortsRetry asserts a context canceled -// before a retry wait would begin stops the loop immediately: no further -// call, no sleep, and the already-failed result is returned as-is. -func TestRetryPolicy_Do_CtxCanceledAbortsRetry(t *testing.T) { +// TestRetryPolicy_Do_CtxAlreadyCanceled_NoRetry asserts a context +// canceled before do() even starts stops the loop after the first +// attempt: the real (ctx-aware) defaultWaitFn resolves the already-closed +// ctx.Done() case rather than blocking out the backoff. +func TestRetryPolicy_Do_CtxAlreadyCanceled_NoRetry(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) cancel() // canceled before do() ever runs var calls int32 - p := &retryPolicy{enabled: true, attempts: 3, backoff: time.Millisecond, sleepFn: func(time.Duration) { - t.Error("sleepFn called after ctx was already canceled") - }} + p := &retryPolicy{enabled: true, attempts: 3, backoff: 50 * time.Millisecond, waitFn: defaultWaitFn} call := callSequence(&calls, func() (*http.Response, error) { return fakeResp(http.StatusServiceUnavailable, nil), nil }) resp, err := p.do(ctx, call) if err != nil { t.Fatalf("do: %v", err) } - if calls != 1 { - t.Errorf("calls = %d, want 1 (ctx already done, no retry)", calls) + if got := atomic.LoadInt32(&calls); got != 1 { + t.Errorf("calls = %d, want 1 (ctx already done, no retry)", got) } if resp.StatusCode != http.StatusServiceUnavailable { t.Errorf("resp.StatusCode = %d, want 503", resp.StatusCode) } } +// TestRetryPolicy_Do_CtxCanceledDuringWait_ReturnsPromptly proves the +// ctx-aware waitFn aborts an in-progress backoff wait the moment ctx +// ends, rather than blocking out the full duration: a 2s backoff against +// a 20ms context deadline must return in well under 2s. +func TestRetryPolicy_Do_CtxCanceledDuringWait_ReturnsPromptly(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond) + defer cancel() + + var calls int32 + p := &retryPolicy{enabled: true, attempts: 2, backoff: 2 * time.Second, waitFn: defaultWaitFn} + call := callSequence(&calls, func() (*http.Response, error) { return fakeResp(http.StatusServiceUnavailable, nil), nil }) + + start := time.Now() + resp, err := p.do(ctx, call) + elapsed := time.Since(start) + + if err != nil { + t.Fatalf("do: %v", err) + } + if elapsed > 500*time.Millisecond { + t.Errorf("do took %v, want well under the 2s backoff (ctx should abort the wait promptly)", elapsed) + } + if got := atomic.LoadInt32(&calls); got != 1 { + t.Errorf("calls = %d, want 1 (ctx canceled during the wait, no further attempt)", got) + } + if resp.StatusCode != http.StatusServiceUnavailable { + t.Errorf("resp.StatusCode = %d, want 503 (last attempt's result)", resp.StatusCode) + } +} + +// trackingReadCloser wraps an io.Reader with a Close method and an EOF +// hook, for asserting a retried-away response body was both drained +// (read to EOF) and closed — not just closed with data left unread. +type trackingReadCloser struct { + io.Reader + onClose func() + onEOF func() +} + +func (c *trackingReadCloser) Read(p []byte) (int, error) { + n, err := c.Reader.Read(p) + if err == io.EOF { + c.onEOF() + } + return n, err +} + +func (c *trackingReadCloser) Close() error { + c.onClose() + return nil +} + // TestRetryPolicy_Do_ClosesFailedAttemptBody asserts a retried-away -// response's body is drained and closed before the next attempt, so the -// underlying connection is eligible for reuse instead of being forced -// closed by an un-drained body. +// response's body is both drained to EOF and closed before the next +// attempt, so the underlying connection is eligible for reuse instead of +// being forced closed by an un-drained body. func TestRetryPolicy_Do_ClosesFailedAttemptBody(t *testing.T) { var calls int32 var waits []time.Duration closed := false - firstBody := &trackingCloser{Reader: strings.NewReader("stale body"), onClose: func() { closed = true }} - p := &retryPolicy{enabled: true, attempts: 2, backoff: time.Millisecond, sleepFn: spySleep(&waits)} + drained := false + firstBody := &trackingReadCloser{ + Reader: strings.NewReader("stale body"), + onClose: func() { closed = true }, + onEOF: func() { drained = true }, + } + p := &retryPolicy{enabled: true, attempts: 1, backoff: time.Millisecond, waitFn: spyWait(&waits)} call := callSequence(&calls, func() (*http.Response, error) { return &http.Response{StatusCode: http.StatusServiceUnavailable, Header: http.Header{}, Body: firstBody}, nil @@ -369,21 +597,35 @@ func TestRetryPolicy_Do_ClosesFailedAttemptBody(t *testing.T) { if _, err := p.do(context.Background(), call); err != nil { t.Fatalf("do: %v", err) } + if !drained { + t.Error("failed attempt's response body was never fully drained (read to EOF)") + } if !closed { t.Error("failed attempt's response body was never closed") } } -// trackingCloser wraps an io.Reader with a Close method that records the -// call, for asserting a retried-away response body was actually closed. -type trackingCloser struct { - io.Reader - onClose func() -} - -func (c *trackingCloser) Close() error { - c.onClose() - return nil +// TestUpstreamJSON_RequestBuildFailure_NotRetried proves a failure to +// build the outgoing *http.Request (an invalid method, here) is +// classified non-transient and never retried — retrying an identically +// malformed request would only waste attempts on a guaranteed repeat +// failure. +func TestUpstreamJSON_RequestBuildFailure_NotRetried(t *testing.T) { + p := &retryPolicy{enabled: true, attempts: 3, backoff: time.Millisecond, waitFn: func(context.Context, time.Duration) bool { + t.Error("waitFn called: a request-build failure must not be retried") + return true + }} + client := &http.Client{} + _, err := upstreamJSON(context.Background(), client, "IN VALID", "http://127.0.0.1:1/x", nil, nil, p) + if err == nil { + t.Fatal("upstreamJSON: want an error for an invalid method, got nil") + } + if !errors.Is(err, errRequestBuildFailed) { + t.Errorf("err = %v, want errors.Is(err, errRequestBuildFailed)", err) + } + if !errors.Is(err, errUpstream) { + t.Errorf("err = %v, want it to still wrap errUpstream too", err) + } } // TestBuildAdapters_WiresRetryPolicyFromConfig asserts buildAdapters @@ -409,6 +651,9 @@ func TestBuildAdapters_WiresRetryPolicyFromConfig(t *testing.T) { if !oa.retry.enabled || oa.retry.attempts != 2 || oa.retry.backoff != 50*time.Millisecond { t.Errorf("oa.retry = %+v, want enabled=true attempts=2 backoff=50ms", oa.retry) } + if got := oa.retry.totalTries(); got != 3 { + t.Errorf("oa.retry.totalTries() = %d, want 3 (1 initial + 2 retries)", got) + } } // TestBuildAdapters_InvalidRetryConfig_ReturnsError asserts an @@ -446,6 +691,44 @@ func TestNewGateway_RetryDisabledByDefault_Unwired(t *testing.T) { } } +// TestRetryConfig_JSONDecode_WiresIntoGateway proves RetryConfig decodes +// correctly off the wire (JSON — Traefik's YAML dynamic config decodes +// to the same shape, per manifest_test.go's own note) and that +// newGateway builds a policy reflecting exactly what was decoded: +// attempts=2 means 2 retries (3 total tries), not the literal number of +// tries. +func TestRetryConfig_JSONDecode_WiresIntoGateway(t *testing.T) { + cfg := CreateConfig() + blob := `{"providers":{"openai":{"type":"openai","apiKey":"k"}}, + "retry":{"enabled":true,"attempts":2,"backoff":"100ms"}}` + if err := json.Unmarshal([]byte(blob), cfg); err != nil { + t.Fatalf("decode: %v", err) + } + if !cfg.Retry.Enabled || cfg.Retry.Attempts != 2 || cfg.Retry.Backoff != "100ms" { + t.Fatalf("decoded cfg.Retry = %+v, want {Enabled:true Attempts:2 Backoff:100ms}", cfg.Retry) + } + + next := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {}) + h, err := New(context.Background(), next, cfg, "llmgw") + if err != nil { + t.Fatalf("New: %v", err) + } + gw := h.(*Gateway) + oa, ok := gw.adapters["openai"].(*openaiAdapter) + if !ok { + t.Fatalf("adapters[openai] = %T, want *openaiAdapter", gw.adapters["openai"]) + } + if oa.retry == nil || !oa.retry.enabled { + t.Fatalf("oa.retry = %+v, want an enabled policy", oa.retry) + } + if oa.retry.attempts != 2 { + t.Errorf("oa.retry.attempts = %d, want 2 (retries after the first try)", oa.retry.attempts) + } + if got := oa.retry.totalTries(); got != 3 { + t.Errorf("oa.retry.totalTries() = %d, want 3 (1 initial + 2 retries)", got) + } +} + // ---- adapter-level wiring tests (stream-initiation path) ---- // TestOpenAIAdapter_ChatCompletion_NonStream_RetriedOnTransientFailure @@ -467,15 +750,15 @@ func TestOpenAIAdapter_ChatCompletion_NonStream_RetriedOnTransientFailure(t *tes defer srv.Close() a := newOpenAIAdapter("p1", srv.URL, "sk-test") - a.retry = &retryPolicy{enabled: true, attempts: 2, backoff: time.Millisecond, sleepFn: func(time.Duration) {}} + a.retry = &retryPolicy{enabled: true, attempts: 2, backoff: time.Millisecond, waitFn: func(context.Context, time.Duration) bool { return true }} rec := httptest.NewRecorder() u, err := a.chatCompletion(context.Background(), rec, map[string]any{"model": "gpt-5", "messages": []any{}}) if err != nil { t.Fatalf("chatCompletion: %v", err) } - if calls != 2 { - t.Errorf("upstream calls = %d, want 2", calls) + if got := atomic.LoadInt32(&calls); got != 2 { + t.Errorf("upstream calls = %d, want 2", got) } if rec.Code != http.StatusOK { t.Errorf("rec.Code = %d, want 200", rec.Code) @@ -512,15 +795,15 @@ func TestOpenAIAdapter_ChatCompletion_Stream_RetriedBeforeFirstChunk(t *testing. defer srv.Close() a := newOpenAIAdapter("p1", srv.URL, "sk-test") - a.retry = &retryPolicy{enabled: true, attempts: 2, backoff: time.Millisecond, sleepFn: func(time.Duration) {}} + a.retry = &retryPolicy{enabled: true, attempts: 2, backoff: time.Millisecond, waitFn: func(context.Context, time.Duration) bool { return true }} rec := httptest.NewRecorder() u, err := a.chatCompletion(context.Background(), rec, map[string]any{"model": "gpt-5", "messages": []any{}, "stream": true}) if err != nil { t.Fatalf("chatCompletion: %v", err) } - if calls != 2 { - t.Errorf("upstream calls = %d, want 2", calls) + if got := atomic.LoadInt32(&calls); got != 2 { + t.Errorf("upstream calls = %d, want 2", got) } if u.prompt != 7 || u.completion != 9 { t.Errorf("usage = %+v, want {7 9}", u) @@ -549,21 +832,28 @@ func TestOpenAIAdapter_ChatCompletion_Stream_MidDeathNotRetried(t *testing.T) { _, _ = fmt.Fprintf(w, "data: %s\n\n", streamFrames[0]) fl.Flush() + // t.Error (not t.Fatal) — this handler runs on its own goroutine, + // spawned by net/http's server; t.FailNow (which t.Fatal calls) + // is documented to be safe only from the goroutine running the + // test function itself. hj, ok := w.(http.Hijacker) if !ok { - t.Fatal("ResponseWriter does not support Hijacker") + t.Error("ResponseWriter does not support Hijacker") + return } conn, _, err := hj.Hijack() if err != nil { - t.Fatalf("Hijack: %v", err) + t.Errorf("Hijack: %v", err) + return } _ = conn.Close() // abrupt drop mid-chunked-body: client sees a read error, not a clean EOF })) defer srv.Close() a := newOpenAIAdapter("p1", srv.URL, "sk-test") - a.retry = &retryPolicy{enabled: true, attempts: 3, backoff: time.Millisecond, sleepFn: func(time.Duration) { - t.Error("sleepFn called: a mid-stream death must not trigger a retry") + a.retry = &retryPolicy{enabled: true, attempts: 3, backoff: time.Millisecond, waitFn: func(context.Context, time.Duration) bool { + t.Error("waitFn called: a mid-stream death must not trigger a retry") + return true }} rec := httptest.NewRecorder() @@ -574,8 +864,8 @@ func TestOpenAIAdapter_ChatCompletion_Stream_MidDeathNotRetried(t *testing.T) { if !errors.Is(err, errUpstream) { t.Errorf("chatCompletion err = %v, want it to wrap errUpstream", err) } - if calls != 1 { - t.Errorf("upstream calls = %d, want 1 (mid-stream death is not retried)", calls) + if got := atomic.LoadInt32(&calls); got != 1 { + t.Errorf("upstream calls = %d, want 1 (mid-stream death is not retried)", got) } if !strings.Contains(rec.Body.String(), "Hel") { t.Errorf("rec.Body = %q, want it to contain the chunk already forwarded before the drop", rec.Body.String())