From 503fc339330b38f0b6575669c9426b62b515be10 Mon Sep 17 00:00:00 2001 From: Lukasz Raczylo Date: Sun, 23 Aug 2026 01:36:34 +0100 Subject: [PATCH 1/7] feat(providers): add progress-based upstream request timeout Adapters previously issued upstream requests with no timeout at all. A provider that accepted a connection and then stalled hung the gateway request forever. newAdapterHTTPClient now sets Transport.ResponseHeaderTimeout, and every response body read (providers.go's upstreamBytes, routes_passthrough.go's proxyUpstream) runs through a watchdogBody that resets on every successful read and cancels the request once timeout passes with no progress. The idle-progress design lets a long stream that keeps emitting tokens run indefinitely. Only silence, never total duration, ends a request. Config.RequestTimeout sets the global default (five minutes when unset). ProviderConfig.RequestTimeout overrides it per provider. Both are Go duration strings, validated at construction; a zero or negative value is rejected rather than treated as unlimited, since that would reintroduce the hang this feature fixes. Native passthrough (routes_passthrough.go) and the MCP/A2A target proxy (mcp_a2a.go) share the identical watchdog mechanism through proxyUpstream's new timeout parameter. A watchdog-fired timeout wraps errProviderTimeout and names the provider. It never wraps context.Canceled, so it stays distinguishable from a genuine client disconnect in every existing log classification. --- llmgateway.go | 92 ++++-- mcp_a2a.go | 2 +- provider_anthropic.go | 16 +- provider_gemini.go | 20 +- provider_openai.go | 36 ++- providers.go | 102 ++++++- registry_test.go | 3 +- retry_test.go | 2 +- routes_messages.go | 2 +- routes_passthrough.go | 36 ++- timeout.go | 185 +++++++++++ timeout_test.go | 694 ++++++++++++++++++++++++++++++++++++++++++ 12 files changed, 1130 insertions(+), 60 deletions(-) create mode 100644 timeout.go create mode 100644 timeout_test.go diff --git a/llmgateway.go b/llmgateway.go index ed15ff0..ddca1e2 100644 --- a/llmgateway.go +++ b/llmgateway.go @@ -11,6 +11,7 @@ import ( "os" "runtime" "sync" + "time" telemetry "github.com/lukaszraczylo/oss-telemetry" ) @@ -24,27 +25,7 @@ import ( // original doc comment) — see providerState's own doc comment (registry.go) // for the general pointer-first/scalar-last convention this follows. type Config struct { - Redis *RedisConfig `json:"redis,omitempty"` - // Admin gates the read-only admin dashboard (spec §4, v0.2): nil or - // Admin.Enabled false means the /admin* routes are not registered at - // all — ServeHTTP falls through to its existing 404/passthroughUnknown - // handling for those paths, preserving v0.1 behavior exactly. - Admin *AdminConfig `json:"admin,omitempty"` - Pricing map[string]*ModelPricing `json:"pricing,omitempty"` - MCPServers map[string]*TargetConfig `json:"mcpServers,omitempty"` - Agents map[string]*AgentConfig `json:"agents,omitempty"` - Users *UsersConfig `json:"users,omitempty"` - Groups map[string]*GroupConfig `json:"groups,omitempty"` - // ModelAliases maps an operator-defined alias id to a target model id - // (spec §5, v0.2) — e.g. {"aliased/coding": "anthropic/claude-sonnet-4-5"}. - // An exact alias match wins resolution before any other rule - // (modelRegistry.resolve, registry.go); the target then resolves - // through the normal rules. nil/omitted preserves v0.1 behavior - // exactly: resolve never consults an empty alias map, so no existing - // model id's resolution changes. Validated at construction - // (validateModelAliases, registry.go). - ModelAliases map[string]string `json:"modelAliases,omitempty"` - Providers map[string]*ProviderConfig `json:"providers,omitempty"` + Providers map[string]*ProviderConfig `json:"providers,omitempty"` // ModelMeta declares per-model metadata overrides (feature v0.23): // context window size and per-token cost, keyed by an exact // "provider/model" id or a bare model id (applying wherever that bare @@ -58,7 +39,43 @@ type Config struct { // prior behavior exactly: every model's metadata falls through to // discovery/builtin/absent, same as before this feature existed. // Validated at construction (validateModelMeta, modelmeta.go). - ModelMeta map[string]*ModelMetaConfig `json:"modelMeta,omitempty"` + ModelMeta map[string]*ModelMetaConfig `json:"modelMeta,omitempty"` + Pricing map[string]*ModelPricing `json:"pricing,omitempty"` + MCPServers map[string]*TargetConfig `json:"mcpServers,omitempty"` + Agents map[string]*AgentConfig `json:"agents,omitempty"` + Users *UsersConfig `json:"users,omitempty"` + // Admin gates the read-only admin dashboard (spec §4, v0.2): nil or + // Admin.Enabled false means the /admin* routes are not registered at + // all — ServeHTTP falls through to its existing 404/passthroughUnknown + // handling for those paths, preserving v0.1 behavior exactly. + Admin *AdminConfig `json:"admin,omitempty"` + Groups map[string]*GroupConfig `json:"groups,omitempty"` + Redis *RedisConfig `json:"redis,omitempty"` + // ModelAliases maps an operator-defined alias id to a target model id + // (spec §5, v0.2) — e.g. {"aliased/coding": "anthropic/claude-sonnet-4-5"}. + // An exact alias match wins resolution before any other rule + // (modelRegistry.resolve, registry.go); the target then resolves + // through the normal rules. nil/omitted preserves v0.1 behavior + // exactly: resolve never consults an empty alias map, so no existing + // model id's resolution changes. Validated at construction + // (validateModelAliases, registry.go). + ModelAliases map[string]string `json:"modelAliases,omitempty"` + // 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"` // Breaker configures the discovery circuit breaker (feat/provider- // health): how many consecutive discovery-refresh failures open a // provider's breaker, and how long it then backs off before probing @@ -132,6 +149,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, @@ -526,6 +551,13 @@ type Gateway struct { // contract. bodyAdmission chan struct{} name string + // 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 } // telemetryStartupOnce keeps the anonymous "plugin loaded" ping to one per @@ -685,7 +717,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 { diff --git a/mcp_a2a.go b/mcp_a2a.go index 607eaed..a7168c7 100644 --- a/mcp_a2a.go +++ b/mcp_a2a.go @@ -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 diff --git a/provider_anthropic.go b/provider_anthropic.go index ea403a6..aad4fcb 100644 --- a/provider_anthropic.go +++ b/provider_anthropic.go @@ -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 } diff --git a/provider_gemini.go b/provider_gemini.go index cd51e45..94f4cd9 100644 --- a/provider_gemini.go +++ b/provider_gemini.go @@ -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 } diff --git a/provider_openai.go b/provider_openai.go index ffd6beb..b3f102e 100644 --- a/provider_openai.go +++ b/provider_openai.go @@ -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 } diff --git a/providers.go b/providers.go index 7f28a33..9e87b1d 100644 --- a/providers.go +++ b/providers.go @@ -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,22 @@ 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) { 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 +336,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 +352,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, providerName) return resp, nil } @@ -432,6 +472,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 +506,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 +538,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 +547,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) diff --git a/registry_test.go b/registry_test.go index febf747..08e9272 100644 --- a/registry_test.go +++ b/registry_test.go @@ -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. diff --git a/retry_test.go b/retry_test.go index bbf51b8..8ea4a1e 100644 --- a/retry_test.go +++ b/retry_test.go @@ -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") } diff --git a/routes_messages.go b/routes_messages.go index 3e5afaa..861123e 100644 --- a/routes_messages.go +++ b/routes_messages.go @@ -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 } diff --git a/routes_passthrough.go b/routes_passthrough.go index 6ed4d71..71c248d 100644 --- a/routes_passthrough.go +++ b/routes_passthrough.go @@ -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 @@ -1002,6 +1019,7 @@ func (g *Gateway) proxyUpstream(w http.ResponseWriter, r *http.Request, upstream 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 +1028,7 @@ func (g *Gateway) proxyUpstream(w http.ResponseWriter, r *http.Request, upstream writeOAIError(w, http.StatusBadGateway, "server_error", "upstream connection error") return proxyResult{}, false } + resp.Body = newWatchdogBody(resp.Body, cancel, timeout, logPrefix) defer resp.Body.Close() //nolint:errcheck // read-side close; nothing actionable on failure copyHeadersExcept(w.Header(), resp.Header, hopByHopHeaders, dangerousResponseHeaders) @@ -1031,6 +1050,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 } diff --git a/timeout.go b/timeout.go new file mode 100644 index 0000000..e020b70 --- /dev/null +++ b/timeout.go @@ -0,0 +1,185 @@ +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 — a watchdog firing is always this gateway +// giving up on an upstream that stopped making progress: a provider +// failure to attribute to that provider, never the client's own doing. +// watchdogBody.Read never lets a timeout error wrap context.Canceled, so +// the two stay mutually exclusive in every caller's classification. +var errProviderTimeout = errors.New("llmgateway: provider request timed out") + +// 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 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 providerName, 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. +// +// 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 + providerName 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). +func newWatchdogBody(rc io.ReadCloser, cancel context.CancelFunc, timeout time.Duration, providerName string) *watchdogBody { + wb := &watchdogBody{rc: rc, cancel: cancel, timeout: timeout, providerName: providerName} + 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 provider-attributed timeout error, then +// cancels the request context, unblocking whatever Read is currently +// parked waiting on the network. +func (wb *watchdogBody) fire() { + atomic.StoreInt32(&wb.timedOut, 1) + wb.cancel() +} + +// Read implements io.Reader. A successful read (err == nil) 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. A failed read while timedOut is set (fire already +// ran) is reported as errProviderTimeout, naming providerName and the +// configured timeout, 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, fmt.Errorf("%w: %w: provider %q: no upstream progress for %s", errUpstream, errProviderTimeout, wb.providerName, wb.timeout) + } + return n, err + } + 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() +} diff --git a/timeout_test.go b/timeout_test.go new file mode 100644 index 0000000..e4db7e2 --- /dev/null +++ b/timeout_test.go @@ -0,0 +1,694 @@ +package traefikllmgateway + +import ( + "context" + "errors" + "fmt" + "io" + "net/http" + "net/http/httptest" + "runtime" + "strings" + "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. +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") + + 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 := 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() + + hangSrv, release := newStallingServer(nil) + + 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) + } +} From 057c06132653ee149eb916cd6e1b49b752cd4e08 Mon Sep 17 00:00:00 2001 From: Lukasz Raczylo Date: Sun, 23 Aug 2026 01:41:02 +0100 Subject: [PATCH 2/7] test(providers): exercise fire, not ResponseHeaderTimeout, in the leak test TestWatchdogBody_NoBlockingGoroutineLeak_AcrossManyRequests drove its timed-out iterations against a server that never wrote headers. That hits Transport.ResponseHeaderTimeout inside client.Do and returns before a response ever exists, so watchdogBody.fire never runs and the test could not have caught a leak in fire itself. The mutation-verification pass (manually adding a permanently blocked goroutine to fire, then reverting) confirmed this: the test passed even with that goroutine present. Fixing the stalling server to write headers first, mirroring the mid-body-stall tests above it, makes fire run on every iteration. Re-running the same mutation now fails the test as intended. --- timeout_test.go | 13 ++++++++++++- 1 file changed, 12 insertions(+), 1 deletion(-) diff --git a/timeout_test.go b/timeout_test.go index e4db7e2..c3ffdd3 100644 --- a/timeout_test.go +++ b/timeout_test.go @@ -577,7 +577,18 @@ func TestWatchdogBody_NoBlockingGoroutineLeak_AcrossManyRequests(t *testing.T) { })) defer fastSrv.Close() - hangSrv, release := newStallingServer(nil) + // 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() From 5062017cf1940c20aa178613afbb1cb076ea6c55 Mon Sep 17 00:00:00 2001 From: Lukasz Raczylo Date: Sun, 23 Aug 2026 01:44:54 +0100 Subject: [PATCH 3/7] test(yaegi-check): probe a hung provider aborted under the interpreter Adds exerciseRequestTimeout: a provider (hungtimeout) with an 80ms ProviderConfig.RequestTimeout, backed by an upstream that stays silent for 600ms. The request carries no client-side deadline of its own, so a pass proves the gateway's own adapter-level timeout aborts the hang, not a caller-supplied context. This complements the existing slowProviderName probe, which proves a CLIENT-set deadline classifies correctly. It does not prove the adapter's own timeout mechanism runs correctly interpreted, since it never exercises it without client help. Verified the probe is not a no-op: temporarily removing Transport.ResponseHeaderTimeout from newAdapterHTTPClient made make yaegi-check fail with status 200 instead of 502. Reverted before this commit. timeoutProviderModel uses a distinct id, not testDataWantModel: reusing it made the new provider win bare-id model resolution and silently hijacked an unrelated existing probe. --- tools/yaegi-check/main.go | 98 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 98 insertions(+) diff --git a/tools/yaegi-check/main.go b/tools/yaegi-check/main.go index af663a2..bb7b80a 100644 --- a/tools/yaegi-check/main.go +++ b/tools/yaegi-check/main.go @@ -106,6 +106,39 @@ 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 +) + // 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 +293,18 @@ 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() + // mcpProbeUpstream answers federation's outbound tools/call with a // body deliberately larger than mcpBackendCallResponseMaxBytes, so // exerciseHandler's POST /mcp probe below drives doBackendJSONRPC's @@ -350,6 +395,10 @@ 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 + `"]},` + // "anthropic" backs the /v1/messages passthrough probes // (exerciseMessagesRoute, below): a real anthropic-type provider, // interpreted end to end through @@ -702,6 +751,10 @@ func exerciseHandler(handler http.Handler, builtinContextTokens int) error { return err } + if err := exerciseRequestTimeout(handler); err != nil { + return err + } + if err := exerciseFederatedTooLarge(handler); err != nil { return err } @@ -1013,6 +1066,51 @@ func exerciseFederatedTooLarge(handler http.Handler) 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 +} + func exerciseAttemptAccounting(handler http.Handler) error { chatReq := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader( `{"model":"`+testDataWantModel+`","messages":[{"role":"user","content":"hi"}]}`, From 834238d141ec433cd4e22babf464b04c35d1fd89 Mon Sep 17 00:00:00 2001 From: Lukasz Raczylo Date: Sun, 23 Aug 2026 01:46:49 +0100 Subject: [PATCH 4/7] docs: document the request timeout feature and its default Adds a "Request timeout" section explaining the progress-based design, the two mechanisms, the global/per-provider override rule, and error classification. States plainly that this changes behavior for every deployment, since no timeout existed before. Adds requestTimeout rows to the Config and ProviderConfig reference tables. --- README.md | 45 +++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 45 insertions(+) diff --git a/README.md b/README.md index a2f0871..f29665e 100644 --- a/README.md +++ b/README.md @@ -193,6 +193,7 @@ config accepts them as YAML, which decodes to the same JSON shape. | `users` | `UsersConfig` | — | Inline and/or file-backed API-key holders. With none configured, every request is unauthenticated and gets 401. | | `redis` | `RedisConfig` | — | Distributed limit-counter backend. Omitted means in-process counters only (per-replica, approximate across multiple Traefik instances). | | `retry` | `RetryConfig` | `{}` (disabled) | Same-provider retry for transient upstream failures — see [Retry](#retry). Omitted or `enabled: false` means no retry: every request makes exactly one upstream attempt, byte-identical to a gateway built before this field existed. | +| `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. | @@ -232,6 +233,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` @@ -846,6 +848,49 @@ request. is never double-counted. Request counters still increment once per client request, unchanged. +## Request timeout + +Every upstream request this gateway makes — provider adapter calls, native +passthrough (`/{provider}/...`), and the MCP/A2A target proxy — is bounded +by a progress-based timeout: silence ends a request, never elapsed time. + +**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. +- **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 a provider failure (`502`, `server_error`), naming the + provider, and is never classified as a client disconnect + (`context.Canceled`) — the two stay distinguishable in every log line. + ## Caching An opt-in, Redis-backed response cache for the unified, non-streaming From 51e675acb07668fcdc73704ad4a6c6fb0cad6f89 Mon Sep 17 00:00:00 2001 From: Lukasz Raczylo Date: Sun, 23 Aug 2026 02:40:14 +0100 Subject: [PATCH 5/7] fix(providers): address adversarial review findings F1, F4, F6, F9 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit F1: TestModelRegistry_MaybeRefresh_Throttled_OnePerInterval mutated a plain `now` variable while a background refresh goroutine read it through nowFn, racing under -race in about 14% of full-suite runs once this branch's timer-heavy tests shared the parallel phase. Guards the clock with a mutex instead. registry.go is untouched. F4: a mid-body watchdog timeout never reached provider-health accounting. limiter.recordProviderAttempt only ran at header-arrival time, so a provider that reliably stalled mid-body reported attempts=1 failures=0 forever. watchdogBody.fire now invokes the same attemptRecorder a second time, as a failure, when it fires. Reuses the existing recorder rather than adding new plumbing; this does mean two attempts are recorded for one logical request (one success at header time, one failure at watchdog-fire time), not one attempt marked failed outright. F6: watchdogBody.Read reset the timer on any err == nil read, regardless of n. A reader returning (0, nil) forever would have held the watchdog open indefinitely. Reset now only fires when n > 0. F9 (three items): - proxyUpstream passed logPrefix as a bare providerName, which newWatchdogBody wrapped in a hardcoded `provider %q` — labeling an MCP/A2A target as a provider. watchdogBody now takes an already-formatted label from the caller instead. - errUpstream was wrapped twice: once in watchdogBody's own error, once again by every caller that already wraps it. watchdogBody now wraps only errProviderTimeout; callers' existing wraps still make errors.Is(err, errUpstream) hold. - timeout_test.go read wb.timedOut directly in one place while every other access is atomic; switched to atomic.LoadInt32. Also rewords errProviderTimeout's message from "provider request timed out" to "upstream progress stalled" (finding F3, documented not fixed per coordinator ruling): the watchdog measures time since the last successful upstream read, which can go quiet because of a slow client write between reads, not only a slow provider. The new wording no longer asserts which side caused it. New tests: TestChatCompletion_Stream_StallsMidBody_RecordsProviderFailure, TestWatchdogBody_ZeroByteRead_DoesNotResetTimer, TestWatchdogBody_TimeoutError_UsesCallerLabel_NotHardcodedProvider. Each mutation-verified per its own doc comment. --- providers.go | 13 ++- registry_test.go | 24 +++++- routes_passthrough.go | 15 +++- timeout.go | 138 +++++++++++++++++++++++-------- timeout_test.go | 185 +++++++++++++++++++++++++++++++++++++++++- 5 files changed, 332 insertions(+), 43 deletions(-) diff --git a/providers.go b/providers.go index 9e87b1d..e440abd 100644 --- a/providers.go +++ b/providers.go @@ -310,6 +310,17 @@ func upstreamRawBytes(ctx context.Context, client *http.Client, method, url stri // "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, 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 { @@ -355,7 +366,7 @@ func upstreamBytes(ctx context.Context, client *http.Client, method, url string, cancel() return nil, fmt.Errorf("%w: %w", errUpstream, err) } - resp.Body = newWatchdogBody(resp.Body, cancel, timeout, providerName) + resp.Body = newWatchdogBody(resp.Body, cancel, timeout, fmt.Sprintf("provider %q", providerName), rec) return resp, nil } diff --git a/registry_test.go b/registry_test.go index 08e9272..e92ff58 100644 --- a/registry_test.go +++ b/registry_test.go @@ -505,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 }) @@ -518,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 }) } diff --git a/routes_passthrough.go b/routes_passthrough.go index 71c248d..d0f33dd 100644 --- a/routes_passthrough.go +++ b/routes_passthrough.go @@ -1014,8 +1014,13 @@ 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 { @@ -1028,7 +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 } - resp.Body = newWatchdogBody(resp.Body, cancel, timeout, logPrefix) + // 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) diff --git a/timeout.go b/timeout.go index e020b70..b7a5913 100644 --- a/timeout.go +++ b/timeout.go @@ -30,12 +30,22 @@ const defaultRequestTimeout = 5 * time.Minute // 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 — a watchdog firing is always this gateway -// giving up on an upstream that stopped making progress: a provider -// failure to attribute to that provider, never the client's own doing. -// watchdogBody.Read never lets a timeout error wrap context.Canceled, so -// the two stay mutually exclusive in every caller's classification. -var errProviderTimeout = errors.New("llmgateway: provider request timed out") +// 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) @@ -94,9 +104,9 @@ func resolveProviderTimeout(providerName, providerRaw string, globalTimeout time // 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 read, so a stream (or a large -// non-streaming body) that keeps making progress runs indefinitely; only -// SILENCE, never total duration, ends it. +// 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 @@ -105,25 +115,47 @@ func resolveProviderTimeout(providerName, providerRaw string, globalTimeout time // 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 providerName, 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. +// 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 - providerName string - timeout time.Duration - timedOut int32 + 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 @@ -132,8 +164,19 @@ type watchdogBody struct { // 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). -func newWatchdogBody(rc io.ReadCloser, cancel context.CancelFunc, timeout time.Duration, providerName string) *watchdogBody { - wb := &watchdogBody{rc: rc, cancel: cancel, timeout: timeout, providerName: providerName} +// +// 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 } @@ -142,32 +185,59 @@ func newWatchdogBody(rc io.ReadCloser, cancel context.CancelFunc, timeout time.D // 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 provider-attributed timeout error, then -// cancels the request context, unblocking whatever Read is currently -// parked waiting on the network. +// 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()) + } } -// Read implements io.Reader. A successful read (err == nil) 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. A failed read while timedOut is set (fire already -// ran) is reported as errProviderTimeout, naming providerName and the -// configured timeout, 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 +// 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, fmt.Errorf("%w: %w: provider %q: no upstream progress for %s", errUpstream, errProviderTimeout, wb.providerName, wb.timeout) + return n, wb.timeoutError() } return n, err } - wb.timer.Reset(wb.timeout) + if n > 0 { + wb.timer.Reset(wb.timeout) + } return n, nil } diff --git a/timeout_test.go b/timeout_test.go index c3ffdd3..cbce1eb 100644 --- a/timeout_test.go +++ b/timeout_test.go @@ -9,6 +9,8 @@ import ( "net/http/httptest" "runtime" "strings" + "sync" + "sync/atomic" "testing" "time" ) @@ -521,18 +523,25 @@ func (f *fakeReadCloser) Close() error { // (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") + 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 := wb.timedOut; got != 0 { + 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 { @@ -703,3 +712,175 @@ func TestNewGateway_TargetTimeout_ResolvedFromConfig(t *testing.T) { 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) + } +} From c811de3dfb2a38f02eb0083406d001a470fac4fe Mon Sep 17 00:00:00 2001 From: Lukasz Raczylo Date: Sun, 23 Aug 2026 02:53:53 +0100 Subject: [PATCH 6/7] test(yaegi-check): probe the mid-body-stall half of the watchdog too MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit exerciseRequestTimeout only drove a provider that never sends headers, so it exercised Transport.ResponseHeaderTimeout alone. watchdogBody.fire — the interpreted method value handed to time.AfterFunc, plus context.CancelFunc and sync/atomic's function API — never ran under that probe, since the request never got far enough to construct a watchdogBody at all (finding F7, coordinator adversarial review, 2026-08-23). Adds hungbody, a provider whose upstream sends headers immediately and then stalls, and exerciseRequestTimeoutMidBodyStall to drive it. Verified the two probes exercise different code paths: removing watchdogBody's construction from upstreamBytes left the existing header-timeout probe passing unchanged, while the new probe failed with status 200 instead of 502. Reverted before this commit. --- tools/yaegi-check/main.go | 89 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 89 insertions(+) diff --git a/tools/yaegi-check/main.go b/tools/yaegi-check/main.go index bb7b80a..e01d6cf 100644 --- a/tools/yaegi-check/main.go +++ b/tools/yaegi-check/main.go @@ -139,6 +139,27 @@ const ( 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. @@ -305,6 +326,24 @@ func run() error { })) 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 @@ -399,6 +438,10 @@ func run() error { // 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 @@ -755,6 +798,10 @@ func exerciseHandler(handler http.Handler, builtinContextTokens int) error { return err } + if err := exerciseRequestTimeoutMidBodyStall(handler); err != nil { + return err + } + if err := exerciseFederatedTooLarge(handler); err != nil { return err } @@ -1111,6 +1158,48 @@ func exerciseRequestTimeout(handler http.Handler) error { 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"}]}`, From edd618279f55cbab5f2bee29f9c1c2e4400f9c0d Mon Sep 17 00:00:00 2001 From: Lukasz Raczylo Date: Sun, 23 Aug 2026 02:55:22 +0100 Subject: [PATCH 7/7] docs: correct request-timeout claims found by adversarial review MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit F2: the idle timer resets on DECODED body bytes, after net/http's own gzip decompression. A gzip'd response whose compressed bytes arrive steadily but whose deflate stream yields nothing for the whole window still gets aborted. The previous wording claimed a slow-but-progressing body is always unaffected; adds a qualifying paragraph explaining the gzip case and why fixing it (measuring wire bytes) is out of scope at the five-minute default. F3: reworded the "silence ends a request, never elapsed time" framing. A streaming forwarder reads a chunk, writes it to the client, then reads again; the write is not watched, so time blocked on a slow client counts against the same idle window as a silent provider. Aborting is still correct, but the log line no longer claims the provider specifically caused it. F5: documents that neither timeout is ever retried, and why: a ResponseHeaderTimeout failure wraps context.DeadlineExceeded, which Retry's own transient classification excludes; a mid-body watchdog timeout fires after the retry mechanism has already returned. F8: softened "Every upstream request this gateway makes" to the routes actually covered, and added a "Federated MCP" note: the federated /mcp endpoint's two direct calls are not wrapped by the idle-progress watchdog, but are not unbounded either — they still get ResponseHeaderTimeout from the shared client, plus their own 20s/120s/ 5s context.WithTimeout budgets. Worst case is a bounded stall, not a forever one. --- README.md | 56 +++++++++++++++++++++++++++++++++++++++++++++++++------ 1 file changed, 50 insertions(+), 6 deletions(-) diff --git a/README.md b/README.md index f29665e..484dde3 100644 --- a/README.md +++ b/README.md @@ -850,9 +850,13 @@ request. ## Request timeout -Every upstream request this gateway makes — provider adapter calls, native -passthrough (`/{provider}/...`), and the MCP/A2A target proxy — is bounded -by a progress-based timeout: silence ends a request, never elapsed time. +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 @@ -873,6 +877,29 @@ request forever. Default, when neither `requestTimeout` field is set: 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 @@ -887,9 +914,26 @@ request forever. Default, when neither `requestTimeout` field is set: 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 a provider failure (`502`, `server_error`), naming the - provider, and is never classified as a client disconnect - (`context.Canceled`) — the two stay distinguishable in every log line. + 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