From 497e4b24a1dfe9fa381a6d6772f65a351f6264ec Mon Sep 17 00:00:00 2001 From: Ilker Manap Date: Sat, 10 Oct 2026 16:11:09 +0200 Subject: [PATCH] A band may wait longer than the rest; a busy refusal says when to return MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A knowledge-base import sends far more than the cluster can take at once, and every band waited the same 30 s. A band now has its own queue_ttl, so one for bulk work waits behind people instead of being turned away. And a request turned away because the cluster stayed busy — the queue full (503) or waited as long as its band may (504) — now carries Retry-After: 10. Without it llmgw-kb, which retries only what says when to, counted each such passage as failed; that is what turned parts of the metallurgy import away. Co-Authored-By: Claude Opus 5.5 (1M context) --- CHANGELOG.md | 11 +++++++ docs/clients.md | 3 +- docs/knowledge.md | 18 ++++++++++++ docs/reference.md | 19 ++++++++++++ internal/config/config.go | 4 +++ internal/docgen/reference_gen.go | 2 ++ internal/router/admission/admission_test.go | 32 +++++++++++++++++++++ internal/router/admission/controller.go | 16 +++++++++-- internal/router/finish_test.go | 6 ++-- internal/router/handlers.go | 6 ++++ internal/router/router.go | 1 + 11 files changed, 113 insertions(+), 5 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 99d6d8e..8f01ddc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -26,6 +26,13 @@ APIs change between minors without a deprecation period. ## Unreleased +### Added + +- **A band's own `queue_ttl`** (`router.admission.bands[].queue_ttl`): how + long its requests may wait for room, instead of the admission's 30 s. A + band for bulk work — a knowledge-base import — waits behind people rather + than being turned away. + ### Fixed - **An engine's output no longer breaks when its agent restarts.** Engines @@ -43,6 +50,10 @@ APIs change between minors without a deprecation period. hour of quiet per event, so a budget spent on a team both served was announced twice. The control plane that claims an event in the store (migration 0043) sends it; with the store unreachable it is sent anyway. +- **A request turned away for a busy cluster says when to come back.** A + request that waited as long as it may (504) or found the queue full (503) + had no `Retry-After`, so `llmgw-kb` counted the passage as failed instead + of sending it again; both now say 10 seconds. - The usage sums and exports count pictures (`images`), as the request records did; pictures made before are added to the sums. diff --git a/docs/clients.md b/docs/clients.md index f1247a7..4fe04af 100644 --- a/docs/clients.md +++ b/docs/clients.md @@ -547,7 +547,8 @@ its usual exception with the gateway's message in it. The ones worth knowing: | 401 | No key, a mistyped one, or one that was revoked or has expired. | | 403 | The key may not use that model. | | 429 | A rate limit or budget of the key, its project or its team was reached. `Retry-After` says when to try again, and is absent when waiting will not help — a spent monthly budget. | -| 503 | No model of that name is served (the message lists the ones that are), the model is still loading, or every replica is busy and the request waited as long as it may. The SDKs retry by themselves; a name that is not served will not start to be. | +| 503 | No model of that name is served (the message lists the ones that are), the model is still loading, or every replica is busy and the queue is full. The SDKs retry by themselves; a name that is not served will not start to be. | +| 504 | Every replica was busy for as long as the key's band may wait. `Retry-After` says when to send it again. | ## Priority diff --git a/docs/knowledge.md b/docs/knowledge.md index c714d5a..83ac54a 100644 --- a/docs/knowledge.md +++ b/docs/knowledge.md @@ -146,6 +146,24 @@ Rules that apply: - Every document added or removed is in the audit log (`llmgw ctl audit`), by name. The text is never recorded. +### A large import + +A few thousand documents ask the embedding model for far more than it can +answer at once. Give the key that sends them a band of its own, below the +people's, that may wait for room for minutes rather than the 30 seconds the +others wait: + +```sh +llmgw ctl config set router.admission.bands '[{name: gold, priority: 300, saturation_ceiling: 0.85}, + {name: bronze, priority: 100, saturation_ceiling: 0.3}, {name: bulk, priority: 50, saturation_ceiling: 0.3, queue_ttl: 5m}]' +llmgw keys create import --project Acme/docs --band bulk +``` + +The people's questions then go first, and the import waits behind them. A +request that waits as long as its band allows is answered 504 with +`Retry-After`, and `llmgw-kb` sends it again rather than counting the +passage as failed. It gives up after 15 minutes. + ## For everyone: `llmgw-kb` ### Get it diff --git a/docs/reference.md b/docs/reference.md index 84b3d78..467d8c7 100644 --- a/docs/reference.md +++ b/docs/reference.md @@ -356,6 +356,7 @@ rather than repeated. | `name` | `string` | [admission.Band](#admissionband) | | `priority` | `int` | [admission.Band](#admissionband) | | `saturation_ceiling` | `float64` | [admission.Band](#admissionband) | +| `queue_ttl` | `time.Duration` | [admission.Band](#admissionband) | | `bands` | `[]Band` | [admission.Config](#admissionconfig) | | `default_band` | `string` | [admission.Config](#admissionconfig) | | `queue_ttl` | `time.Duration` | [admission.Config](#admissionconfig) | @@ -385,6 +386,7 @@ rather than repeated. | `name` | `string` | [config.Band](#configband) | | `priority` | `int` | [config.Band](#configband) | | `saturation_ceiling` | `float64` | [config.Band](#configband) | +| `queue_ttl` | `time.Duration` | [config.Band](#configband) | | `prompt_per_1k` | `int64` | [config.Price](#configprice) | | `completion_per_1k` | `int64` | [config.Price](#configprice) | | `per_image` | `int64` | [config.Price](#configprice) | @@ -3013,6 +3015,15 @@ Priority orders bands; higher is served first. SaturationCeiling is the pool saturation above which this band waits. 1.0 means "never hold this band back". +### `queue_ttl` + +`time.Duration` + +QueueTTL, set, is how long this band's requests may wait instead of +the controller's: a band for bulk work — a knowledge-base import, a +backfill — whose requests had rather wait ten minutes behind people +than be turned away after thirty seconds. + ## admission.Config ### `bands` @@ -3226,6 +3237,14 @@ Takes a [Tracing](#tracing) block. `float64` +### `queue_ttl` + +`time.Duration` + +QueueTTL is how long this band's requests may wait for room; the +admission's queue_ttl unless set. Long for a band of bulk work, which +had rather wait behind people than be turned away. + ## config.Price ### `prompt_per_1k` diff --git a/internal/config/config.go b/internal/config/config.go index b136961..0624d05 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -689,6 +689,10 @@ type Band struct { Name string `koanf:"name"` Priority int `koanf:"priority"` SaturationCeiling float64 `koanf:"saturation_ceiling"` + // QueueTTL is how long this band's requests may wait for room; the + // admission's queue_ttl unless set. Long for a band of bulk work, which + // had rather wait behind people than be turned away. + QueueTTL time.Duration `koanf:"queue_ttl"` } // Timeouts bounds each phase of an upstream request. diff --git a/internal/docgen/reference_gen.go b/internal/docgen/reference_gen.go index ec728e3..285a438 100644 --- a/internal/docgen/reference_gen.go +++ b/internal/docgen/reference_gen.go @@ -464,6 +464,7 @@ var Reference = []Block{ {Key: "name", Type: "string", Doc: "", Nested: ""}, {Key: "priority", Type: "int", Doc: "Priority orders bands; higher is served first.", Nested: ""}, {Key: "saturation_ceiling", Type: "float64", Doc: "SaturationCeiling is the pool saturation above which this band waits.\n1.0 means \"never hold this band back\".", Nested: ""}, + {Key: "queue_ttl", Type: "time.Duration", Doc: "QueueTTL, set, is how long this band's requests may wait instead of\nthe controller's: a band for bulk work — a knowledge-base import, a\nbackfill — whose requests had rather wait ten minutes behind people\nthan be turned away after thirty seconds.", Nested: ""}, }}, {Package: "admission", Name: "Config", Title: "admission.Config", Doc: "", Fields: []Field{ {Key: "bands", Type: "[]Band", Doc: "", Nested: ""}, @@ -499,6 +500,7 @@ var Reference = []Block{ {Key: "name", Type: "string", Doc: "", Nested: ""}, {Key: "priority", Type: "int", Doc: "", Nested: ""}, {Key: "saturation_ceiling", Type: "float64", Doc: "", Nested: ""}, + {Key: "queue_ttl", Type: "time.Duration", Doc: "QueueTTL is how long this band's requests may wait for room; the\nadmission's queue_ttl unless set. Long for a band of bulk work, which\nhad rather wait behind people than be turned away.", Nested: ""}, }}, {Package: "config", Name: "Price", Title: "config.Price", Doc: "", Fields: []Field{ {Key: "prompt_per_1k", Type: "int64", Doc: "", Nested: ""}, diff --git a/internal/router/admission/admission_test.go b/internal/router/admission/admission_test.go index be4987f..d1b0cf2 100644 --- a/internal/router/admission/admission_test.go +++ b/internal/router/admission/admission_test.go @@ -511,3 +511,35 @@ func TestSaturationWithNoUsableEndpoints(t *testing.T) { t.Errorf("an empty pool should report full saturation, got %v", got) } } + +func TestABandForBulkWorkWaitsLongerThanTheRest(t *testing.T) { + // A knowledge-base import sends far more than the cluster can take at + // once; turned away after the 30 s every band shared, its requests had + // to be sent again. A band of its own may wait as long as it likes. + s := &fixedSat{} + s.set(1.0) + cfg := testConfig() + cfg.QueueTTL = 100 * time.Millisecond + for i := range cfg.Bands { + if cfg.Bands[i].Name == "bronze" { + cfg.Bands[i].QueueTTL = 600 * time.Millisecond + } + } + c := newController(t, cfg, s.fn) + + start := time.Now() + if _, err := c.Acquire(t.Context(), Request{Model: "m", Band: "gold"}); !errors.Is(err, ErrQueueTimeout) { + t.Fatalf("gold: %v", err) + } + if d := time.Since(start); d > 400*time.Millisecond { + t.Errorf("gold waited %v, past the controller's queue_ttl", d) + } + start = time.Now() + _, err := c.Acquire(t.Context(), Request{Model: "m", Band: "bronze"}) + if !errors.Is(err, ErrQueueTimeout) || !strings.Contains(err.Error(), "600ms") { + t.Fatalf("bronze: %v", err) + } + if d := time.Since(start); d < 500*time.Millisecond { + t.Errorf("bronze gave up after %v, before its own queue_ttl", d) + } +} diff --git a/internal/router/admission/controller.go b/internal/router/admission/controller.go index 6eeb3ed..deeefb6 100644 --- a/internal/router/admission/controller.go +++ b/internal/router/admission/controller.go @@ -19,6 +19,11 @@ type Band struct { // SaturationCeiling is the pool saturation above which this band waits. // 1.0 means "never hold this band back". SaturationCeiling float64 `koanf:"saturation_ceiling"` + // QueueTTL, set, is how long this band's requests may wait instead of + // the controller's: a band for bulk work — a knowledge-base import, a + // backfill — whose requests had rather wait ten minutes behind people + // than be turned away after thirty seconds. + QueueTTL time.Duration `koanf:"queue_ttl"` } // Config configures the controller. @@ -86,6 +91,9 @@ func (c *Config) normalize() error { if b.SaturationCeiling <= 0 { b.SaturationCeiling = 1 } + if b.QueueTTL < 0 { + return fmt.Errorf("band %q: queue_ttl %v is negative", b.Name, b.QueueTTL) + } } if c.DefaultBand == "" { c.DefaultBand = c.Bands[len(c.Bands)-1].Name @@ -279,12 +287,16 @@ func (c *Controller) Acquire(ctx context.Context, r Request) (*Lease, error) { ErrQueueFull, c.cfg.QueueCapacityTotal) } + ttl := c.cfg.QueueTTL + if band.QueueTTL > 0 { + ttl = band.QueueTTL + } w := &waiter{ tenant: tenantKey(r), model: r.Model, ready: make(chan struct{}), enqueued: time.Now(), - deadline: time.Now().Add(c.cfg.QueueTTL), + deadline: time.Now().Add(ttl), } c.queues[idx].push(w) c.mu.Unlock() @@ -312,7 +324,7 @@ func (c *Controller) Acquire(ctx context.Context, r Request) (*Lease, error) { case <-time.After(time.Until(w.deadline)): c.remove(idx, w, "ttl") - return nil, fmt.Errorf("%w after %v", ErrQueueTimeout, c.cfg.QueueTTL) + return nil, fmt.Errorf("%w after %v", ErrQueueTimeout, ttl) } } diff --git a/internal/router/finish_test.go b/internal/router/finish_test.go index a1a467e..5df983c 100644 --- a/internal/router/finish_test.go +++ b/internal/router/finish_test.go @@ -84,10 +84,12 @@ func TestARefusalBeforeDispatchSaysWhetherWaitingHelps(t *testing.T) { }{ {"queue full", func(w http.ResponseWriter) { rt.rejectAdmission(w, &parsedRequest{Model: "m"}, admission.ErrQueueFull, time.Now()) - }, 503, ""}, + }, 503, "10"}, + // Busy, not broken: a client told when to come back comes back, + // where one told nothing — llmgw-kb among them — gives up. {"queued too long", func(w http.ResponseWriter) { rt.rejectAdmission(w, &parsedRequest{Model: "m"}, admission.ErrQueueTimeout, time.Now()) - }, 504, ""}, + }, 504, "10"}, {"displaced for a higher band", func(w http.ResponseWriter) { rt.rejectAdmission(w, &parsedRequest{Model: "m"}, admission.ErrDisplaced, time.Now()) }, 503, "5"}, diff --git a/internal/router/handlers.go b/internal/router/handlers.go index 0da3c84..b17aaec 100644 --- a/internal/router/handlers.go +++ b/internal/router/handlers.go @@ -62,7 +62,13 @@ func (r *Router) rejectAdmission(w http.ResponseWriter, req *parsedRequest, err r.m.observeRequest(r.metricModel(req.Model), "", 499, time.Since(start), nil, 0) return case errors.Is(err, admission.ErrQueueTimeout): + // The cluster was busy for as long as the band may wait, not broken: + // said so a client comes back. Without it llmgw-kb, which retries + // only what says when to, counted every such passage as failed. status = http.StatusGatewayTimeout + w.Header().Set("Retry-After", "10") + case errors.Is(err, admission.ErrQueueFull): + w.Header().Set("Retry-After", "10") case errors.Is(err, admission.ErrDisplaced): // Turned away for higher-priority work, not for good: said so the // client backs off and comes back rather than giving up. diff --git a/internal/router/router.go b/internal/router/router.go index 0c11c1c..9771443 100644 --- a/internal/router/router.go +++ b/internal/router/router.go @@ -432,6 +432,7 @@ func admissionConfig(rc config.Router) admission.Config { Name: b.Name, Priority: b.Priority, SaturationCeiling: b.SaturationCeiling, + QueueTTL: b.QueueTTL, }) } return admission.Config{