A band may wait longer than the rest; a busy refusal says when to return

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) <noreply@anthropic.com>

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.
 

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
 

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

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`

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.

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: ""},

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)
+	}
+}

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)
 	}
 }
 

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"},

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.

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{