From 3e3517eb6227944c349003c394366b1f72b9b003 Mon Sep 17 00:00:00 2001 From: Lukasz Raczylo Date: Sun, 18 Oct 2026 13:22:29 +0100 Subject: [PATCH] feat(config): advertise a concurrency limit to clients Self-hosted upstreams serve few requests at once. A client that opens many requests overloads them. The new maxConcurrency option sends X-Concurrency-Limit on every response. Clients that honour the header cap their own in-flight requests. The default of 0 sends no header, so behaviour does not change. --- README.md | 1 + concurrency_header_test.go | 55 ++++++++++++++++++++++++++++++++++++++ llmgateway.go | 17 ++++++++++++ 3 files changed, 73 insertions(+) create mode 100644 concurrency_header_test.go diff --git a/README.md b/README.md index 2b09c46..ea494b4 100644 --- a/README.md +++ b/README.md @@ -383,6 +383,7 @@ config accepts them as YAML, which decodes to the same JSON shape. | `requireModelPrefix` | `bool` | `false` | When `true`, a model resolves only as `provider/model` or as a `modelAliases` id. A bare id returns 404 `unknown model` and `GET /v1/models` lists no bare ids. `false` keeps the sorted-first bare-id collision winner. | | `stripClientAnthropicBeta` | `bool` | `false` | When `true`, `POST /v1/messages` does not forward the client's `anthropic-beta` header to the provider, as native passthrough already does. `false` forwards it, so a client can enable beta features on the operator's own Anthropic account. | | `modelMeta` | `map[string]*ModelMetaConfig` | `{}` | Per-model context-window and cost overrides, keyed by exact `provider/model` or a bare model id — see [Model metadata](#model-metadata). Omitted or empty means no overrides; every model's metadata still resolves through discovery and the built-in table. | +| `maxConcurrency` | `int` | `0` | When above `0`, every response carries `X-Concurrency-Limit: `, so clients that honour it (omps) keep at most `n` requests in flight. Set it to what your upstreams serve at once. The gateway only advertises the value; it does not queue or reject on it. `0` sends no header. Negative values are rejected at startup. | | `passthroughUnknown` | `bool` | `false` | `false`: a request matching none of the plugin's routes gets a 404 JSON envelope. `true`: it falls through to the router's own backing service. **Does not cover the three media routes** — see the upgrade note below. | | `maxRequestBytes` | `int` | `0` (4 MiB) | Raises the unified chat/embeddings request-body cap. `0` or negative keeps 4 MiB; a positive value is clamped to 64 MiB. Pair with `maxInFlightBodyRequests` and the pod memory limit: an adversarial body decodes to ~12x live heap. | | `maxInFlightBodyRequests` | `int` | `0` (self-tuned) | Caps how many unified/media requests may concurrently be inside their own request-body read+decode step — see [Body-admission limit](#body-admission-limit). `0` self-tunes from the host's own `GOMAXPROCS`; a positive value pins an exact cap instead, clamped to a maximum of 10,000. | diff --git a/concurrency_header_test.go b/concurrency_header_test.go new file mode 100644 index 0000000..131ea47 --- /dev/null +++ b/concurrency_header_test.go @@ -0,0 +1,55 @@ +package traefikllmgateway + +import ( + "context" + "net/http" + "net/http/httptest" + "testing" +) + +func newConcurrencyTestGateway(t *testing.T, maxConcurrency int) (http.Handler, error) { + t.Helper() + cfg := CreateConfig() + cfg.MaxConcurrency = maxConcurrency + cfg.Providers = map[string]*ProviderConfig{ + "openai": {Type: "openai", APIKey: "sk-test", Models: []string{"gpt-test"}}, + } + cfg.Groups = map[string]*GroupConfig{"default": {}} + cfg.Users = &UsersConfig{Inline: []*UserConfig{{Name: "tester", Group: "default", APIKey: "sk-user"}}} + return New(context.Background(), http.HandlerFunc(func(http.ResponseWriter, *http.Request) {}), cfg, "llmgw") +} + +func TestMaxConcurrency_AdvertisedOnResponses(t *testing.T) { + t.Parallel() + tests := []struct { + name string + header string + max int + }{ + {"configured limit is advertised", "3", 3}, + {"zero sends no header", "", 0}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + h, err := newConcurrencyTestGateway(t, tc.max) + if err != nil { + t.Fatalf("New: %v", err) + } + req := httptest.NewRequest(http.MethodGet, "/v1/models", nil) + req.Header.Set("Authorization", "Bearer sk-user") + rec := httptest.NewRecorder() + h.ServeHTTP(rec, req) + if got := rec.Header().Get("X-Concurrency-Limit"); got != tc.header { + t.Fatalf("X-Concurrency-Limit = %q, want %q", got, tc.header) + } + }) + } +} + +func TestMaxConcurrency_NegativeRejected(t *testing.T) { + t.Parallel() + if _, err := newConcurrencyTestGateway(t, -1); err == nil { + t.Fatal("New accepted a negative maxConcurrency") + } +} diff --git a/llmgateway.go b/llmgateway.go index cab4dec..e493741 100644 --- a/llmgateway.go +++ b/llmgateway.go @@ -12,6 +12,7 @@ import ( "net/url" "os" "runtime" + "strconv" "strings" "sync" "time" @@ -183,6 +184,13 @@ type Config struct { // resolved (404 unknown model) and is not listed by GET /v1/models. It // removes the sorted-first bare-id collision winner. Defaults to false. RequireModelPrefix bool `json:"requireModelPrefix,omitempty"` + // MaxConcurrency, when > 0, is advertised on every response as + // X-Concurrency-Limit so clients that honour it (omps) cap their own + // in-flight requests to this gateway. It documents what the upstreams + // can serve at once; the gateway does not queue or reject on it. 0 + // (the default) sends no header. Negative values are rejected at + // construction. + MaxConcurrency int `json:"maxConcurrency,omitempty"` } // ProviderConfig describes one upstream LLM provider. @@ -927,6 +935,9 @@ func newGateway(ctx context.Context, next http.Handler, config *Config, name str if err := validateTargetURLs(config); err != nil { return nil, err } + if config.MaxConcurrency < 0 { + return nil, errors.New("llmgateway: maxConcurrency must not be negative") + } // Finding F4, 2026-09 review: validate every Config.Pricing override // at construction — non-nil, finite, >= 0 — rather than letting a // negative/NaN/Inf entry reach costMicrosKnownFor (pricing.go), where it @@ -1347,9 +1358,15 @@ func attachUsersFile(auth *authStore, path string, log gatewayLogger) error { } // ServeHTTP is the internal router. It grows in later tasks. +// concurrencyLimitHeader advertises Config.MaxConcurrency to clients. +const concurrencyLimitHeader = "X-Concurrency-Limit" + func (g *Gateway) ServeHTTP(w http.ResponseWriter, r *http.Request) { sw := &statusTrackingWriter{ResponseWriter: w} defer recoverPanic(sw, g) + if g.cfg.MaxConcurrency > 0 { + sw.Header().Set(concurrencyLimitHeader, strconv.Itoa(g.cfg.MaxConcurrency)) + } // pruneRejectionsAfterReload only runs when maybeReload just performed // a REAL reload (users_file.go's own doc comment explains why it must // never run on every request) — a deleted file-sourced user must not