From bc4c5d5ac9a84dccdec04021ef7d8ab8df005aa3 Mon Sep 17 00:00:00 2001 From: Ilker Manap Date: Fri, 16 Oct 2026 08:35:28 +0200 Subject: [PATCH] A judge replica that hangs is overtaken by another at half the budget On the three-VM cluster about one answer in twenty-five came back as content_filter: both judge replicas timed out within the same millisecond. The tries shared the policy's 5 s budget, so the first replica, busy on an oversubscribed machine, used it all and the retry on the second had nothing left. A judge request now goes to a second replica too when the first has not answered by half the budget, and the first answer is taken. The first is not cut short. Co-Authored-By: Claude Opus 5.5 (1M context) --- CHANGELOG.md | 5 +++ internal/router/guard_test.go | 37 ++++++++++++++++++++ internal/router/handlers.go | 5 +++ internal/router/judge.go | 62 +++++++++++++++++++++++++++++++++- internal/router/router.go | 10 ++++++ internal/router/router_test.go | 6 ++++ 6 files changed, 124 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index b1e1996b..67957f39 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -98,6 +98,11 @@ APIs change between minors without a deprecation period. round after it returns, so the leader renews it, and a leader that is truly gone is replaced a round later. With that, the default lease is 10 seconds (was 15). A node notices a control plane that died without closing its connection in 8 seconds (was 15), so it is on its way to the new leader as that one takes over. +- **A judge replica that hangs no longer costs the request.** All the tries of a judge request shared the policy's + budget, so a replica that hung used it all and the try on another replica had none left: under `fail: closed` the + request was refused while a judge stood idle. When the first replica has not answered by half the budget, the same + question now goes to another replica too, and the first answer is taken. The first replica is not cut short, which on + a slow machine would refuse answers it was about to give. - **A new leader waits for the judge instead of refusing at once.** The moment a control plane takes the lead it knows no judge yet. When every judge fails at once, they are asked again every 250 ms while the policy's `latency_budget` lasts, so `fail: closed` no longer answers every request in those seconds with 503 (about 17 seconds of them in the test). A diff --git a/internal/router/guard_test.go b/internal/router/guard_test.go index 8dee911b..64b0f513 100644 --- a/internal/router/guard_test.go +++ b/internal/router/guard_test.go @@ -269,3 +269,40 @@ func TestSignInThroughAProviderNeedsNoAPIKey(t *testing.T) { t.Errorf("%d %s", rec.Code, rec.Body.String()) } } + +// One replica of the judge hangs: the other is asked at half the budget, and +// answers within it, rather than the hung one taking the whole budget and +// the request being refused. +func TestAJudgeThatHangsIsOvertakenByAnother(t *testing.T) { + hung, quick := newFakeEngine(t, "judge"), newFakeEngine(t, "judge") + hung.hang.Store(true) + cfg := baseConfig(config.Endpoint{ID: "hung", URL: hung.srv.URL, Models: []string{"judge"}}, + config.Endpoint{ID: "quick", URL: quick.srv.URL, Models: []string{"judge"}}) + judge := catModel("judge", nil) + judge.Role, judge.Guard = catalog.RoleGuard, &catalog.GuardModel{Format: "qwen3guard"} + cfg.Models = []catalog.Model{judge} + rt := newTestRouter(t, cfg) + deadline := time.Now().Add(3 * time.Second) + for len(rt.reg.Snapshots("judge")) < 2 && time.Now().Before(deadline) { + time.Sleep(20 * time.Millisecond) + } + if n := len(rt.reg.Snapshots("judge")); n != 2 { + t.Fatalf("%d replicas of the judge", n) + } + body := []byte(`{"model":"judge","messages":[{"role":"user","content":"x"}]}`) + for i := 0; i < 12 && hung.calls.Load() == 0; i++ { + ctx, cancel := context.WithTimeout(t.Context(), time.Second) + start := time.Now() + out, err := rt.Judge(ctx, "judge", "/v1/chat/completions", body) + cancel() + if err != nil || !strings.Contains(string(out), `"hi"`) { + t.Fatalf("no verdict in %s: %v", time.Since(start), err) + } + if took := time.Since(start); took > 900*time.Millisecond { + t.Errorf("answered in %s, at the end of the budget", took) + } + } + if hung.calls.Load() == 0 { + t.Fatal("the hung replica was never asked first") + } +} diff --git a/internal/router/handlers.go b/internal/router/handlers.go index ffe01a7f..acdf2d7a 100644 --- a/internal/router/handlers.go +++ b/internal/router/handlers.go @@ -474,6 +474,11 @@ type parsedRequest struct { // internal is a request the gateway makes of itself — a judge's — which // no guard filters. internal bool + // avoid are endpoints this request is not sent to while another will do, + // and picked is told the endpoint it is sent to: a judge asked twice + // asks two replicas. + avoid []string + picked func(endpointID string) // cache is where its answer is kept, when its project reuses answers // and this one may be. cache cacheState diff --git a/internal/router/judge.go b/internal/router/judge.go index eb4bda3c..638ec0d8 100644 --- a/internal/router/judge.go +++ b/internal/router/judge.go @@ -5,6 +5,8 @@ import ( "context" "fmt" "net/http" + "sync" + "time" "gitea.mynodes.xyz/ilker/llmgw/internal/guard" ) @@ -18,6 +20,63 @@ func (r *Router) Judge(ctx context.Context, model, path string, body []byte) ([] if path != "/v1/chat/completions" && path != "/v1/completions" { return nil, fmt.Errorf("a judge is asked at /v1/chat/completions or /v1/completions, not %s", path) } + deadline, ok := ctx.Deadline() + canonical, _ := r.resolve(model) + if !ok || len(r.reg.Snapshots(canonical)) < 2 { + return r.judgeOnce(ctx, path, body, nil, nil) + } + + // Asked of a second replica too when the first has not answered by half + // the budget, and the first answer taken. One replica that hung — busy, + // or on a machine that is going — took the whole budget, and the retry + // on the other found none left: the request was refused (fail: closed) + // though a judge stood idle. The first is not cut short, which on a slow + // machine would refuse what it was about to answer. + ctx, cancel := context.WithCancel(ctx) + defer cancel() + type answer struct { + out []byte + err error + } + answers := make(chan answer, 2) + var mu sync.Mutex + var first string + go func() { + out, err := r.judgeOnce(ctx, path, body, nil, func(id string) { + mu.Lock() + first = id + mu.Unlock() + }) + answers <- answer{out, err} + }() + hedge := time.NewTimer(time.Until(deadline) / 2) + defer hedge.Stop() + asking := 1 + var lastErr error + for asking > 0 { + select { + case a := <-answers: + asking-- + if a.err == nil { + return a.out, nil + } + lastErr = a.err + case <-hedge.C: + mu.Lock() + avoid := []string{first} + mu.Unlock() + asking++ + go func() { + out, err := r.judgeOnce(ctx, path, body, avoid, nil) + answers <- answer{out, err} + }() + } + } + return nil, lastErr +} + +// judgeOnce asks one judge request of the router, as it would a client's. +func (r *Router) judgeOnce(ctx context.Context, path string, body []byte, avoid []string, picked func(string)) ([]byte, error) { hreq, err := http.NewRequestWithContext(ctx, http.MethodPost, path, bytes.NewReader(body)) if err != nil { return nil, err @@ -30,13 +89,14 @@ func (r *Router) Judge(ctx context.Context, model, path string, body []byte) ([] req.Stream = false req.RequestID = requestID("") req.Header = http.Header{"X-Request-Id": []string{req.RequestID}} + req.avoid, req.picked = avoid, picked w := &bufferWriter{header: http.Header{}} res := r.dispatch(ctx, w, req) if res.err != nil { return nil, res.err } if w.status != 0 && w.status != http.StatusOK { - return nil, fmt.Errorf("the judge %s answered %d", model, w.status) + return nil, fmt.Errorf("the judge %s answered %d", req.Model, w.status) } return w.buf.Bytes(), nil } diff --git a/internal/router/router.go b/internal/router/router.go index 4f403bd0..5d6d1f1b 100644 --- a/internal/router/router.go +++ b/internal/router/router.go @@ -1022,6 +1022,13 @@ func (r *Router) dispatchOne(ctx context.Context, w http.ResponseWriter, req *pa } candidates = visited } + if len(req.avoid) > 0 { + if rest := slices.DeleteFunc(slices.Clone(candidates), func(e endpoint.Snapshot) bool { + return slices.Contains(req.avoid, e.ID) + }); len(rest) > 0 { + candidates = rest + } + } if len(candidates) == 0 { // Nothing is serving it. That may be because nothing ever will, or // because it loads when asked and nobody has asked yet — and the second @@ -1106,6 +1113,9 @@ func (r *Router) dispatchOne(ctx context.Context, w http.ResponseWriter, req *pa candidates = without(candidates, dec.Endpoint.ID) continue } + if req.picked != nil { + req.picked(dec.Endpoint.ID) + } // The replica loads the adapter first if it does not hold it. // Counted against its inflight while it does, so the requests diff --git a/internal/router/router_test.go b/internal/router/router_test.go index f953aea1..f57cf746 100644 --- a/internal/router/router_test.go +++ b/internal/router/router_test.go @@ -42,6 +42,8 @@ type fakeEngine struct { last atomic.Value // traceparent is the one the last completion came with. traceparent atomic.Value + // hang, when set, has a completion wait until its request is given up. + hang atomic.Bool } func newFakeEngine(t *testing.T, model string) *fakeEngine { @@ -63,6 +65,10 @@ func newFakeEngine(t *testing.T, model string) *fakeEngine { http.Error(w, "unavailable", code) return } + if f.hang.Load() { + <-r.Context().Done() + return + } w.Header().Set("Content-Type", "application/json") fmt.Fprintf(w, `{"model":%q,"choices":[{"message":{"content":"hi"}}],"usage":{"prompt_tokens":4,"completion_tokens":2}}`, model) })