merge: progress-based upstream request timeout, 5m default

# Conflicts:
#	README.md
#	llmgateway.go
#	tools/yaegi-check/main.go
This commit is contained in:
2026-08-23 03:07:48 +01:00
14 changed files with 1691 additions and 45 deletions
+89
View File
@@ -194,6 +194,7 @@ config accepts them as YAML, which decodes to the same JSON shape.
| `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. |
| `failover` | `FailoverConfig` | `{}` (disabled) | Cross-provider failover for a bare model id more than one configured provider serves — see [Failover](#failover). Omitted or `enabled: false` means no failover, byte-identical to a gateway built before this field existed. `enabled: true` opts in; a deployment with no overlapping providers behaves exactly as before regardless of this setting either way. |
| `requestTimeout` | `string` (Go duration) | `5m` | Progress-based upstream request timeout, applied to every provider adapter call, native passthrough, and the MCP/A2A target proxy — see [Request timeout](#request-timeout). **Behavior change for every deployment**: before this field existed, no timeout was set at all, so a provider that accepted a connection and then went silent hung the request forever. A provider's own `requestTimeout` (`ProviderConfig`) overrides this. Empty means the default; a value that fails to parse, or parses to zero or a negative duration, is a construction error. |
| `cache` | `CacheConfig` | `{}` (disabled) | Opt-in Redis-backed response cache for unified non-streaming chat/embeddings — see [Caching](#caching). Omitted or `enabled: false` means no caching, byte-identical to a gateway built before this field existed. |
| `breaker` | `BreakerConfig` | `{}` (every default) | Per-provider discovery circuit breaker — see [Provider health (discovery circuit breaker)](#provider-health-discovery-circuit-breaker). Every field is individually zero-means-default; a provider whose discovery never fails is unaffected regardless of what this block contains. |
| `admin` | `*AdminConfig` | `nil` (disabled) | Read-only admin dashboard — see [Admin](#admin). `nil` or `enabled: false` means the `/admin*` routes are not registered at all. |
@@ -234,6 +235,7 @@ also a construction error, never a panic (`providers.go`, `mcp_a2a.go`,
| `discovery` | `bool` | `false` | Pull the provider's own model-listing endpoint at startup and on `discoveryInterval`. A discovery failure is logged and non-fatal; construction still succeeds on `models` alone. |
| `metadataPath` | `string` | `""` (no metadata capture) | Only read by `openai`-type providers. A second endpoint, fetched alongside `discovery`, that reports per-model context length — for example `/api/v0/models` on an LM Studio server. Must start with `/` when set (checked at construction). See [Model metadata](#model-metadata). |
| `passthrough` | `*bool` | `nil` (enabled) | `nil` or `true`: this provider's native passthrough route (`/{name}/...`) stays reachable, unchanged from before this field existed. `false`: the route is disabled — `ServeHTTP` treats it exactly like an unconfigured provider name, falling through to the ordinary unknown-route 404 without even reaching authentication. Does not affect the MCP/A2A target proxy, which has its own routing prefix. |
| `requestTimeout` | `string` (Go duration) | Inherits the top-level `requestTimeout` (`5m` if that is also unset) | Overrides the global request timeout for this provider alone, including its native passthrough route — see [Request timeout](#request-timeout). An explicit override always wins over the global default. Empty inherits; a value that fails to parse, or parses to zero or a negative duration, is a construction error. |
### `GroupConfig`
@@ -970,6 +972,93 @@ never a second candidate to try.
failover, and applies to every route, not only this one; it is not
addressed in this change.
## Request timeout
Every provider adapter call and every native passthrough
(`/{provider}/...`) request is bounded by a progress-based timeout:
aborted once this gateway observes no upstream read progress for the
configured duration, never simply because elapsed time crossed a fixed
deadline. The MCP/A2A target proxy (`/mcp/{name}/...`, `/a2a/{name}/...`)
shares the identical mechanism. The federated `/mcp` endpoint
(`mcp_federation.go`) does **not** — see the "Federated MCP" note below.
**Behavior change for every deployment, not an invisible bug fix.** Before
this feature, an adapter's `*http.Client` set no timeout of any kind: a
provider that accepted the connection and then went silent — sent no
headers, or sent headers and then stalled mid-body — hung the gateway
request forever. Default, when neither `requestTimeout` field is set:
**five minutes**.
- **Two mechanisms, one config value**: time to the first response header
(`Transport.ResponseHeaderTimeout`), and idle time WITHIN the response
body once headers arrive. The second is what catches a provider that
answers promptly and then stalls mid-stream or mid-JSON-body —
`ResponseHeaderTimeout` alone only bounds the wait for headers.
- **Progress-based, not total-duration.** A streaming response resets the
idle timer on every chunk it forwards, so a long generation that keeps
producing tokens completes even if the whole exchange runs for 20
minutes — a total-duration cap was deliberately rejected for exactly
this reason. Non-streaming responses share the identical mechanism: a
large or slow-but-progressing body is unaffected; only a body that goes
quiet for the full timeout is aborted.
- **What "progress" means, precisely.** The idle timer resets on bytes
read from the *decoded* body — after `net/http`'s own transparent gzip
decompression, when the upstream used it. A gzip'd response whose
compressed bytes keep arriving on the wire but whose deflate stream
yields no decompressed output for the whole timeout window is still
aborted: this gateway has no visibility into wire-level progress
without a transport or connection wrapper, which is disproportionate at
the five-minute default. In the same vein, a streaming forwarder reads
one chunk from the upstream, writes it to the client, then reads again
— the write is not itself watched, so time spent blocked writing to a
slow or stalled client counts against the SAME idle window as a silent
upstream. Aborting in that case is still the right call (holding an
upstream connection open indefinitely for a stalled client is its own
problem), but the failure is not always the provider's fault even
though the request is attributed to one for accounting purposes.
- **Never retried.** A `ResponseHeaderTimeout` failure surfaces as a
`context.DeadlineExceeded`-shaped error, which [Retry](#retry)'s own
transient classification explicitly excludes. A mid-body watchdog
timeout happens after the retry mechanism has already returned its
response to the caller, so it is never in a position to be retried
either way. Not retrying a hang is the intended behavior — retrying a
provider that just proved it cannot make progress would only waste the
same amount of time again.
- **Scope and override**: `requestTimeout` (top level) sets the global
default. A provider's own `requestTimeout` overrides it — an explicit
override always wins. Native passthrough for a provider uses that same
provider's resolved value, so a client hitting `/{provider}/...` gets
the identical bound as a request through `/v1/chat/completions`. The
MCP/A2A target proxy has no per-target override; it always uses the
global default.
- **Config form and validation**: a Go duration string (`"5m"`, `"90s"`),
matching `retry.backoff`/`cache.ttl`'s own convention. Empty means the
default. A value that fails to parse, or parses to zero or a negative
duration, is a construction error — unlike `cache.ttl`, zero/negative is
never accepted here as "no timeout": that would silently reintroduce the
exact hang this feature fixes.
- **Classification**: a timeout firing is always logged and reported to
the client as an upstream failure (`502`, `server_error`), naming
whichever provider or target the request was addressed to, and is never
classified as a client disconnect (`context.Canceled`) — the two stay
distinguishable in every log line. The log wording says "upstream
progress stalled", not "provider timed out": for the reason described
above (a slow client can trip the same watchdog a slow provider does),
the message reports what was observed without asserting which side
caused it.
**Federated MCP (`POST /mcp`) is bounded differently, not left unbounded.**
`mcp_federation.go`'s two direct upstream calls
(`doBackendJSONRPC`/`mcpBackendCloseSession`) are not wrapped by the
idle-progress watchdog described above — a federated backend that sends
headers and then stalls mid-body is not caught by it. They are not
unbounded either: both still run through `g.targetClient`, so
`Transport.ResponseHeaderTimeout` (the global `requestTimeout`) still
bounds the wait for headers, and each call additionally runs under its
own fixed `context.WithTimeout` — 20s for `tools/list`, 120s for
`tools/call`, 5s for the best-effort session-close DELETE. A worst-case
stall there is bounded at whichever of those applies, not indefinite.
## Caching
An opt-in, Redis-backed response cache for the unified, non-streaming
+49 -3
View File
@@ -12,6 +12,7 @@ import (
"os"
"runtime"
"sync"
"time"
telemetry "github.com/lukaszraczylo/oss-telemetry"
)
@@ -82,8 +83,24 @@ type Config struct {
// FailoverConfig's own doc comment (failover.go) for the full
// contract; a struct value, not a pointer, for the same
// zero-means-default reason as Breaker/Retry/Cache below.
Failover FailoverConfig `json:"failover"`
Breaker BreakerConfig `json:"breaker"`
// RequestTimeout is the progress-based upstream request timeout
// (feature: request timeout) applied to every provider adapter call
// and to the MCP/A2A target proxy, unless a provider sets its own
// ProviderConfig.RequestTimeout override. A Go duration string (e.g.
// "5m", "90s"), parsed and validated by resolveRequestTimeout
// (timeout.go). Empty (the default) resolves to defaultRequestTimeout,
// five minutes — a deliberate behavior change for every deployment,
// not an invisible bug fix: before this field existed, an upstream
// provider that accepted a connection and then hung could hang the
// gateway request forever (see defaultRequestTimeout's own doc
// comment, and README.md's "Request timeout" section). A non-empty
// value that fails to parse, or parses to a duration <= 0, is a
// construction error — zero or negative never means "no timeout" here,
// since that would silently reintroduce the exact hang this feature
// fixes.
RequestTimeout string `json:"requestTimeout,omitempty"`
Failover FailoverConfig `json:"failover"`
Breaker BreakerConfig `json:"breaker"`
// 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":
@@ -145,6 +162,14 @@ type ProviderConfig struct {
BaseURL string `json:"baseUrl,omitempty"`
APIKey string `json:"apiKey"`
DiscoveryInterval string `json:"discoveryInterval,omitempty"`
// RequestTimeout overrides the global Config.RequestTimeout for this
// provider alone — an explicit per-provider override always wins over
// the global default (house rule). Same Go-duration-string form,
// same validation (resolveRequestTimeout, timeout.go): empty inherits
// the global (or its own default of five minutes when the global is
// also unset); a non-empty value that fails to parse, or parses to a
// duration <= 0, is a construction error naming this provider.
RequestTimeout string `json:"requestTimeout,omitempty"`
// MetadataPath is an optional second discovery endpoint (feature
// v0.23), fetched alongside the provider's normal listModels call
// when set: a path such as "/api/v0/models" (LM Studio's own,
@@ -565,6 +590,13 @@ type Gateway struct {
// on a fresh Gateway always logs. Deliberately does NOT gate the 404
// loud-log line (operator ruling: that one must always log).
failoverLogGate logGate
// targetTimeout is targetClient's own resolved request timeout —
// Config.RequestTimeout only; an MCP/A2A target has no per-target
// override the way a provider does. proxyUpstream's own callers
// (mcp_a2a.go's handleTargetProxy) pass this through so its
// idle-progress body watchdog (timeout.go) matches the
// ResponseHeaderTimeout already set on targetClient's Transport.
targetTimeout time.Duration
// failover is Config.Failover, validated and resolved once by
// newGateway (validateFailoverConfig, failover.go).
failover failoverConfig
@@ -748,7 +780,21 @@ func newGateway(ctx context.Context, next http.Handler, config *Config, name str
return nil, err
}
g.adapters = adapters
g.targetClient = newAdapterHTTPClient()
// targetTimeout: the global default only (resolveRequestTimeout,
// timeout.go) — an MCP/A2A target has no per-target override the way a
// provider does (ProviderConfig.RequestTimeout). Resolved again here,
// separately from buildAdapters' own identical call, rather than
// widening buildAdapters' return signature to hand the value back:
// buildAdapters(cfg *Config) (map[string]providerAdapter, error) is
// called directly by several existing tests, and parsing the same
// short duration string twice at construction time is cheap.
targetTimeout, err := resolveRequestTimeout(config.RequestTimeout, "requestTimeout")
if err != nil {
return nil, err
}
g.targetTimeout = targetTimeout
g.targetClient = newAdapterHTTPClient(targetTimeout)
registry, err := newModelRegistry(adapters, config, g.errorf)
if err != nil {
+1 -1
View File
@@ -216,7 +216,7 @@ func (g *Gateway) handleTargetProxy(w http.ResponseWriter, r *http.Request, u *u
upstreamURL += "?" + r.URL.RawQuery
}
g.proxyUpstream(w, r, upstreamURL, g.targetClient, nil, nil, false, kind+" target (name "+name+")")
g.proxyUpstream(w, r, upstreamURL, g.targetClient, nil, nil, false, kind+" target (name "+name+")", g.targetTimeout)
}
// validateTargetURLs checks every configured MCP-server and agent entry at
+13 -3
View File
@@ -29,6 +29,12 @@ type anthropicAdapter struct {
adapterName string
baseURL string
apiKey string
// timeout is this adapter's resolved per-request timeout —
// resolveProviderTimeout, timeout.go; see openaiAdapter.timeout's own
// doc comment (provider_openai.go) for why this is set to
// defaultRequestTimeout at construction rather than left at its zero
// value the way retry is.
timeout time.Duration
}
// newAnthropicAdapter returns an anthropicAdapter for provider name, with
@@ -45,7 +51,8 @@ func newAnthropicAdapter(name, base, apiKey string) (*anthropicAdapter, error) {
adapterName: name,
baseURL: base,
apiKey: apiKey,
client: newAdapterHTTPClient(),
client: newAdapterHTTPClient(defaultRequestTimeout),
timeout: defaultRequestTimeout,
}, nil
}
@@ -61,6 +68,9 @@ func (a *anthropicAdapter) base() string { return a.baseURL }
// httpClient implements providerAdapter.
func (a *anthropicAdapter) httpClient() *http.Client { return a.client }
// requestTimeout implements providerAdapter.
func (a *anthropicAdapter) requestTimeout() time.Duration { return a.timeout }
// injectAuth implements providerAdapter: Anthropic's x-api-key scheme,
// plus the anthropic-version header every request must carry. Unlike
// openai-type's injectAuth, this is never a no-op — the constructor
@@ -112,7 +122,7 @@ func (a *anthropicAdapter) chatCompletion(ctx context.Context, w http.ResponseWr
hdr.Set("Accept", "text/event-stream")
}
resp, err := upstreamJSON(ctx, a.client, http.MethodPost, a.baseURL+anthropicMessagesPath, hdr, body, a.retry)
resp, err := upstreamJSON(ctx, a.client, http.MethodPost, a.baseURL+anthropicMessagesPath, hdr, body, a.retry, a.timeout, a.adapterName)
if err != nil {
return usage{}, err
}
@@ -226,7 +236,7 @@ func (a *anthropicAdapter) audioTranscription(_ context.Context, _ http.Response
// falls back to a provider's explicitly configured models on error rather
// than this adapter guessing at a fallback itself.
func (a *anthropicAdapter) listModels(ctx context.Context) ([]string, error) {
resp, err := upstreamJSON(ctx, a.client, http.MethodGet, a.baseURL+anthropicModelsPath, a.requestHeaders(false), nil, a.retry)
resp, err := upstreamJSON(ctx, a.client, http.MethodGet, a.baseURL+anthropicModelsPath, a.requestHeaders(false), nil, a.retry, a.timeout, a.adapterName)
if err != nil {
return nil, err
}
+15 -5
View File
@@ -23,6 +23,12 @@ type geminiAdapter struct {
adapterName string
baseURL string
apiKey string
// timeout is this adapter's resolved per-request timeout —
// resolveProviderTimeout, timeout.go; see openaiAdapter.timeout's own
// doc comment (provider_openai.go) for why this is set to
// defaultRequestTimeout at construction rather than left at its zero
// value the way retry is.
timeout time.Duration
}
// newGeminiAdapter returns a geminiAdapter for provider name, with base as
@@ -38,7 +44,8 @@ func newGeminiAdapter(name, base, apiKey string) (*geminiAdapter, error) {
adapterName: name,
baseURL: base,
apiKey: apiKey,
client: newAdapterHTTPClient(),
client: newAdapterHTTPClient(defaultRequestTimeout),
timeout: defaultRequestTimeout,
}, nil
}
@@ -54,6 +61,9 @@ func (a *geminiAdapter) base() string { return a.baseURL }
// httpClient implements providerAdapter.
func (a *geminiAdapter) httpClient() *http.Client { return a.client }
// requestTimeout implements providerAdapter.
func (a *geminiAdapter) requestTimeout() time.Duration { return a.timeout }
// injectAuth implements providerAdapter: Gemini's x-goog-api-key header
// scheme. Never a no-op — the constructor already rejected an empty apiKey.
func (a *geminiAdapter) injectAuth(r *http.Request) {
@@ -125,7 +135,7 @@ func (a *geminiAdapter) chatCompletion(ctx context.Context, w http.ResponseWrite
endpoint = a.baseURL + geminiAPIPrefix + escapedModel + ":streamGenerateContent?alt=sse"
}
resp, err := upstreamJSON(ctx, a.client, http.MethodPost, endpoint, hdr, body, a.retry)
resp, err := upstreamJSON(ctx, a.client, http.MethodPost, endpoint, hdr, body, a.retry, a.timeout, a.adapterName)
if err != nil {
return usage{}, err
}
@@ -234,7 +244,7 @@ func (a *geminiAdapter) embeddings(ctx context.Context, w http.ResponseWriter, r
endpoint = a.baseURL + geminiAPIPrefix + escapedModel + ":batchEmbedContents"
}
resp, err := upstreamJSON(ctx, a.client, http.MethodPost, endpoint, a.requestHeaders(true), body, a.retry)
resp, err := upstreamJSON(ctx, a.client, http.MethodPost, endpoint, a.requestHeaders(true), body, a.retry, a.timeout, a.adapterName)
if err != nil {
return usage{}, err
}
@@ -286,7 +296,7 @@ func (a *geminiAdapter) imagesGeneration(ctx context.Context, w http.ResponseWri
escapedModel := url.PathEscape(model)
endpoint := a.baseURL + geminiAPIPrefix + escapedModel + ":predict"
resp, err := upstreamJSON(ctx, a.client, http.MethodPost, endpoint, a.requestHeaders(true), body, a.retry)
resp, err := upstreamJSON(ctx, a.client, http.MethodPost, endpoint, a.requestHeaders(true), body, a.retry, a.timeout, a.adapterName)
if err != nil {
return usage{}, err
}
@@ -345,7 +355,7 @@ type geminiModelsListPayload struct {
// model ids, matching the shape openai-type and anthropic-type adapters
// both return.
func (a *geminiAdapter) listModels(ctx context.Context) ([]string, error) {
resp, err := upstreamJSON(ctx, a.client, http.MethodGet, a.baseURL+"/v1beta/models", a.requestHeaders(false), nil, a.retry)
resp, err := upstreamJSON(ctx, a.client, http.MethodGet, a.baseURL+"/v1beta/models", a.requestHeaders(false), nil, a.retry, a.timeout, a.adapterName)
if err != nil {
return nil, err
}
+27 -9
View File
@@ -7,6 +7,7 @@ import (
"io"
"net/http"
"strings"
"time"
)
// openaiAdapter is the providerAdapter for provider type "openai": OpenAI
@@ -29,18 +30,32 @@ type openaiAdapter struct {
// newOpenAIAdapter call site (every test, and any caller that never
// sets this) keeps behaving exactly as before this feature existed.
metadataPath string
// timeout is this adapter's resolved per-request timeout
// (resolveProviderTimeout, timeout.go) — UNLIKE retry above, it is NOT
// left at its zero value by a bare newOpenAIAdapter call: a
// zero-duration timeout would make watchdogBody's watchdog fire
// immediately (time.AfterFunc(0, ...) runs at the next scheduler
// tick), the opposite of "no timeout configured". newOpenAIAdapter
// sets it to defaultRequestTimeout; buildAdapters (providers.go)
// overwrites it with the real, config-resolved value for production
// use, the same way it overwrites retry/metadataPath above.
timeout time.Duration
}
// newOpenAIAdapter returns an openaiAdapter for provider name, with base as
// its already-defaulted, trailing-slash-trimmed base URL and apiKey as
// already resolved by resolveSecret (empty means keyless). It builds one
// shared *http.Client, reused for every request this adapter makes.
// shared *http.Client, reused for every request this adapter makes, with
// its Transport.ResponseHeaderTimeout set to defaultRequestTimeout — see
// the timeout field's own doc comment for why this constructor cannot
// leave timeout at its zero value the way it leaves retry at nil.
func newOpenAIAdapter(name, base, apiKey string) *openaiAdapter {
return &openaiAdapter{
adapterName: name,
baseURL: base,
apiKey: apiKey,
client: newAdapterHTTPClient(),
client: newAdapterHTTPClient(defaultRequestTimeout),
timeout: defaultRequestTimeout,
}
}
@@ -56,6 +71,9 @@ func (a *openaiAdapter) base() string { return a.baseURL }
// httpClient implements providerAdapter.
func (a *openaiAdapter) httpClient() *http.Client { return a.client }
// requestTimeout implements providerAdapter.
func (a *openaiAdapter) requestTimeout() time.Duration { return a.timeout }
// injectAuth implements providerAdapter: OpenAI's bearer-token scheme. A
// no-op when a.apiKey is empty, per the keyless-upstream ruling — a request
// against a keyless upstream must carry no Authorization header at all,
@@ -143,7 +161,7 @@ func (a *openaiAdapter) chatCompletion(ctx context.Context, w http.ResponseWrite
hdr.Set("Accept", "text/event-stream")
}
resp, err := upstreamJSON(ctx, a.client, http.MethodPost, a.baseURL+"/v1/chat/completions", hdr, req, a.retry)
resp, err := upstreamJSON(ctx, a.client, http.MethodPost, a.baseURL+"/v1/chat/completions", hdr, req, a.retry, a.timeout, a.adapterName)
if err != nil {
return usage{}, err
}
@@ -168,7 +186,7 @@ func (a *openaiAdapter) chatCompletion(ctx context.Context, w http.ResponseWrite
// non-streaming path.
func (a *openaiAdapter) embeddings(ctx context.Context, w http.ResponseWriter, req map[string]any) (usage, error) {
delete(req, gatewayAliasKey) // see chatCompletion's identical delete for why
resp, err := upstreamJSON(ctx, a.client, http.MethodPost, a.baseURL+"/v1/embeddings", a.requestHeaders(true), req, a.retry)
resp, err := upstreamJSON(ctx, a.client, http.MethodPost, a.baseURL+"/v1/embeddings", a.requestHeaders(true), req, a.retry, a.timeout, a.adapterName)
if err != nil {
return usage{}, err
}
@@ -277,7 +295,7 @@ func (a *openaiAdapter) imagesGeneration(ctx context.Context, w http.ResponseWri
// it — the same defensive posture chatCompletion/embeddings already
// take.
delete(req, gatewayAliasKey)
resp, err := upstreamJSON(ctx, a.client, http.MethodPost, a.baseURL+"/v1/images/generations", a.requestHeaders(true), req, a.retry)
resp, err := upstreamJSON(ctx, a.client, http.MethodPost, a.baseURL+"/v1/images/generations", a.requestHeaders(true), req, a.retry, a.timeout, a.adapterName)
if err != nil {
return usage{}, err
}
@@ -303,7 +321,7 @@ func (a *openaiAdapter) audioSpeech(ctx context.Context, w http.ResponseWriter,
if contentType != "" {
hdr.Set("Content-Type", contentType)
}
resp, err := upstreamRawBytes(ctx, a.client, http.MethodPost, a.baseURL+"/v1/audio/speech", hdr, body, a.retry)
resp, err := upstreamRawBytes(ctx, a.client, http.MethodPost, a.baseURL+"/v1/audio/speech", hdr, body, a.retry, a.timeout, a.adapterName)
if err != nil {
return usage{}, err
}
@@ -340,7 +358,7 @@ func (a *openaiAdapter) audioTranscription(ctx context.Context, w http.ResponseW
if contentType != "" {
hdr.Set("Content-Type", contentType)
}
resp, err := upstreamRawBytes(ctx, a.client, http.MethodPost, a.baseURL+"/v1/audio/transcriptions", hdr, body, a.retry)
resp, err := upstreamRawBytes(ctx, a.client, http.MethodPost, a.baseURL+"/v1/audio/transcriptions", hdr, body, a.retry, a.timeout, a.adapterName)
if err != nil {
return usage{}, err
}
@@ -381,7 +399,7 @@ type modelsPayload struct {
// listModels implements providerAdapter.
func (a *openaiAdapter) listModels(ctx context.Context) ([]string, error) {
resp, err := upstreamJSON(ctx, a.client, http.MethodGet, a.baseURL+"/v1/models", a.requestHeaders(false), nil, a.retry)
resp, err := upstreamJSON(ctx, a.client, http.MethodGet, a.baseURL+"/v1/models", a.requestHeaders(false), nil, a.retry, a.timeout, a.adapterName)
if err != nil {
return nil, err
}
@@ -444,7 +462,7 @@ func (a *openaiAdapter) fetchModelMetadata(ctx context.Context) (map[string]int,
return nil, nil
}
resp, err := upstreamJSON(ctx, a.client, http.MethodGet, a.baseURL+a.metadataPath, a.requestHeaders(false), nil, a.retry)
resp, err := upstreamJSON(ctx, a.client, http.MethodGet, a.baseURL+a.metadataPath, a.requestHeaders(false), nil, a.retry, a.timeout, a.adapterName)
if err != nil {
return nil, err
}
+100 -13
View File
@@ -10,6 +10,7 @@ import (
"net/http"
"regexp"
"strings"
"time"
)
// Provider type identifiers, matched against ProviderConfig.Type.
@@ -159,16 +160,35 @@ type providerAdapter interface {
// request. Named httpClient rather than client to avoid colliding with
// every implementation's own "client" struct field of the same name.
httpClient() *http.Client
// requestTimeout returns this adapter's resolved per-request timeout
// (Config.RequestTimeout or ProviderConfig.RequestTimeout,
// resolveProviderTimeout in timeout.go) — the idle-progress bound every
// upstream call through this adapter's shared client is held to.
// Exposed on the interface so a caller outside this file (native
// passthrough, routes_passthrough.go's handlePassthrough) can apply the
// identical watchdog proxyUpstream needs to its own resp.Body read,
// instead of the timeout only ever protecting upstreamJSON/
// upstreamRawBytes' own callers.
requestTimeout() time.Duration
}
// newAdapterHTTPClient returns an *http.Client for a provider adapter to
// hold and reuse across every request it makes: no client-level timeout,
// since a streaming chat completion can legitimately run for minutes and
// callers bound requests with a context deadline instead, and a Transport
// cloned from http.DefaultTransport with a larger per-host idle connection
// pool, since every request from one adapter targets the same upstream
// host.
func newAdapterHTTPClient() *http.Client {
// hold and reuse across every request it makes: no CLIENT-level timeout
// (Client.Timeout is deliberately left unset), since that would bound the
// whole request including body read — killing exactly the long streams
// this gateway must preserve (a 20-minute generation that keeps producing
// tokens must complete) — and a Transport cloned from http.DefaultTransport
// with a larger per-host idle connection pool, since every request from
// one adapter targets the same upstream host.
//
// timeout sets Transport.ResponseHeaderTimeout: the first of this
// feature's two mechanisms — it bounds only the wait for a response's
// headers, never the body, so it is safe for streaming. The second
// mechanism, bounding idle time WITHIN the body (a provider that sends
// headers and then stalls), is watchdogBody (timeout.go), applied by each
// caller of upstreamBytes/proxyUpstream to the *http.Response.Body this
// client's Do returns — Transport has no equivalent field for that half.
func newAdapterHTTPClient(timeout time.Duration) *http.Client {
tr, ok := http.DefaultTransport.(*http.Transport)
if ok {
tr = tr.Clone()
@@ -181,6 +201,7 @@ func newAdapterHTTPClient() *http.Client {
tr = &http.Transport{}
}
tr.MaxIdleConnsPerHost = adapterIdleConnsPerHost
tr.ResponseHeaderTimeout = timeout
return &http.Client{Transport: tr}
}
@@ -239,6 +260,12 @@ func attemptRecorderFromContext(ctx context.Context) attemptRecorder {
// listModels) have nothing to write through and every current caller
// checks resp.StatusCode itself.
//
// timeout and providerName (an adapter's own resolved requestTimeout()/
// name()) arm the idle-progress watchdog upstreamBytes wraps resp.Body
// in, once a response is actually obtained — see watchdogBody's own doc
// comment, timeout.go, for the mechanism and why it, not
// http.Client.Timeout, is what bounds a stalled body.
//
// policy (retry.go) wraps the whole request-and-receive-status exchange:
// a nil policy, or a disabled one, makes exactly one attempt (v0.1
// behavior, byte-identical). Every caller of upstreamJSON — chatCompletion,
@@ -250,7 +277,7 @@ func attemptRecorderFromContext(ctx context.Context) attemptRecorder {
// already returned its final, retried-or-not response, so a stream that
// dies while forwardStream is mid-flight never re-enters this function
// and is therefore never retried.
func upstreamJSON(ctx context.Context, client *http.Client, method, url string, hdr http.Header, body any, policy *retryPolicy) (*http.Response, error) {
func upstreamJSON(ctx context.Context, client *http.Client, method, url string, hdr http.Header, body any, policy *retryPolicy, timeout time.Duration, providerName string) (*http.Response, error) {
var bodyBytes []byte
if body != nil {
b, err := json.Marshal(body)
@@ -259,7 +286,7 @@ func upstreamJSON(ctx context.Context, client *http.Client, method, url string,
}
bodyBytes = b
}
return upstreamBytes(ctx, client, method, url, hdr, bodyBytes, policy)
return upstreamBytes(ctx, client, method, url, hdr, bodyBytes, policy, timeout, providerName)
}
// upstreamRawBytes is upstreamJSON's counterpart for a body the caller has
@@ -271,8 +298,8 @@ func upstreamJSON(ctx context.Context, client *http.Client, method, url string,
// multipart payload upstreamJSON's json.Marshal would corrupt outright.
// Shares upstreamBytes with upstreamJSON, so both get the identical
// retry/zero-bytes-reached invariant documented on upstreamJSON.
func upstreamRawBytes(ctx context.Context, client *http.Client, method, url string, hdr http.Header, body []byte, policy *retryPolicy) (*http.Response, error) {
return upstreamBytes(ctx, client, method, url, hdr, body, policy)
func upstreamRawBytes(ctx context.Context, client *http.Client, method, url string, hdr http.Header, body []byte, policy *retryPolicy, timeout time.Duration, providerName string) (*http.Response, error) {
return upstreamBytes(ctx, client, method, url, hdr, body, policy, timeout, providerName)
}
// upstreamBytes is the shared request-build-and-send core behind both
@@ -282,12 +309,33 @@ func upstreamRawBytes(ctx context.Context, client *http.Client, method, url stri
// request body unchanged, and Content-Type defaults to
// "application/json" only when bodyBytes is non-nil and hdr set none of
// its own.
func upstreamBytes(ctx context.Context, client *http.Client, method, url string, hdr http.Header, bodyBytes []byte, policy *retryPolicy) (*http.Response, error) {
func upstreamBytes(ctx context.Context, client *http.Client, method, url string, hdr http.Header, bodyBytes []byte, policy *retryPolicy, timeout time.Duration, providerName string) (*http.Response, error) {
// rec is looked up once, outside the retry loop's call closure: every
// attempt shares the identical ctx, so the recorder (nil when no
// caller ever wired one via withAttemptRecorder) is the same across
// attempts. watchdogBody uses it too (coordinator adversarial review,
// 2026-08-23, finding F4): retryPolicy.do's own rec(resp, err) call
// (retry.go) already ran, at header-arrival time, before any byte of
// the body was read — a mid-body stall was otherwise invisible to
// provider-health accounting, since nothing else ever called rec
// again for that attempt.
rec := attemptRecorderFromContext(ctx)
call := func() (*http.Response, error) {
var r io.Reader
if bodyBytes != nil {
r = bytes.NewReader(bodyBytes)
}
// reqCtx/cancel bound this attempt's response body to timeout via
// watchdogBody (timeout.go), derived fresh per attempt so a retried
// request gets its own full timeout window rather than inheriting
// an already-ticking one. cancel is released either by
// watchdogBody (once resp.Body exists — see its own Close/fire) or
// directly below, on a build/send failure that never produced a
// body to wrap.
reqCtx, cancel := context.WithCancel(ctx)
// gosec G704 (SSRF via taint analysis) flags url below: every call
// site builds it from an operator-configured value —
// adapter.baseURL (ProviderConfig.BaseURL, or defaultBaseURLByType
@@ -299,8 +347,9 @@ func upstreamBytes(ctx context.Context, client *http.Client, method, url string,
// proxyUpstream for the one adapter call path where a request-
// derived path segment genuinely reaches the outgoing URL, and its
// own comment for why that case is safe.
req, err := http.NewRequestWithContext(ctx, method, url, r) //nolint:gosec // operator-configured host; see comment above
req, err := http.NewRequestWithContext(reqCtx, method, url, r) //nolint:gosec // operator-configured host; see comment above
if err != nil {
cancel()
return nil, fmt.Errorf("%w: %w: build request: %w", errUpstream, errRequestBuildFailed, err)
}
for k, vs := range hdr {
@@ -314,8 +363,10 @@ func upstreamBytes(ctx context.Context, client *http.Client, method, url string,
resp, err := client.Do(req) //nolint:bodyclose,gosec // caller closes resp.Body; upstreamBytes hands the response, not its lifecycle, back — same operator-configured URL as above
if err != nil {
cancel()
return nil, fmt.Errorf("%w: %w", errUpstream, err)
}
resp.Body = newWatchdogBody(resp.Body, cancel, timeout, fmt.Sprintf("provider %q", providerName), rec)
return resp, nil
}
@@ -432,6 +483,18 @@ func buildAdapters(cfg *Config) (map[string]providerAdapter, error) {
return nil, err
}
// globalTimeout is resolved once, up front, so a malformed
// Config.RequestTimeout fails construction even for a deployment where
// every provider happens to set its own ProviderConfig.RequestTimeout
// override — resolveProviderTimeout's own doc comment (timeout.go).
// timeoutErr, not err (govet shadow): a second err reused across this
// function's top-level scope made every err re-declared inside the
// provider loop below read as a suspicious shadow.
globalTimeout, timeoutErr := resolveRequestTimeout(cfg.RequestTimeout, "requestTimeout")
if timeoutErr != nil {
return nil, timeoutErr
}
adapters := make(map[string]providerAdapter, len(cfg.Providers))
for name, pc := range cfg.Providers {
if pc == nil {
@@ -454,11 +517,31 @@ func buildAdapters(cfg *Config) (map[string]providerAdapter, error) {
return nil, err
}
// timeoutErr, not err (govet shadow — see globalTimeout's identical
// note above): this loop already declares its own err via
// apiKey, err := resolveSecret above; reusing it here for a second,
// later purpose is what made validateMetadataPath's unrelated
// if err := ...; err != nil read as a suspicious shadow of it.
timeout, timeoutErr := resolveProviderTimeout(name, pc.RequestTimeout, globalTimeout)
if timeoutErr != nil {
return nil, timeoutErr
}
switch pc.Type {
case providerTypeOpenAI:
a := newOpenAIAdapter(name, base, apiKey)
a.retry = policy
a.metadataPath = pc.MetadataPath
// timeout/client are re-resolved here rather than left at the
// bare constructor's defaultRequestTimeout default (mirroring
// how a.retry/a.metadataPath are also only ever set to their
// real, config-derived values here, not by newOpenAIAdapter
// itself) — client is rebuilt, not mutated in place, since
// Transport.ResponseHeaderTimeout is set once at construction
// (newAdapterHTTPClient, above) and every adapter's Transport
// is otherwise identical.
a.timeout = timeout
a.client = newAdapterHTTPClient(timeout)
adapters[name] = a
case providerTypeAnthropic:
a, err := newAnthropicAdapter(name, base, apiKey)
@@ -466,6 +549,8 @@ func buildAdapters(cfg *Config) (map[string]providerAdapter, error) {
return nil, err
}
a.retry = policy
a.timeout = timeout
a.client = newAdapterHTTPClient(timeout)
adapters[name] = a
case providerTypeGemini:
g, err := newGeminiAdapter(name, base, apiKey)
@@ -473,6 +558,8 @@ func buildAdapters(cfg *Config) (map[string]providerAdapter, error) {
return nil, err
}
g.retry = policy
g.timeout = timeout
g.client = newAdapterHTTPClient(timeout)
adapters[name] = g
default:
return nil, fmt.Errorf("llmgateway: provider %q: unknown type %q", name, pc.Type)
+23 -4
View File
@@ -52,7 +52,8 @@ func (f *fakeAdapter) listModels(ctx context.Context) ([]string, error) {
}
return f.listModelsFn(ctx)
}
func (f *fakeAdapter) httpClient() *http.Client { return nil }
func (f *fakeAdapter) httpClient() *http.Client { return nil }
func (f *fakeAdapter) requestTimeout() time.Duration { return defaultRequestTimeout }
// recordingLog is a thread-safe log func for tests, capturing formatted
// lines instead of writing to stderr.
@@ -504,8 +505,26 @@ func TestModelRegistry_MaybeRefresh_Throttled_OnePerInterval(t *testing.T) {
if err != nil {
t.Fatalf("newModelRegistry: %v", err)
}
now := time.Now()
reg.nowFn = func() time.Time { return now }
// clockMu guards clock: reg.nowFn runs on maybeRefresh's own
// background refresh goroutine (registry.go), concurrently with this
// test's own goroutine advancing the clock below a few lines down. A
// plain `now` variable closed over by both was a genuine data race
// (coordinator adversarial review, 2026-08-23) — the 10ms sleep below
// gave the background goroutine a chance to finish reading it first,
// but that is a race won by scheduling luck, not a happens-before
// guarantee, and -race caught it in roughly 14% of full-suite runs.
var clockMu sync.Mutex
clock := time.Now()
reg.nowFn = func() time.Time {
clockMu.Lock()
defer clockMu.Unlock()
return clock
}
advanceClock := func(d time.Duration) {
clockMu.Lock()
clock = clock.Add(d)
clockMu.Unlock()
}
reg.maybeRefresh(context.Background())
waitUntil(t, time.Second, func() bool { return atomic.LoadInt32(&calls) == 1 })
@@ -517,7 +536,7 @@ func TestModelRegistry_MaybeRefresh_Throttled_OnePerInterval(t *testing.T) {
t.Errorf("listModels called %d times within one interval, want 1", got)
}
now = now.Add(2 * time.Hour) // past the interval: next entry should refetch
advanceClock(2 * time.Hour) // past the interval: next entry should refetch
reg.maybeRefresh(context.Background())
waitUntil(t, time.Second, func() bool { return atomic.LoadInt32(&calls) == 2 })
}
+1 -1
View File
@@ -616,7 +616,7 @@ func TestUpstreamJSON_RequestBuildFailure_NotRetried(t *testing.T) {
return true
}}
client := &http.Client{}
_, err := upstreamJSON(context.Background(), client, "IN VALID", "http://127.0.0.1:1/x", nil, nil, p)
_, err := upstreamJSON(context.Background(), client, "IN VALID", "http://127.0.0.1:1/x", nil, nil, p, defaultRequestTimeout, "p1")
if err == nil {
t.Fatal("upstreamJSON: want an error for an invalid method, got nil")
}
+1 -1
View File
@@ -222,7 +222,7 @@ func (g *Gateway) callAnthropicMessagesPassthrough(ctx context.Context, w http.R
// documented, not incidental.
adapter.injectAuth(&http.Request{Header: hdr})
resp, err := upstreamJSON(ctx, adapter.httpClient(), http.MethodPost, adapter.base()+anthropicMessagesPath, hdr, req, retry)
resp, err := upstreamJSON(ctx, adapter.httpClient(), http.MethodPost, adapter.base()+anthropicMessagesPath, hdr, req, retry, adapter.requestTimeout(), adapter.name())
if err != nil {
return usage{}, err
}
+44 -5
View File
@@ -9,6 +9,7 @@ import (
"net/http"
"net/url"
"strings"
"time"
)
// maxPassthroughBytes caps a client passthrough request body: 32MiB,
@@ -862,7 +863,7 @@ func (g *Gateway) handlePassthrough(w http.ResponseWriter, r *http.Request, u *u
g.limiter.recordProviderAttempt(providerName, "", resp, attemptErr)
}))
result, ok := g.proxyUpstream(w, r, upstreamURL, adapter.httpClient(), adapter.injectAuth, providerCredentialRetargetHeaders, true, "passthrough (provider "+providerName+")")
result, ok := g.proxyUpstream(w, r, upstreamURL, adapter.httpClient(), adapter.injectAuth, providerCredentialRetargetHeaders, true, "passthrough (provider "+providerName+")", adapter.requestTimeout())
if !ok || !result.isJSON {
// A build/connection/copy failure already wrote its own response
// (or, for a canceled client context, wrote nothing at all — see
@@ -952,7 +953,16 @@ type proxyResult struct {
// passes false, since target-proxy accounting never goes past the
// request-count checkAndCount already ran before calling in, and teeing a
// response nobody will ever read back would only cost memory for nothing.
func (g *Gateway) proxyUpstream(w http.ResponseWriter, r *http.Request, upstreamURL string, client *http.Client, injectAuth func(*http.Request), extraStrip map[string]bool, accountJSON bool, logPrefix string) (result proxyResult, ok bool) {
//
// timeout (feature: request timeout) arms an idle-progress watchdog
// (watchdogBody, timeout.go) around resp.Body: handlePassthrough passes
// its adapter's own requestTimeout(), handleTargetProxy passes
// g.targetTimeout (mcp_a2a.go) — both share the SAME defect this feature
// fixes (client, above, built via newAdapterHTTPClient, previously had no
// timeout at all), so both are covered the same way. The Transport-level
// half (ResponseHeaderTimeout) is already set on client itself and needs
// no separate wiring here.
func (g *Gateway) proxyUpstream(w http.ResponseWriter, r *http.Request, upstreamURL string, client *http.Client, injectAuth func(*http.Request), extraStrip map[string]bool, accountJSON bool, logPrefix string, timeout time.Duration) (result proxyResult, ok bool) {
bodyReader := io.LimitReader(r.Body, maxPassthroughBytes)
// gosec G704 (SSRF via taint analysis) flags upstreamURL as
// request-derived: it is, by design — this is a reverse proxy, and its
@@ -971,8 +981,15 @@ func (g *Gateway) proxyUpstream(w http.ResponseWriter, r *http.Request, upstream
// a client cannot smuggle an encoded "/" (%2f) past this gateway's own
// routing only to have a permissive upstream reinterpret it as a real
// separator.
upstreamReq, err := http.NewRequestWithContext(r.Context(), r.Method, upstreamURL, bodyReader) //nolint:gosec // operator-fixed host, rest is traversal-checked and forwarded escaped; see comment above
// reqCtx/cancel bound the response body to timeout via watchdogBody
// (timeout.go), the same idle-progress mechanism upstreamBytes
// (providers.go) applies to every provider-adapter call — cancel is
// released either by watchdogBody (once resp.Body exists) or directly
// below, on a build/send failure that never produced a body to wrap.
reqCtx, cancel := context.WithCancel(r.Context())
upstreamReq, err := http.NewRequestWithContext(reqCtx, r.Method, upstreamURL, bodyReader) //nolint:gosec // operator-fixed host, rest is traversal-checked and forwarded escaped; see comment above
if err != nil {
cancel()
g.errorf("%s: build upstream request: %v", logPrefix, err)
writeOAIError(w, http.StatusBadGateway, "server_error", "upstream connection error")
return proxyResult{}, false
@@ -997,11 +1014,17 @@ func (g *Gateway) proxyUpstream(w http.ResponseWriter, r *http.Request, upstream
// carries a recorder (attemptRecorderFromContext, providers.go) —
// handleTargetProxy (mcp_a2a.go), this function's other caller, never
// wraps r's context this way, so an MCP/A2A target proxy attempt is
// correctly never accounted as provider traffic.
if rec := attemptRecorderFromContext(r.Context()); rec != nil {
// correctly never accounted as provider traffic. rec is kept in scope
// (not just checked inline) so the watchdogBody below can reuse it
// too — a mid-body stall must reach the SAME provider-health
// accounting a build/send failure already does (coordinator
// adversarial review, 2026-08-23, finding F4).
rec := attemptRecorderFromContext(r.Context())
if rec != nil {
rec(resp, err)
}
if err != nil {
cancel()
if errors.Is(err, context.Canceled) {
g.logf("%s: client canceled request: %v", logPrefix, err)
return proxyResult{}, false
@@ -1010,6 +1033,11 @@ func (g *Gateway) proxyUpstream(w http.ResponseWriter, r *http.Request, upstream
writeOAIError(w, http.StatusBadGateway, "server_error", "upstream connection error")
return proxyResult{}, false
}
// logPrefix, not a hardcoded "provider %q": handleTargetProxy's own
// calls here (mcp_a2a.go) pass "mcp target (name ...)"/"a2a target
// (name ...)" — an MCP/A2A target is not a provider, and the watchdog
// error text must not claim it is (finding F9).
resp.Body = newWatchdogBody(resp.Body, cancel, timeout, logPrefix, rec)
defer resp.Body.Close() //nolint:errcheck // read-side close; nothing actionable on failure
copyHeadersExcept(w.Header(), resp.Header, hopByHopHeaders, dangerousResponseHeaders)
@@ -1031,6 +1059,17 @@ func (g *Gateway) proxyUpstream(w http.ResponseWriter, r *http.Request, upstream
}
if _, err := io.Copy(fw, reader); err != nil {
// errors.Is(err, context.Canceled) here means the CLIENT went away
// mid-copy — reader wraps resp.Body, which is now a watchdogBody
// (above): its own Read never lets a timeout error wrap
// context.Canceled (watchdogBody's own doc comment, timeout.go),
// so this check still correctly separates a genuine client
// disconnect from this gateway's own idle-progress watchdog firing
// — the latter falls through to the errorf below, unchanged.
if errors.Is(err, context.Canceled) {
g.logf("%s: client canceled request: %v", logPrefix, err)
return proxyResult{isJSON: isJSON, tee: tee}, false
}
g.errorf("%s: stream response body: %v", logPrefix, err)
return proxyResult{isJSON: isJSON, tee: tee}, false
}
+255
View File
@@ -0,0 +1,255 @@
package traefikllmgateway
import (
"context"
"errors"
"fmt"
"io"
"sync/atomic"
"time"
)
// defaultRequestTimeout is the progress-based upstream request timeout
// applied when neither Config.RequestTimeout nor a provider's own
// ProviderConfig.RequestTimeout is set: five minutes.
//
// Before this feature, newAdapterHTTPClient (providers.go) set no timeout
// at all — Client.Timeout was left unset, and the Transport cloned from
// http.DefaultTransport set no ResponseHeaderTimeout either. Only the dial
// (30s) and TLS handshake (10s) were bounded. A provider that accepted the
// connection and then hung — sent no headers, or sent headers and then
// stalled mid-body — hung the gateway request forever; nothing recovered
// it. This constant is a deliberate behavior change for every deployment,
// not an invisible bug fix — see README.md's "Request timeout" section.
const defaultRequestTimeout = 5 * time.Minute
// errProviderTimeout is the sentinel wrapped into the error a watchdogBody
// (below) returns once its idle-progress watchdog fires: a period of the
// resolved timeout passed with no successful read from the upstream
// response body. Distinct from context.Canceled — a genuine client
// disconnect, classified separately by handleAdapterErrorEnvelope
// (routes_unified.go) and proxyUpstream (routes_passthrough.go), both of
// which check errors.Is(err, context.Canceled) ahead of their generic
// upstream-error fallback — watchdogBody.Read never lets a timeout error
// wrap context.Canceled, so the two stay mutually exclusive in every
// caller's classification.
//
// The message deliberately says "upstream progress stalled", not
// "provider timed out" (coordinator adversarial review, 2026-08-23,
// finding F3): the watchdog measures time since this gateway's last
// successful READ from the upstream body, which can go quiet either
// because the provider itself stalled, or because this gateway spent the
// whole window blocked WRITING to a slow/stalled client between reads (a
// streaming forwarder reads one chunk, writes it out, then reads again —
// the write is not itself watchdog-guarded). Aborting in either case is
// still correct (holding an upstream connection open for a stalled
// client is its own problem), but asserting the PROVIDER specifically
// caused it is not always true, so the message no longer claims it.
var errProviderTimeout = errors.New("llmgateway: upstream progress stalled")
// resolveRequestTimeout parses raw (a Go duration string, e.g. "5m" or
// "90s" — matching RetryConfig.Backoff/CacheConfig.TTL's own convention)
// and returns defaultRequestTimeout when raw is empty: Config.
// RequestTimeout and ProviderConfig.RequestTimeout's own zero value.
//
// A non-empty raw that fails to parse, or parses to a duration <= 0, is a
// construction error. This is stricter than CacheConfig.TTL (validated by
// validateCacheConfig, cache.go), which only rejects a malformed string,
// never a non-positive one: a request timeout of zero or less is never a
// meaningful value for THIS feature specifically, because it exists only
// to fix a hang bug (newAdapterHTTPClient's pre-feature shape, described
// on defaultRequestTimeout above) — an operator writing "0s" hoping to
// restore "wait forever" must fail loudly at construction, not silently
// reintroduce the exact hang this feature exists to close. label names
// the config field in the resulting error message (e.g. "requestTimeout"
// or `provider "openai": requestTimeout`) so a bad value is attributable
// to its source.
func resolveRequestTimeout(raw, label string) (time.Duration, error) {
if raw == "" {
return defaultRequestTimeout, nil
}
d, err := time.ParseDuration(raw)
if err != nil {
return 0, fmt.Errorf("llmgateway: %s: %w", label, err)
}
if d <= 0 {
return 0, fmt.Errorf("llmgateway: %s must be positive, got %q", label, raw)
}
return d, nil
}
// resolveProviderTimeout resolves the request timeout for one provider:
// providerRaw (ProviderConfig.RequestTimeout) when non-empty, validated
// via resolveRequestTimeout under a provider-attributed label — an
// explicit per-provider override always wins over the global default,
// house rule. Otherwise globalTimeout, the caller's already-resolved
// Config.RequestTimeout (buildAdapters, providers.go, resolves it once up
// front via resolveRequestTimeout, so a malformed global value fails
// construction even for a deployment where every provider happens to set
// its own override — a config bug must not lie dormant only to surface
// once someone adds a provider that relies on the default).
func resolveProviderTimeout(providerName, providerRaw string, globalTimeout time.Duration) (time.Duration, error) {
if providerRaw == "" {
return globalTimeout, nil
}
return resolveRequestTimeout(providerRaw, fmt.Sprintf("provider %q: requestTimeout", providerName))
}
// watchdogBody wraps an upstream response body so that a period of
// timeout with no successful read aborts the request instead of blocking
// forever. It is the idle-progress half of this feature's two mechanisms
// (Transport.ResponseHeaderTimeout, set in newAdapterHTTPClient, is the
// other — it only bounds the wait for headers, never the body).
//
// Deliberately NOT http.Client.Timeout: that bounds the whole request
// including the body read, which would truncate a healthy, long-running
// stream — exactly what this feature must never do. A watchdogBody
// instead resets on every successful, non-empty read, so a stream (or a
// large non-streaming body) that keeps making progress runs indefinitely;
// only SILENCE, never total duration, ends it.
//
// The mechanism: cancel is the CancelFunc for the *http.Request's own
// context, derived via context.WithCancel by the caller that built the
// request (upstreamBytes, providers.go; proxyUpstream, routes_passthrough.
// go) before this watchdogBody existed. Every successful Read resets timer
// to fire timeout in the future again; if timer fires first, fire cancels
// the request context, which unblocks whatever Read is currently in
// flight (or the next one) with an error. Read then substitutes that
// error for one wrapping errProviderTimeout, naming label, instead of
// passing through whatever the canceled context's own error text says —
// a real client disconnect (the parent context canceling instead of this
// watchdog) never sets timedOut, so it is never misreported this way.
// Close stops the timer (a no-op once already fired) and always calls
// cancel, so this watchdog never leaves a live timer or an un-canceled
// derived context behind on the hot path — every upstream request this
// gateway makes.
//
// onTimeout, when non-nil, is invoked once from fire (coordinator
// adversarial review, 2026-08-23, finding F4): a provider that sends
// headers and then stalls mid-body previously left provider-health
// accounting blind to the failure entirely — limiter.recordProviderAttempt
// is only ever reached from retryPolicy.do's own attemptRecorder call (at
// header-arrival time, before any body byte is read) and from
// proxyUpstream's identical call — so a chronic mid-body staller looked
// like 100% successful attempts to the discovery breaker and any
// request-path health signal that consumes those counters. onTimeout is
// that same attemptRecorder, reused rather than reinvented: callers look
// it up once via attemptRecorderFromContext and pass it straight through.
// This does mean a request that stalls mid-body is recorded as TWO
// attempts (one success at header time, one failure when the watchdog
// fires) rather than one attempt marked failed outright — deliberately:
// the alternative was deferring the FIRST recorder call until the body
// finishes, which would change when every other (already-tested,
// already-shipped) success path reports its attempt too, a much larger
// blast radius for what this fix needs. Two real, honestly-reported
// signals (headers arrived; the body then never finished) beats the
// previous zero.
//
// timedOut is accessed via sync/atomic's function API (atomic.StoreInt32/
// LoadInt32), not the newer atomic.Bool type, per this plugin's
// yaegi-interpreted-stdlib guardrail.
type watchdogBody struct {
rc io.ReadCloser
cancel context.CancelFunc
timer *time.Timer
onTimeout attemptRecorder
label string
timeout time.Duration
timedOut int32
}
// newWatchdogBody returns a watchdogBody wrapping rc, arming its watchdog
// for timeout starting now — the moment the response's headers arrived,
// so even the very first body read is bounded by the same window ("sends
// headers then stalls mid-body" is caught here, not left to
// ResponseHeaderTimeout, which has already been satisfied by the time a
// body exists to wrap).
//
// label names what this watchdog guards, already formatted by the
// caller: providers.go passes `provider "openai"`; routes_passthrough.go
// passes its own logPrefix verbatim (`passthrough (provider "openai")` or
// `mcp target (name "foo")`) — NOT reworded here into a hardcoded
// "provider %q", which previously mislabeled an MCP/A2A target as a
// provider (coordinator adversarial review, 2026-08-23, finding F9).
// onTimeout may be nil (no attemptRecorder was ever wired onto this
// request's context — the common case for discovery's own listModels
// calls and the MCP/A2A target proxy, neither of which Feature A
// accounts).
func newWatchdogBody(rc io.ReadCloser, cancel context.CancelFunc, timeout time.Duration, label string, onTimeout attemptRecorder) *watchdogBody {
wb := &watchdogBody{rc: rc, cancel: cancel, timeout: timeout, label: label, onTimeout: onTimeout}
wb.timer = time.AfterFunc(timeout, wb.fire)
return wb
}
// fire runs once, in its own goroutine (time.AfterFunc's own contract —
// no goroutine is ever spawned for a request whose timer is Stopped
// before it fires, which is the common, non-timed-out case), the moment
// timeout passes with no successful Read: it marks timedOut so the next
// Read error substitutes the timeout error, cancels the request context
// (unblocking whatever Read is currently parked waiting on the network),
// and — finding F4 — notifies onTimeout, if this watchdog has one, so
// provider-health accounting sees the failure even though no caller will
// ever read a body byte past this point.
func (wb *watchdogBody) fire() {
atomic.StoreInt32(&wb.timedOut, 1)
wb.cancel()
if wb.onTimeout != nil {
wb.onTimeout(nil, wb.timeoutError())
}
}
// timeoutError builds the error Read returns once timedOut is set, and
// the error fire hands onTimeout — one construction, reused both places,
// so the two can never drift apart. Wraps errProviderTimeout only, not
// errUpstream: every caller that reads a body to completion already
// wraps errUpstream around whatever Read returns (forwardJSON/
// forwardStream/forwardMediaBody/listModels, providers.go and the
// translate-aware adapters alike), so wrapping it again here produced a
// doubled "upstream request failed: ... upstream request failed: ..."
// message (finding F9) with no compensating benefit — errors.Is(err,
// errUpstream) still holds once the caller's own wrap runs.
func (wb *watchdogBody) timeoutError() error {
return fmt.Errorf("%w: %s: no upstream progress for %s", errProviderTimeout, wb.label, wb.timeout)
}
// Read implements io.Reader. A successful, non-empty read (err == nil and
// n > 0) resets the watchdog for another full timeout window — this is
// what lets a stream that keeps producing chunks run indefinitely: only
// silence, never total duration, ends it. n == 0 with err == nil (a
// legal, if unusual, io.Reader response — not currently produced by
// net/http's own body reader against a non-empty buffer, but not
// forbidden by the io.Reader contract either) does NOT reset the timer
// (finding F6): resetting on a read that delivered nothing would let a
// reader that keeps returning (0, nil) hold the watchdog open forever,
// exactly the naive-watchdog hole this feature exists to close. A failed
// read while timedOut is set (fire already ran) is reported via
// timeoutError, instead of whatever error canceling the request context
// actually produced. Any other error — a real client disconnect, a clean
// EOF, a genuine network failure that has nothing to do with this
// watchdog — passes through unchanged.
func (wb *watchdogBody) Read(p []byte) (int, error) {
n, err := wb.rc.Read(p)
if err != nil {
if atomic.LoadInt32(&wb.timedOut) == 1 {
return n, wb.timeoutError()
}
return n, err
}
if n > 0 {
wb.timer.Reset(wb.timeout)
}
return n, nil
}
// Close stops the watchdog timer (a harmless no-op if it already fired)
// and always calls cancel, so the derived request context this
// watchdogBody's cancel belongs to is released promptly — whether Close
// is reached because the body was read to completion, because the caller
// gave up early, or because the watchdog itself already fired — instead
// of being left to be reclaimed only once its parent context ends. Then
// closes the wrapped body.
func (wb *watchdogBody) Close() error {
wb.timer.Stop()
wb.cancel()
return wb.rc.Close()
}
+886
View File
@@ -0,0 +1,886 @@
package traefikllmgateway
import (
"context"
"errors"
"fmt"
"io"
"net/http"
"net/http/httptest"
"runtime"
"strings"
"sync"
"sync/atomic"
"testing"
"time"
)
// newTimeoutTestAdapter returns an *openaiAdapter pointed at baseURL with
// its request timeout resolved to timeout — mirroring exactly what
// buildAdapters (providers.go) does after newOpenAIAdapter's own bare
// construction: set .timeout, then rebuild .client via
// newAdapterHTTPClient(timeout) so Transport.ResponseHeaderTimeout
// reflects it too.
func newTimeoutTestAdapter(baseURL string, timeout time.Duration) *openaiAdapter {
a := newOpenAIAdapter("p1", baseURL, "sk-test")
a.timeout = timeout
a.client = newAdapterHTTPClient(timeout)
return a
}
// newStallingServer returns an httptest.Server whose single handler runs
// prewrite (nil is a no-op — writes nothing) and then blocks until
// release is called.
//
// Deliberately NOT `<-r.Context().Done()`: verified empirically that,
// in this environment, neither a client-side Transport.
// ResponseHeaderTimeout nor a client-side context.CancelFunc reliably
// unblocks a server handler parked on its own r.Context().Done() within
// a bounded time — httptest.Server.Close hung past 60s waiting for the
// connection in both cases. That is a property of net/http's server-side
// close detection under httptest, not of this feature's client-side
// behavior (which every test below still verifies independently, via the
// CLIENT-side call returning promptly). release decouples test cleanup
// from that detection entirely: the handler exits the instant the test
// explicitly says so, so Close never has anything left to wait for.
func newStallingServer(prewrite func(http.ResponseWriter)) (srv *httptest.Server, release func()) {
done := make(chan struct{})
srv = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if prewrite != nil {
prewrite(w)
}
<-done
}))
return srv, func() {
close(done)
srv.Close()
}
}
// TestChatCompletion_StreamKeepsEmittingPastTimeout_CompletesIntact is the
// single highest-priority test in this feature (written first, per the
// task brief): a streaming response whose TOTAL duration exceeds the
// configured timeout, but whose every INTER-CHUNK gap stays well under
// it, must complete intact — this is the exact behavior a total-duration
// cap (explicitly rejected by the operator) would break, and the whole
// reason the watchdog resets on every successful read instead of firing
// once at a fixed deadline.
//
// MUTATION VERIFIED: commenting out watchdogBody.Read's `wb.timer.Reset(
// wb.timeout)` (timeout.go) turns the idle watchdog into a de facto
// total-duration cap — the timer set at construction fires once, at
// timeout after the first byte, regardless of intervening progress. Under
// that mutation this test failed with "chatCompletion: ... provider
// request timed out" partway through the stream, confirming it kills the
// exact regression this test exists to catch. Reverted before committing.
func TestChatCompletion_StreamKeepsEmittingPastTimeout_CompletesIntact(t *testing.T) {
const (
timeout = 80 * time.Millisecond
chunkGap = 30 * time.Millisecond
numChunks = 6 // 6*30ms = 180ms total: well past timeout, but no single gap is
)
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "text/event-stream")
w.WriteHeader(http.StatusOK)
fl := w.(http.Flusher)
for i := 0; i < numChunks; i++ {
time.Sleep(chunkGap)
_, _ = fmt.Fprintf(w, "data: {\"id\":\"c\",\"choices\":[{\"delta\":{\"content\":\"tok%d\"}}]}\n\n", i)
fl.Flush()
}
_, _ = fmt.Fprint(w, "data: [DONE]\n\n")
fl.Flush()
}))
defer srv.Close()
a := newTimeoutTestAdapter(srv.URL, timeout)
rec := httptest.NewRecorder()
if _, err := a.chatCompletion(context.Background(), rec, map[string]any{"model": "gpt-5", "messages": []any{}, "stream": true}); err != nil {
t.Fatalf("chatCompletion: %v (a stream whose total duration exceeds the timeout, but whose every gap stays under it, must complete — a total-duration cap was explicitly rejected)", err)
}
body := rec.Body.String()
for i := 0; i < numChunks; i++ {
want := fmt.Sprintf("tok%d", i)
if !strings.Contains(body, want) {
t.Errorf("rec.Body missing %q — stream was truncated: %s", want, body)
}
}
if !strings.Contains(body, "[DONE]") {
t.Errorf("rec.Body missing [DONE] terminator — stream did not complete: %s", body)
}
}
// TestChatCompletion_NeverSendsHeaders_AbortedAtTimeout proves the
// Transport.ResponseHeaderTimeout mechanism (newAdapterHTTPClient,
// providers.go): a provider that accepts the connection and then writes
// nothing at all is aborted, not left hanging forever.
//
// MUTATION VERIFIED: removing `tr.ResponseHeaderTimeout = timeout` from
// newAdapterHTTPClient (providers.go) made this test hang past a 5s
// `go test -run ... -timeout 5s` bound (test process killed, reported
// FAIL) — confirming no other mechanism bounds a provider that never
// sends headers. Reverted before committing.
func TestChatCompletion_NeverSendsHeaders_AbortedAtTimeout(t *testing.T) {
const timeout = 60 * time.Millisecond
srv, release := newStallingServer(nil)
defer release()
a := newTimeoutTestAdapter(srv.URL, timeout)
rec := httptest.NewRecorder()
start := time.Now()
_, err := a.chatCompletion(context.Background(), rec, map[string]any{"model": "gpt-5", "messages": []any{}})
elapsed := time.Since(start)
if err == nil {
t.Fatal("chatCompletion: want an error for a provider that never sends headers, got nil")
}
if !errors.Is(err, errUpstream) {
t.Errorf("err = %v, want it to wrap errUpstream", err)
}
if elapsed > 2*time.Second {
t.Errorf("chatCompletion took %s to abort a %s ResponseHeaderTimeout", elapsed, timeout)
}
}
// TestChatCompletion_Stream_StallsMidBody_AbortedAtTimeout proves the
// idle-progress watchdog (watchdogBody, timeout.go): a provider that
// sends headers and one chunk, then goes silent, is aborted — proving
// ResponseHeaderTimeout alone (already satisfied once headers arrived) is
// not enough, and something must also bound the body.
//
// MUTATION VERIFIED: deleting the `resp.Body = newWatchdogBody(...)` line
// in upstreamBytes (providers.go) made this test hang past a 5s
// `-timeout 5s` bound (same failure mode as the ResponseHeaderTimeout
// test above, confirming this is a genuinely separate mechanism).
// Reverted before committing.
func TestChatCompletion_Stream_StallsMidBody_AbortedAtTimeout(t *testing.T) {
const timeout = 60 * time.Millisecond
srv, release := newStallingServer(func(w http.ResponseWriter) {
w.Header().Set("Content-Type", "text/event-stream")
w.WriteHeader(http.StatusOK)
fl := w.(http.Flusher)
_, _ = fmt.Fprint(w, "data: {\"id\":\"c\",\"choices\":[{\"delta\":{\"content\":\"hi\"}}]}\n\n")
fl.Flush()
})
defer release()
a := newTimeoutTestAdapter(srv.URL, timeout)
rec := httptest.NewRecorder()
start := time.Now()
_, err := a.chatCompletion(context.Background(), rec, map[string]any{"model": "gpt-5", "messages": []any{}, "stream": true})
elapsed := time.Since(start)
if err == nil {
t.Fatal("chatCompletion: want an error for a stream that stalls mid-body, got nil")
}
if !errors.Is(err, errProviderTimeout) {
t.Errorf("err = %v, want errors.Is(err, errProviderTimeout)", err)
}
if elapsed > 2*time.Second {
t.Errorf("chatCompletion took %s to abort a %s idle timeout", elapsed, timeout)
}
}
// TestChatCompletion_NonStream_StallsMidBody_AbortedAtTimeout is the
// non-streaming counterpart: "Non-streaming: at most timeout waiting for
// the response" (the task brief) is satisfied by the SAME watchdogBody
// mechanism as streaming — forwardJSON's io.ReadAll blocks on Read just
// like forwardStream's readSSE does, so one mechanism covers both
// documented behaviors without a second, total-duration-shaped one.
func TestChatCompletion_NonStream_StallsMidBody_AbortedAtTimeout(t *testing.T) {
const timeout = 60 * time.Millisecond
srv, release := newStallingServer(func(w http.ResponseWriter) {
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(http.StatusOK)
fl := w.(http.Flusher)
_, _ = fmt.Fprint(w, `{"id":"c1",`) // deliberately incomplete JSON: a real close never arrives
fl.Flush()
})
defer release()
a := newTimeoutTestAdapter(srv.URL, timeout)
rec := httptest.NewRecorder()
start := time.Now()
_, err := a.chatCompletion(context.Background(), rec, map[string]any{"model": "gpt-5", "messages": []any{}})
elapsed := time.Since(start)
if err == nil {
t.Fatal("chatCompletion: want an error for a non-streaming body that stalls mid-write, got nil")
}
if !errors.Is(err, errProviderTimeout) {
t.Errorf("err = %v, want errors.Is(err, errProviderTimeout)", err)
}
if elapsed > 2*time.Second {
t.Errorf("chatCompletion took %s to abort a %s idle timeout", elapsed, timeout)
}
}
// TestChatCompletion_Timeout_ClassifiedAsProviderFailure_DistinctFromClientDisconnect
// proves the non-negotiable classification requirement: a watchdog-fired
// timeout must wrap errProviderTimeout and name the provider, and must
// never be confusable with a genuine client disconnect (context.Canceled)
// — and vice versa. handleAdapterErrorEnvelope (routes_unified.go) relies
// on exactly this mutual exclusivity to decide whether to log-only (a
// client that walked away) or report a provider failure with a 502.
//
// MUTATION VERIFIED: changing watchdogBody.Read to return the raw
// underlying error unchanged (dropping the errProviderTimeout wrap
// entirely) made the first subtest fail its errors.Is(err,
// errProviderTimeout) assertion, confirming the test actually checks the
// wrap rather than merely "an error occurred". Reverted before
// committing.
func TestChatCompletion_Timeout_ClassifiedAsProviderFailure_DistinctFromClientDisconnect(t *testing.T) {
t.Run("watchdog timeout wraps errProviderTimeout, names the provider, never context.Canceled", func(t *testing.T) {
const timeout = 60 * time.Millisecond
srv, release := newStallingServer(func(w http.ResponseWriter) {
w.Header().Set("Content-Type", "text/event-stream")
w.WriteHeader(http.StatusOK)
w.(http.Flusher).Flush()
})
defer release()
a := newTimeoutTestAdapter(srv.URL, timeout)
rec := httptest.NewRecorder()
_, err := a.chatCompletion(context.Background(), rec, map[string]any{"model": "gpt-5", "messages": []any{}, "stream": true})
if err == nil {
t.Fatal("want an error")
}
if !errors.Is(err, errProviderTimeout) {
t.Errorf("err = %v, want errors.Is(err, errProviderTimeout)", err)
}
if errors.Is(err, context.Canceled) {
t.Errorf("err = %v, must NOT satisfy errors.Is(err, context.Canceled) — a watchdog firing is a provider failure, not a client disconnect", err)
}
if !strings.Contains(err.Error(), `"p1"`) {
t.Errorf("err = %v, want it to name the provider (%q)", err, "p1")
}
})
t.Run("client cancellation before headers wraps context.Canceled, never errProviderTimeout", func(t *testing.T) {
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
time.Sleep(200 * time.Millisecond)
w.WriteHeader(http.StatusOK)
}))
defer srv.Close()
a := newTimeoutTestAdapter(srv.URL, 5*time.Second) // generous: must never fire
ctx, cancel := context.WithCancel(context.Background())
go func() {
time.Sleep(30 * time.Millisecond)
cancel()
}()
rec := httptest.NewRecorder()
_, err := a.chatCompletion(ctx, rec, map[string]any{"model": "gpt-5", "messages": []any{}})
if err == nil {
t.Fatal("want an error")
}
if !errors.Is(err, context.Canceled) {
t.Errorf("err = %v, want errors.Is(err, context.Canceled)", err)
}
if errors.Is(err, errProviderTimeout) {
t.Errorf("err = %v, must NOT satisfy errors.Is(err, errProviderTimeout) — this is a client disconnect, not a provider failure", err)
}
})
t.Run("client cancellation mid-stream (after headers) wraps context.Canceled, never errProviderTimeout", func(t *testing.T) {
srv, release := newStallingServer(func(w http.ResponseWriter) {
w.Header().Set("Content-Type", "text/event-stream")
w.WriteHeader(http.StatusOK)
fl := w.(http.Flusher)
_, _ = fmt.Fprint(w, "data: {\"choices\":[{\"delta\":{\"content\":\"hi\"}}]}\n\n")
fl.Flush()
})
defer release()
a := newTimeoutTestAdapter(srv.URL, 5*time.Second) // generous: must never fire
ctx, cancel := context.WithCancel(context.Background())
rec := httptest.NewRecorder()
go func() {
time.Sleep(30 * time.Millisecond)
cancel()
}()
_, err := a.chatCompletion(ctx, rec, map[string]any{"model": "gpt-5", "messages": []any{}, "stream": true})
if err == nil {
t.Fatal("want an error")
}
if !errors.Is(err, context.Canceled) {
t.Errorf("err = %v, want errors.Is(err, context.Canceled)", err)
}
if errors.Is(err, errProviderTimeout) {
t.Errorf("err = %v, must NOT satisfy errors.Is(err, errProviderTimeout) — watchdogBody must pass a real client cancel through unchanged", err)
}
})
}
// TestResolveRequestTimeout is table-driven over resolveRequestTimeout's
// full contract: empty means the default; a valid duration parses; a
// malformed string, a zero, and a negative duration are all construction
// errors — the zero/negative case never means "no timeout" (that is the
// bug this feature fixes).
//
// MUTATION VERIFIED: removing the `if d <= 0 { return error }` branch
// (timeout.go) made the "zero" and "negative" cases return (0, nil)
// instead of an error, failing this test's assertions for both. Reverted
// before committing.
func TestResolveRequestTimeout(t *testing.T) {
cases := []struct {
name string
raw string
want time.Duration
wantErr bool
}{
{name: "empty means default", raw: "", want: defaultRequestTimeout},
{name: "valid minutes", raw: "5m", want: 5 * time.Minute},
{name: "valid seconds", raw: "90s", want: 90 * time.Second},
{name: "malformed string is an error", raw: "not-a-duration", wantErr: true},
{name: "zero is an error, not unlimited", raw: "0s", wantErr: true},
{name: "negative is an error", raw: "-5s", wantErr: true},
}
for _, c := range cases {
t.Run(c.name, func(t *testing.T) {
got, err := resolveRequestTimeout(c.raw, "requestTimeout")
if c.wantErr {
if err == nil {
t.Fatalf("resolveRequestTimeout(%q) = (%v, nil), want an error", c.raw, got)
}
return
}
if err != nil {
t.Fatalf("resolveRequestTimeout(%q): %v", c.raw, err)
}
if got != c.want {
t.Errorf("resolveRequestTimeout(%q) = %v, want %v", c.raw, got, c.want)
}
})
}
}
// TestResolveProviderTimeout_OverrideBeatsGlobal proves the house rule
// ("explicit override always wins") for this feature specifically: a
// non-empty per-provider raw wins regardless of what globalTimeout is,
// and an empty one inherits globalTimeout exactly.
func TestResolveProviderTimeout_OverrideBeatsGlobal(t *testing.T) {
const global = 2 * time.Minute
got, err := resolveProviderTimeout("p1", "30s", global)
if err != nil {
t.Fatalf("resolveProviderTimeout: %v", err)
}
if got != 30*time.Second {
t.Errorf("provider override: got %v, want 30s (must win over the 2m global)", got)
}
got, err = resolveProviderTimeout("p1", "", global)
if err != nil {
t.Fatalf("resolveProviderTimeout: %v", err)
}
if got != global {
t.Errorf("empty provider override: got %v, want %v (inherit the global)", got, global)
}
if _, err := resolveProviderTimeout("p1", "0s", global); err == nil {
t.Error("resolveProviderTimeout(\"p1\", \"0s\", ...): want an error, got nil")
}
}
// TestBuildAdapters_RequestTimeout covers the full config-resolution
// matrix end to end through buildAdapters, matching the sibling
// TestBuildAdapters_WiresRetryPolicyFromConfig's own style (retry_test.go):
// unset everywhere resolves to five minutes; a global default applies
// when a provider sets no override; an explicit provider override wins
// over the global; and a malformed value at either level fails
// construction — including a malformed GLOBAL value that happens to be
// shadowed by every provider's own valid override, proving the global is
// still validated unconditionally rather than lying dormant.
func TestBuildAdapters_RequestTimeout(t *testing.T) {
t.Run("unset everywhere yields the five-minute default", func(t *testing.T) {
cfg := CreateConfig()
cfg.Providers = map[string]*ProviderConfig{"p1": {Type: "openai", APIKey: "k"}}
adapters, err := buildAdapters(cfg)
if err != nil {
t.Fatalf("buildAdapters: %v", err)
}
oa := adapters["p1"].(*openaiAdapter)
if oa.timeout != defaultRequestTimeout {
t.Errorf("oa.timeout = %v, want %v", oa.timeout, defaultRequestTimeout)
}
})
t.Run("global default applies when the provider sets no override", func(t *testing.T) {
cfg := CreateConfig()
cfg.RequestTimeout = "2m"
cfg.Providers = map[string]*ProviderConfig{"p1": {Type: "openai", APIKey: "k"}}
adapters, err := buildAdapters(cfg)
if err != nil {
t.Fatalf("buildAdapters: %v", err)
}
oa := adapters["p1"].(*openaiAdapter)
if oa.timeout != 2*time.Minute {
t.Errorf("oa.timeout = %v, want 2m", oa.timeout)
}
})
t.Run("provider override wins over the global", func(t *testing.T) {
cfg := CreateConfig()
cfg.RequestTimeout = "2m"
cfg.Providers = map[string]*ProviderConfig{"p1": {Type: "openai", APIKey: "k", RequestTimeout: "30s"}}
adapters, err := buildAdapters(cfg)
if err != nil {
t.Fatalf("buildAdapters: %v", err)
}
oa := adapters["p1"].(*openaiAdapter)
if oa.timeout != 30*time.Second {
t.Errorf("oa.timeout = %v, want 30s (override must win over the 2m global)", oa.timeout)
}
})
t.Run("malformed global fails construction even when every provider overrides it", func(t *testing.T) {
cfg := CreateConfig()
cfg.RequestTimeout = "not-a-duration"
cfg.Providers = map[string]*ProviderConfig{"p1": {Type: "openai", APIKey: "k", RequestTimeout: "30s"}}
if _, err := buildAdapters(cfg); err == nil {
t.Fatal("buildAdapters: want an error for a malformed global requestTimeout, got nil (it must be validated even when shadowed)")
}
})
t.Run("zero global is a construction error", func(t *testing.T) {
cfg := CreateConfig()
cfg.RequestTimeout = "0s"
cfg.Providers = map[string]*ProviderConfig{"p1": {Type: "openai", APIKey: "k"}}
if _, err := buildAdapters(cfg); err == nil {
t.Fatal("buildAdapters: want an error for requestTimeout: \"0s\", got nil")
}
})
t.Run("negative provider override is a construction error", func(t *testing.T) {
cfg := CreateConfig()
cfg.Providers = map[string]*ProviderConfig{"p1": {Type: "openai", APIKey: "k", RequestTimeout: "-1s"}}
if _, err := buildAdapters(cfg); err == nil {
t.Fatal("buildAdapters: want an error for a provider requestTimeout of \"-1s\", got nil")
}
})
}
// TestNewAdapterHTTPClient_SetsResponseHeaderTimeout proves
// newAdapterHTTPClient wires its timeout argument onto the returned
// client's Transport.ResponseHeaderTimeout — the mechanism
// TestChatCompletion_NeverSendsHeaders_AbortedAtTimeout exercises
// end to end; this is the narrower, direct check.
func TestNewAdapterHTTPClient_SetsResponseHeaderTimeout(t *testing.T) {
const want = 42 * time.Second
c := newAdapterHTTPClient(want)
tr, ok := c.Transport.(*http.Transport)
if !ok {
t.Fatalf("client.Transport is %T, want *http.Transport", c.Transport)
}
if tr.ResponseHeaderTimeout != want {
t.Errorf("Transport.ResponseHeaderTimeout = %v, want %v", tr.ResponseHeaderTimeout, want)
}
if c.Timeout != 0 {
t.Errorf("Client.Timeout = %v, want 0 (unset) — it would bound the whole request including body read, truncating a healthy long stream", c.Timeout)
}
}
// fakeReadCloser is a minimal io.ReadCloser test double for watchdogBody's
// own unit tests below: Read always blocks until unblock is closed (or
// forever if it never is), simulating a stalled upstream body without a
// real network connection.
type fakeReadCloser struct {
unblock chan struct{}
closed chan struct{}
}
func newFakeReadCloser() *fakeReadCloser {
return &fakeReadCloser{unblock: make(chan struct{}), closed: make(chan struct{})}
}
func (f *fakeReadCloser) Read(p []byte) (int, error) {
<-f.unblock
return 0, io.EOF
}
func (f *fakeReadCloser) Close() error {
select {
case <-f.closed:
default:
close(f.closed)
}
return nil
}
// TestWatchdogBody_Close_StopsTimer_FireNeverRuns proves Close's "no timer
// or goroutine leak" contract directly and deterministically (not via
// runtime.NumGoroutine, which only observes a timer once it has already
// fired): closing a watchdogBody before its timeout elapses must stop the
// timer outright, so fire (and therefore cancel) never runs even once
// timeout has since passed.
//
// MUTATION VERIFIED: removing `wb.timer.Stop()` from watchdogBody.Close
// (timeout.go) made this test fail — timedOut read back as 1 after the
// sleep, since the un-stopped timer still fired on schedule. Reverted
// before committing.
//
// wb.timedOut is read via atomic.LoadInt32 below, not a bare field read
// (coordinator adversarial review, 2026-08-23, finding F9): every other
// access in timeout.go itself is atomic, so a bare read here was a
// latent -race report waiting on a scheduling difference, even though
// Close's own Stop happens-before makes this specific read safe in
// practice.
func TestWatchdogBody_Close_StopsTimer_FireNeverRuns(t *testing.T) {
const timeout = 30 * time.Millisecond
frc := newFakeReadCloser()
ctx, cancel := context.WithCancel(context.Background())
wb := newWatchdogBody(frc, cancel, timeout, "p1", nil)
if err := wb.Close(); err != nil {
t.Fatalf("Close: %v", err)
}
time.Sleep(3 * timeout) // well past when an un-stopped timer would fire
if got := atomic.LoadInt32(&wb.timedOut); got != 0 {
t.Errorf("wb.timedOut = %d after Close, want 0 — the timer must not fire once stopped", got)
}
select {
case <-ctx.Done():
// Close itself calls cancel unconditionally (its own doc comment),
// so ctx.Done() firing here is expected and does not indicate fire
// ran — only wb.timedOut distinguishes that.
default:
t.Error("ctx.Done() not closed after wb.Close() — Close must call cancel unconditionally")
}
}
// TestWatchdogBody_NoBlockingGoroutineLeak_AcrossManyRequests drives many
// successful (fast) and many timed-out requests through a real adapter
// and asserts the live goroutine count returns to its baseline afterward
// — proving neither path leaves a goroutine permanently blocked. The
// task's own concern ("do not leak a goroutine or a timer per request")
// is specifically about a goroutine that never exits (the classic
// `go func(){ select{ case <-time.After(d): ...; case <-done: } }()`
// leak, when done is never closed on the success path); watchdogBody's
// actual design (time.AfterFunc + Stop) never spawns a goroutine at all
// unless the watchdog fires, and that goroutine runs fire() to
// completion and exits immediately either way.
//
// MUTATION VERIFIED: adding `go func() { <-make(chan struct{}) }()` (a
// permanently blocked goroutine) to the top of watchdogBody.fire
// (timeout.go) made this test fail — the goroutine count after the
// timed-out-request loop stayed elevated by roughly the loop's iteration
// count and never settled within the poll window, confirming the test
// catches a real per-request leak. Reverted before committing.
func TestWatchdogBody_NoBlockingGoroutineLeak_AcrossManyRequests(t *testing.T) {
const (
iterations = 40
fastTimeout = 2 * time.Second // generous: must never fire for the success path
hangTimeout = 20 * time.Millisecond
)
fastSrv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(http.StatusOK)
_, _ = w.Write([]byte(`{"id":"c1","choices":[{"message":{"role":"assistant","content":"hi"}}],"usage":{"prompt_tokens":1,"completion_tokens":1}}`))
}))
defer fastSrv.Close()
// prewrite sends headers before stalling: a "never sends headers" hang
// is caught by Transport.ResponseHeaderTimeout, in a *http.Client.Do
// error, and never even constructs a watchdogBody — this test is
// specifically about the WATCHDOG's own goroutine, so its hang must
// stall AFTER headers, the same shape as the mid-body-stall tests
// above, to actually invoke fire() on every one of the iterations
// below.
hangSrv, release := newStallingServer(func(w http.ResponseWriter) {
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(http.StatusOK)
w.(http.Flusher).Flush()
})
runtime.GC()
baseline := runtime.NumGoroutine()
fast := newTimeoutTestAdapter(fastSrv.URL, fastTimeout)
for i := 0; i < iterations; i++ {
rec := httptest.NewRecorder()
if _, err := fast.chatCompletion(context.Background(), rec, map[string]any{"model": "gpt-5", "messages": []any{}}); err != nil {
t.Fatalf("fast request %d: %v", i, err)
}
}
hang := newTimeoutTestAdapter(hangSrv.URL, hangTimeout)
for i := 0; i < iterations; i++ {
rec := httptest.NewRecorder()
if _, err := hang.chatCompletion(context.Background(), rec, map[string]any{"model": "gpt-5", "messages": []any{}}); err == nil {
t.Fatalf("hung request %d: want a timeout error, got nil", i)
}
}
// release BEFORE measuring: each of the iterations above left its own
// server-side handler goroutine parked on newStallingServer's <-done
// (this test's own fixture design, not this feature's watchdog code)
// — releasing them here, before the goroutine count below, keeps the
// measurement about the watchdog's own goroutines settling, not about
// this test's still-blocked handlers swamping the signal.
release()
deadline := time.Now().Add(3 * time.Second)
var final int
const slack = 5 // unrelated Go runtime/http.Transport bookkeeping goroutines, not this feature's concern
for {
runtime.GC()
final = runtime.NumGoroutine()
if final <= baseline+slack || time.Now().After(deadline) {
break
}
time.Sleep(20 * time.Millisecond)
}
if final > baseline+slack {
t.Errorf("goroutines after %d fast + %d timed-out requests = %d, baseline = %d (slack %d) — possible leak", iterations, iterations, final, baseline, slack)
}
}
// TestHandlePassthrough_StalledUpstream_AbortedAtTimeout proves the
// passthrough route (routes_passthrough.go's proxyUpstream) shares this
// feature's idle-progress watchdog, not just each provider adapter's own
// upstreamBytes call path — the task brief's explicit "also cover the
// passthrough route" requirement. adapter.httpClient() and
// adapter.requestTimeout() are the SAME values chatCompletion uses, so a
// provider's requestTimeout config applies identically through native
// passthrough.
func TestHandlePassthrough_StalledUpstream_AbortedAtTimeout(t *testing.T) {
srv, release := newStallingServer(nil)
defer release()
cfg := CreateConfig()
cfg.Providers = map[string]*ProviderConfig{
"openai": {Type: "openai", BaseURL: srv.URL, APIKey: "sk-up", RequestTimeout: "60ms"},
}
cfg.Groups = map[string]*GroupConfig{"default": {}}
cfg.Users = &UsersConfig{Inline: []*UserConfig{
{Name: "alice", Group: "default", APIKey: "sk-alice", Limits: &LimitsConfig{}},
}}
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)
gw.limiter.spawn = func(f func()) { f() }
req := httptest.NewRequest(http.MethodPost, "/openai/v1/native-endpoint", strings.NewReader(`{}`))
req.Header.Set("Authorization", "Bearer sk-alice")
rec := httptest.NewRecorder()
start := time.Now()
h.ServeHTTP(rec, req)
elapsed := time.Since(start)
if rec.Code != http.StatusBadGateway {
t.Errorf("status = %d, want 502, body=%s", rec.Code, rec.Body.String())
}
if elapsed > 3*time.Second {
t.Errorf("passthrough took %s to abort a 60ms provider requestTimeout", elapsed)
}
}
// TestNewGateway_TargetTimeout_ResolvedFromConfig proves the MCP/A2A
// target proxy's own client (g.targetClient, llmgateway.go) picks up
// Config.RequestTimeout the same way a provider adapter does — g.
// targetTimeout is what proxyUpstream's handleTargetProxy call site
// (mcp_a2a.go) passes as the idle-watchdog bound, and targetClient's own
// Transport.ResponseHeaderTimeout must match it too.
func TestNewGateway_TargetTimeout_ResolvedFromConfig(t *testing.T) {
cfg := CreateConfig()
cfg.RequestTimeout = "90s"
cfg.Providers = map[string]*ProviderConfig{"p1": {Type: "openai", APIKey: "k"}}
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)
if gw.targetTimeout != 90*time.Second {
t.Errorf("gw.targetTimeout = %v, want 90s", gw.targetTimeout)
}
tr, ok := gw.targetClient.Transport.(*http.Transport)
if !ok {
t.Fatalf("targetClient.Transport is %T, want *http.Transport", gw.targetClient.Transport)
}
if tr.ResponseHeaderTimeout != 90*time.Second {
t.Errorf("targetClient Transport.ResponseHeaderTimeout = %v, want 90s", tr.ResponseHeaderTimeout)
}
}
// TestChatCompletion_Stream_StallsMidBody_RecordsProviderFailure proves
// finding F4 (coordinator adversarial review, 2026-08-23): a mid-body
// watchdog timeout must reach provider-health accounting through the
// SAME attemptRecorder mechanism retryPolicy.do already uses at
// header-arrival time. Before this fix, limiter.recordProviderAttempt was
// only ever invoked once per request — at header time — so a provider
// that reliably stalls mid-body looked 100% successful to any consumer
// of those counters (the discovery breaker, cross-provider failover)
// forever: attempts=1 failures=0, identical to a real success.
//
// MUTATION VERIFIED: removing the `if wb.onTimeout != nil { ... }` block
// from watchdogBody.fire (timeout.go) made this test fail — the spy
// recorder was invoked exactly once (the header-time success), never a
// second time recording the failure. Reverted before committing.
func TestChatCompletion_Stream_StallsMidBody_RecordsProviderFailure(t *testing.T) {
const timeout = 60 * time.Millisecond
srv, release := newStallingServer(func(w http.ResponseWriter) {
w.Header().Set("Content-Type", "text/event-stream")
w.WriteHeader(http.StatusOK)
w.(http.Flusher).Flush()
})
defer release()
a := newTimeoutTestAdapter(srv.URL, timeout)
type recorded struct {
resp *http.Response
err error
}
var mu sync.Mutex
var calls []recorded
rec := attemptRecorder(func(resp *http.Response, err error) {
mu.Lock()
calls = append(calls, recorded{resp, err})
mu.Unlock()
})
ctx := withAttemptRecorder(context.Background(), rec)
rw := httptest.NewRecorder()
if _, err := a.chatCompletion(ctx, rw, map[string]any{"model": "gpt-5", "messages": []any{}, "stream": true}); err == nil {
t.Fatal("chatCompletion: want an error for a stream that stalls mid-body, got nil")
}
mu.Lock()
defer mu.Unlock()
if len(calls) != 2 {
t.Fatalf("attemptRecorder invoked %d times, want 2 (header-time success, then watchdog-fire failure): %+v", len(calls), calls)
}
if calls[0].err != nil {
t.Errorf("first call err = %v, want nil (the header-arrival attempt succeeded)", calls[0].err)
}
if calls[1].resp != nil {
t.Errorf("second call resp = %v, want nil", calls[1].resp)
}
if !errors.Is(calls[1].err, errProviderTimeout) {
t.Errorf("second call err = %v, want errors.Is(err, errProviderTimeout)", calls[1].err)
}
}
// zeroByteReadCloser always returns (0, nil) — a legal io.Reader response
// that delivers no bytes without signaling an error or EOF — until
// Close, which switches it to (0, io.EOF). Used by
// TestWatchdogBody_ZeroByteRead_DoesNotResetTimer (finding F6) to prove a
// Read that "succeeds" but delivers nothing does not keep the watchdog
// alive.
type zeroByteReadCloser struct {
closed chan struct{}
}
func newZeroByteReadCloser() *zeroByteReadCloser {
return &zeroByteReadCloser{closed: make(chan struct{})}
}
func (z *zeroByteReadCloser) Read([]byte) (int, error) {
select {
case <-z.closed:
return 0, io.EOF
default:
return 0, nil
}
}
func (z *zeroByteReadCloser) Close() error {
select {
case <-z.closed:
default:
close(z.closed)
}
return nil
}
// TestWatchdogBody_ZeroByteRead_DoesNotResetTimer proves finding F6
// (coordinator adversarial review, 2026-08-23): watchdogBody.Read
// resetting on every err == nil read, regardless of n, let a reader that
// keeps returning (0, nil) hold the watchdog open forever — the
// coordinator's own probe measured this surviving past 20x the
// configured timeout across 1044 reads. Not reachable from net/http
// against a non-empty buffer today, but exactly the naive-watchdog hole
// this feature exists to close, so it is fixed and pinned regardless.
//
// This checks wb.timedOut directly rather than expecting wb.Read to
// surface an error: zeroByteReadCloser never inspects the watchdog's
// canceled context (unlike a real net.Conn, whose blocked Read a
// cancellation actually unblocks with an error), so it would return
// (0, nil) forever either way — the property under test is whether the
// watchdog's internal deadline advances, not what Read eventually
// returns.
//
// MUTATION VERIFIED: reverting watchdogBody.Read's `if n > 0 { ... }`
// guard (timeout.go) to an unconditional `wb.timer.Reset(wb.timeout)`
// made this test fail — timedOut stayed 0 for the full 10-timeout poll
// window, since every (0, nil) read kept re-arming the timer. Reverted
// before committing.
func TestWatchdogBody_ZeroByteRead_DoesNotResetTimer(t *testing.T) {
const timeout = 20 * time.Millisecond
zrc := newZeroByteReadCloser()
defer zrc.Close()
_, cancel := context.WithCancel(context.Background())
wb := newWatchdogBody(zrc, cancel, timeout, "p1", nil)
defer wb.Close()
buf := make([]byte, 16)
deadline := time.Now().Add(10 * timeout)
for time.Now().Before(deadline) && atomic.LoadInt32(&wb.timedOut) == 0 {
n, _ := wb.Read(buf) // always (0, nil) from this fixture, until Close
if n != 0 {
t.Fatalf("Read returned n=%d, want 0 (fixture bug)", n)
}
time.Sleep(time.Millisecond)
}
if got := atomic.LoadInt32(&wb.timedOut); got != 1 {
t.Fatalf("wb.timedOut = %d after %s of continuous (0, nil) reads (timeout=%s), want 1 — a Read returning (0, nil) must not keep resetting the watchdog", got, 10*timeout, timeout)
}
}
// TestWatchdogBody_TimeoutError_UsesCallerLabel_NotHardcodedProvider
// proves finding F9 (coordinator adversarial review, 2026-08-23):
// proxyUpstream (routes_passthrough.go) previously passed logPrefix as a
// bare "providerName" argument that watchdogBody wrapped in a hardcoded
// `provider %q` — producing a timeout error reading `provider
// "mcp target (name foo)"` for an MCP/A2A target proxy request, which is
// not a provider at all. watchdogBody now takes an already-formatted
// label from the caller and uses it verbatim.
//
// MUTATION VERIFIED: changing watchdogBody.timeoutError's format string
// back to `"%w: provider %q: no upstream progress for %s"` (timeout.go)
// made this test fail both assertions — the caller's label no longer
// appeared verbatim, and the hardcoded "provider \"mcp target" prefix
// reappeared. Reverted before committing.
func TestWatchdogBody_TimeoutError_UsesCallerLabel_NotHardcodedProvider(t *testing.T) {
const timeout = 20 * time.Millisecond
const label = `mcp target (name "foo")`
frc := newFakeReadCloser()
_, cancel := context.WithCancel(context.Background())
wb := newWatchdogBody(frc, cancel, timeout, label, nil)
defer wb.Close()
time.Sleep(3 * timeout) // let the watchdog fire
close(frc.unblock) // simulate the underlying connection erroring out once canceled
_, err := wb.Read(make([]byte, 16))
if err == nil {
t.Fatal("Read: want an error once the watchdog has fired, got nil")
}
if !strings.Contains(err.Error(), label) {
t.Errorf("err = %v, want it to contain the caller's own label (%q) verbatim", err, label)
}
if strings.Contains(err.Error(), `provider "mcp target`) {
t.Errorf(`err = %v, must not wrap the label in a hardcoded "provider %%q" — an MCP/A2A target is not a provider`, err)
}
}
+187
View File
@@ -106,6 +106,60 @@ const (
yaegiMetaContextTokens = "128000"
)
// timeoutProviderName/timeoutProviderModel/timeoutProviderRequestTimeout/
// timeoutUpstreamSleep/timeoutProbeMaxWait back exerciseRequestTimeout
// (feature: request timeout): a provider whose own ProviderConfig.
// RequestTimeout ("80ms") is far shorter than timeoutUpstream's
// deliberate 600ms silence before it ever writes a byte. Distinct from
// slowProviderName/slowRequestDeadline above (SHOULD-A): that probe
// proves a CLIENT-set context deadline is classified correctly;
// timeoutProviderName proves the GATEWAY'S OWN adapter-level timeout
// (newAdapterHTTPClient's Transport.ResponseHeaderTimeout, providers.go;
// watchdogBody, timeout.go) aborts a hang under a request that sets NO
// deadline of its own — the shape of a real, unbounded production
// client — interpreted, not merely compiled.
const (
timeoutProviderName = "hungtimeout"
// timeoutProviderModel is deliberately NOT "gpt-test"
// (testDataWantModel): a bare model id shared across providers
// resolves to exactly one winner (modelRegistry's own bare-id rule),
// and reusing testDataWantModel here made "hungtimeout" win it
// instead of "openai" — silently hijacking exerciseAttemptAccounting's
// own bare-"gpt-test" request onto this hung provider and failing
// that unrelated probe. A unique id keeps this provider's own model
// space from ever colliding with another probe's.
timeoutProviderModel = "gpt-timeout-test"
timeoutProviderRequestTimeout = "80ms"
timeoutUpstreamSleep = 600 * time.Millisecond
// timeoutProbeMaxWait bounds exerciseRequestTimeout's own assertion:
// generous relative to timeoutProviderRequestTimeout (80ms), but far
// under timeoutUpstreamSleep (600ms) — a pass proves the request was
// aborted BY the timeout feature, not by timeoutUpstream eventually
// answering on its own.
timeoutProbeMaxWait = 3 * time.Second
)
// hungBodyProviderName/hungBodyProviderModel/hungBodyProviderRequestTimeout/
// hungBodyUpstreamSleep back exerciseRequestTimeoutMidBodyStall (finding
// F7, coordinator adversarial review, 2026-08-23): timeoutProviderName
// above never sends headers at all, so it only exercises Transport.
// ResponseHeaderTimeout — watchdogBody.fire (timeout.go) — the
// interpreted method value handed to time.AfterFunc, plus sync/atomic's
// function API and a context.CancelFunc — never runs under that probe.
// hungBodyUpstream sends headers immediately, then goes silent, so the
// body-read path (upstreamBytes' watchdogBody wrap, providers.go) is
// what has to abort it, proving the OTHER half of this feature
// interpreted.
const (
hungBodyProviderName = "hungbody"
// hungBodyProviderModel: same reasoning as timeoutProviderModel's own
// doc comment above — must not collide with testDataWantModel or any
// other probe's model id.
hungBodyProviderModel = "gpt-bodystall-test"
hungBodyProviderRequestTimeout = "80ms"
hungBodyUpstreamSleep = 600 * time.Millisecond
)
// excludedTopLevelDirs lists repo-root directories the GOPATH copy must
// never include: build tooling, integration fixtures, planning docs and
// VCS metadata have nothing to do with the plugin package Yaegi imports.
@@ -260,6 +314,36 @@ func run() error {
}))
defer slowUpstream.Close()
// timeoutUpstream writes nothing at all for timeoutUpstreamSleep,
// well past timeoutProviderName's own 80ms ProviderConfig.
// RequestTimeout (attemptAccountingOverride below) — see the
// timeoutProviderName const block's own doc comment for why this
// probe exists alongside slowUpstream.
timeoutUpstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
time.Sleep(timeoutUpstreamSleep)
w.Header().Set("Content-Type", "application/json")
_, _ = w.Write([]byte(`{"id":"c4","object":"chat.completion","model":"gpt-test","choices":[{"index":0,"message":{"role":"assistant","content":"hi"},"finish_reason":"stop"}],"usage":{"prompt_tokens":1,"completion_tokens":1}}`))
}))
defer timeoutUpstream.Close()
// hungBodyUpstream sends headers immediately (so ResponseHeaderTimeout
// is satisfied and never fires), then goes silent for
// hungBodyUpstreamSleep before ever writing a body byte — well past
// hungBodyProviderName's own 80ms ProviderConfig.RequestTimeout. See
// the hungBodyProviderName const block's own doc comment (finding F7)
// for why this probe exists alongside timeoutUpstream, not instead of
// it.
hungBodyUpstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(http.StatusOK)
if fl, ok := w.(http.Flusher); ok {
fl.Flush()
}
time.Sleep(hungBodyUpstreamSleep)
_, _ = w.Write([]byte(`{"id":"c5","object":"chat.completion","model":"gpt-test","choices":[{"index":0,"message":{"role":"assistant","content":"hi"},"finish_reason":"stop"}],"usage":{"prompt_tokens":1,"completion_tokens":1}}`))
}))
defer hungBodyUpstream.Close()
// mcpProbeUpstream answers federation's outbound tools/call with a
// body deliberately larger than mcpBackendCallResponseMaxBytes, so
// exerciseHandler's POST /mcp probe below drives doBackendJSONRPC's
@@ -396,6 +480,14 @@ func run() error {
`"providers":{"openai":{"type":"openai","baseUrl":"` + upstream.URL + `","apiKey":"sk-up","models":["` + testDataWantModel + `","` + builtinLookupModelID + `"]},` +
`"brk":{"type":"openai","baseUrl":"` + brkUpstream.URL + `","apiKey":"sk-up","discovery":true,"discoveryInterval":"1ms"},"` +
slowProviderName + `":{"type":"openai","baseUrl":"` + slowUpstream.URL + `","apiKey":"sk-up","models":["` + slowProviderModel + `"]},` +
// timeoutProviderName (feature: request timeout) — see its own
// const block doc comment above for why this provider exists
// alongside slowProviderName.
`"` + timeoutProviderName + `":{"type":"openai","baseUrl":"` + timeoutUpstream.URL + `","apiKey":"sk-up","requestTimeout":"` + timeoutProviderRequestTimeout + `","models":["` + timeoutProviderModel + `"]},` +
// hungBodyProviderName (finding F7) — see its own const block doc
// comment above for why this provider exists alongside
// timeoutProviderName.
`"` + hungBodyProviderName + `":{"type":"openai","baseUrl":"` + hungBodyUpstream.URL + `","apiKey":"sk-up","requestTimeout":"` + hungBodyProviderRequestTimeout + `","models":["` + hungBodyProviderModel + `"]},` +
// "anthropic" backs the /v1/messages passthrough probes
// (exerciseMessagesRoute, below): a real anthropic-type provider,
// interpreted end to end through
@@ -896,6 +988,14 @@ func exerciseHandler(handler http.Handler, builtinContextTokens int, failoverBHi
return err
}
if err := exerciseRequestTimeout(handler); err != nil {
return err
}
if err := exerciseRequestTimeoutMidBodyStall(handler); err != nil {
return err
}
if err := exerciseFederatedTooLarge(handler); err != nil {
return err
}
@@ -1239,6 +1339,93 @@ func exerciseFailover(handler http.Handler, bHits *int64) error {
return nil
}
// exerciseRequestTimeout proves the request-timeout feature aborts a
// hung provider under the REAL interpreter, not merely in compiled
// tests: timeoutProviderName's own ProviderConfig.RequestTimeout ("80ms")
// is far shorter than timeoutUpstream's deliberate 600ms silence, and the
// request below carries NO client-side deadline of its own — unlike
// exerciseAttemptAccounting's slowProviderName probe (SHOULD-A), which
// proves a CLIENT-set context deadline classifies correctly, this proves
// the GATEWAY'S OWN adapter-level timeout aborts a hang no caller ever
// bounded, exactly the shape of a real, unbounded production client.
// A pass here means newAdapterHTTPClient's Transport.
// ResponseHeaderTimeout (providers.go) and watchdogBody's construction
// (timeout.go — context.WithCancel, time.AfterFunc, sync/atomic's
// function API) all compile and run correctly interpreted; a failure
// (a status other than 502, or an elapsed time anywhere near
// timeoutUpstreamSleep) would mean this feature works compiled but not
// under Yaegi, the exact class of divergence this harness exists to
// catch.
//
// MUTATION VERIFIED: temporarily removing
// `tr.ResponseHeaderTimeout = timeout` from newAdapterHTTPClient
// (providers.go) made `make yaegi-check` fail here with "status = 200,
// want 502" — timeoutUpstream's 600ms sleep let the request succeed
// instead of aborting at timeoutProviderRequestTimeout (80ms), proving
// this probe genuinely exercises the interpreted mechanism rather than
// passing regardless. Reverted before committing.
func exerciseRequestTimeout(handler http.Handler) error {
body := `{"model":"` + timeoutProviderName + `/` + timeoutProviderModel + `","messages":[{"role":"user","content":"hi"}]}`
req := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader(body))
req.Header.Set("Authorization", "Bearer "+testDataUserAPIKey)
req.Header.Set("Content-Type", "application/json")
rec := httptest.NewRecorder()
start := time.Now()
handler.ServeHTTP(rec, req)
elapsed := time.Since(start)
if elapsed > timeoutProbeMaxWait {
return fmt.Errorf("POST /v1/chat/completions (hung provider %q) took %s, want well under %s — a hung provider must be aborted under the interpreter, not merely in compiled tests: %s", timeoutProviderName, elapsed, timeoutProbeMaxWait, rec.Body.String())
}
if rec.Code != http.StatusBadGateway {
return fmt.Errorf("POST /v1/chat/completions (hung provider %q): status = %d, want 502 (upstream connection error, provider-attributed), body=%s", timeoutProviderName, rec.Code, rec.Body.String())
}
return nil
}
// exerciseRequestTimeoutMidBodyStall proves the OTHER half of the
// request-timeout feature aborts under the REAL interpreter (finding F7,
// coordinator adversarial review, 2026-08-23): exerciseRequestTimeout
// above drives timeoutProviderName, whose upstream never sends headers
// at all, so it only exercises Transport.ResponseHeaderTimeout —
// watchdogBody.fire (timeout.go) never runs under that probe, since the
// request never gets far enough to construct a watchdogBody in the first
// place. hungBodyProviderName's upstream sends headers immediately, then
// stalls, so this drives the idle-progress body watchdog itself:
// time.AfterFunc handed the interpreted method value wb.fire,
// context.WithCancel/CancelFunc, and sync/atomic's function API
// (atomic.StoreInt32/LoadInt32) all have to work correctly interpreted
// for this to abort rather than hang.
//
// MUTATION VERIFIED: temporarily removing the `resp.Body =
// newWatchdogBody(...)` line from upstreamBytes (providers.go) made
// exerciseRequestTimeout (the header-timeout probe) still PASS
// unchanged, while this probe failed with "status = 200, want 502" —
// hungBodyUpstream's 600ms sleep let the request succeed once nothing
// guarded the body read. This confirms the two probes exercise genuinely
// different code paths, and that this one specifically needs
// watchdogBody to pass. Reverted before committing.
func exerciseRequestTimeoutMidBodyStall(handler http.Handler) error {
body := `{"model":"` + hungBodyProviderName + `/` + hungBodyProviderModel + `","messages":[{"role":"user","content":"hi"}]}`
req := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader(body))
req.Header.Set("Authorization", "Bearer "+testDataUserAPIKey)
req.Header.Set("Content-Type", "application/json")
rec := httptest.NewRecorder()
start := time.Now()
handler.ServeHTTP(rec, req)
elapsed := time.Since(start)
if elapsed > timeoutProbeMaxWait {
return fmt.Errorf("POST /v1/chat/completions (mid-body-stall provider %q) took %s, want well under %s — a provider that sends headers then stalls must still be aborted under the interpreter, not merely in compiled tests: %s", hungBodyProviderName, elapsed, timeoutProbeMaxWait, rec.Body.String())
}
if rec.Code != http.StatusBadGateway {
return fmt.Errorf("POST /v1/chat/completions (mid-body-stall provider %q): status = %d, want 502 (upstream connection error, provider-attributed), body=%s", hungBodyProviderName, rec.Code, rec.Body.String())
}
return nil
}
func exerciseAttemptAccounting(handler http.Handler) error {
chatReq := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader(
`{"model":"`+testDataWantModel+`","messages":[{"role":"user","content":"hi"}]}`,