mirror of
https://github.com/lukaszraczylo/traefik-llmgateway.git
synced 2026-10-18 05:47:38 +00:00
fix(providers): address adversarial review findings F1, F4, F6, F9
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.
This commit is contained in:
+104
-34
@@ -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
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user