A job waiting for the card says what holds it and about how long
CI / test (push) Failing after 6m16s

A 3D job waiting for the picture model to give up the one card had an empty
progress; its client could not tell waiting from stuck. The control plane
says which model holds the cards and when it lets go, plus the waiting
model's last start; the 503 carries it as headers, and a job keeps it as
progress {state: waiting_gpu, held_by, eta_seconds}.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Ilker ManapandClaude Opus 5.5 committed 2026-10-10 22:25:27 +02:00
1 parent cf2edfd995
commit b1490dbf30
12 files changed
+252 -8

No files matched your search

+14
View File
@@ -32,6 +32,20 @@ APIs change between minors without a deprecation period.
needed, generated from a route table a test holds to the router's own;
and `/docs/api`, Swagger UI (served from the gateway) over it and, for
staff, over the admin API's document.
- **A job waiting for the card says so**: `progress` is
`{"state":"waiting_gpu","held_by":…,"eta_seconds":…}` while another
on-demand model holds it; a direct request's 503 carries
`X-Llmgw-Waiting-For` and `X-Llmgw-Eta-Seconds`.
- **On-demand models in `/v1/models`**, with `loaded` saying whether one is
up now; a picture is cut out and put on grey before TripoSR sees it
(`LLMGWPrepareForMesh`, rembg when it has no alpha).
### Fixed
- A model waiting for another on-demand model to give up the card failed at
once as "cannot be placed"; it now waits, and a job with it.
- A model evicted for being idle was placed again on the next pass, ahead of
the one it made room for.
## v0.14.3 — 2026-10-10
+7 -1
View File
@@ -2072,7 +2072,13 @@ curl -X POST http://gateway:8080/v1/3d/generations/job_…/cancel -H "Authorizat
- **Status:** `queued`, `running`, `completed`, `failed` (with `error`),
`cancelling`, `cancelled`, `expired` (not finished within 24 h).
`progress` carries what the engine said of the work so far, once it says
anything; `GET /v1/3d/generations` lists the caller's jobs.
anything; `GET /v1/3d/generations` lists the caller's jobs. While the
model waits for another on-demand model to give up the card,
`progress` is `{"state":"waiting_gpu","held_by":"flux-schnell","eta_seconds":N}`:
N is when the holder is stopped (idle and old enough to evict, or at its
idle timeout) plus how long the model's last start took, and is left out
when that is not known. A request made directly gets the same as a 503
with `X-Llmgw-Waiting-For`, `X-Llmgw-Eta-Seconds` and `Retry-After`.
- **Refused at the door:** no key, a model that is not there or not in the
key's allowlist, a request the model cannot serve. Refused when served,
and said as the job's error: a spent budget, the guardrails, a model
+79
View File
@@ -169,6 +169,85 @@ func (r *Reconciler) Unservable(model string) (string, bool) {
return m.Name + " cannot be placed on any node: " + firstLine(res.Explain()), true
}
// WaitFor says which model holds the cards an on-demand model is waiting
// for, and about how long until it serves: until the holder lets go — evicted
// once idle and old enough, or stopped at its own idle timeout — and then its
// own start, as long as its last one took. ok is false when it is not waiting
// on another model; eta is zero when when the holder lets go is not known.
func (r *Reconciler) WaitFor(model string) (heldBy string, eta time.Duration, ok bool) {
cat := r.cat.Load()
m, found := cat.Resolve(model)
if !found || !m.Scaling.OnDemand() {
return "", 0, false
}
type holder struct {
v victim
release time.Time
}
now := r.now()
var holders []holder
r.mu.Lock()
for _, rep := range r.replicas {
if !rep.State.Counts() || len(nodesOf(rep)) != 1 {
continue
}
if rep.Model == m.Name {
r.mu.Unlock()
return "", 0, false // its own replica is on its way
}
other, found := cat.Resolve(rep.Model)
if !found || !other.Scaling.OnDemand() {
continue
}
h := holder{v: victim{ID: rep.ID, Model: rep.Model}}
if !rep.LastUsedAt.IsZero() {
evict := rep.LastUsedAt.Add(r.cfg.EvictionGrace)
if !rep.ReadyAt.IsZero() && rep.ReadyAt.Add(r.cfg.MinLifetime).After(evict) {
evict = rep.ReadyAt.Add(r.cfg.MinLifetime)
}
h.release = evict
if idle := rep.LastUsedAt.Add(r.idleTimeout(other)); idle.Before(evict) {
h.release = idle
}
}
holders = append(holders, h)
}
cold := r.coldStart[m.Name]
r.mu.Unlock()
// The one whose going alone makes room, letting go soonest; failing
// that, all of them, and the last to let go.
var best *holder
for i := range holders {
h := &holders[i]
if !r.fitsWithout(m, nil, []victim{h.v}) {
continue
}
if best == nil || (!h.release.IsZero() && (best.release.IsZero() || h.release.Before(best.release))) {
best = h
}
}
if best == nil {
all := make([]victim, len(holders))
for i, h := range holders {
all[i] = h.v
}
if len(all) == 0 || !r.fitsWithout(m, nil, all) {
return "", 0, false
}
for i := range holders {
h := &holders[i]
if best == nil || h.release.IsZero() || (!best.release.IsZero() && h.release.After(best.release)) {
best = h
}
}
}
if best.release.IsZero() {
return best.v.Model, 0, true
}
return best.v.Model, max(best.release.Sub(now), 0) + cold, true
}
func firstLine(s string) string {
s = strings.TrimSpace(s)
if i := strings.IndexByte(s, '\n'); i >= 0 {
@@ -264,6 +264,9 @@ type Reconciler struct {
// because it is state about this moment and not about what the model is.
// A model nobody has wanted has no entry and is simply not running.
wanted map[string]time.Time
// coldStart is how long each model's last replica took from placed to
// ready, for telling a request waiting on it about how long it waits.
coldStart map[string]time.Duration
// restartQueue holds, per model, the replicas an operator asked to be
// restarted and not yet begun. One is replaced at a time: replacing them
// all at once would need the memory of every replica twice over.
@@ -330,6 +333,7 @@ func New(cfg Config, cat *catalog.Live, inv *inventory.Store, disp Dispatcher, l
attempts: map[string]int{},
reservedPorts: map[int]struct{}{},
wanted: map[string]time.Time{},
coldStart: map[string]time.Duration{},
restartQueue: map[string][]string{},
restartInPlace: map[string]bool{},
restartPace: map[string]RestartPace{},
@@ -1371,6 +1375,9 @@ func (r *Reconciler) markShard(replicaID, shardID string, st supervisor.State, e
// counting from scratch rather than from an old streak.
r.attempts[rep.Model] = 0
delete(r.nextAttempt, rep.Model)
if !rep.CreatedAt.IsZero() {
r.coldStart[rep.Model] = r.now().Sub(rep.CreatedAt)
}
}
}
@@ -1745,6 +1745,33 @@ func TestAnEvictedModelDoesNotTakeTheCardBack(t *testing.T) {
}
}
func TestAModelWaitingOnAnotherIsToldWhichAndForAboutHowLong(t *testing.T) {
h := newHarness(t, testModel(1), gpuNode("1", 4))
h.r.cat.Store(twoOnDemandModels(t))
h.r.Want("llama")
h.r.Reconcile(t.Context())
h.markReady()
reps := h.r.Replicas()
if len(reps) != 1 {
t.Fatalf("setup: %d replicas", len(reps))
}
h.r.RecordUsage(map[string]time.Time{reps[0].ID: h.clock.now()})
h.advance(time.Minute)
h.r.Want("second")
h.r.Reconcile(t.Context())
// Evicted once it has been up MinLifetime, a minute of which has gone;
// second has never started, so there is no start of its to add.
by, eta, ok := h.r.WaitFor("second")
if !ok || by != "llama" || eta != DefaultConfig().MinLifetime-time.Minute {
t.Errorf("WaitFor = %q %s %v", by, eta, ok)
}
// The model holding the cards waits on nobody.
if _, _, ok := h.r.WaitFor("llama"); ok {
t.Error("a model with its own replica was said to be waiting")
}
}
func TestAModelWaitingForAnotherOnDemandOneToStepAsideIsNotUnservable(t *testing.T) {
// A 3D model asked for straight after a picture, on a cluster with one
// card: the picture model holds it, too freshly started to evict. That
+35
View File
@@ -2,6 +2,7 @@ package router
import (
"context"
"fmt"
"time"
"gitea.mynodes.xyz/ilker/llmgw/internal/notify"
@@ -26,6 +27,40 @@ type unservable interface {
Unservable(model string) (string, bool)
}
// waiter is an activator that can say which model holds the cards an
// on-demand one waits for, and about how long the wait is.
type waiter interface {
WaitFor(model string) (heldBy string, eta time.Duration, ok bool)
}
// waitingErr is errLoading for a model waiting on another to give up its
// cards: who holds them, and the wait when it can be told.
type waitingErr struct {
heldBy string
eta time.Duration
}
func (e *waitingErr) Error() string {
if e.eta > 0 {
return fmt.Sprintf("%s: waiting for %s to give up the GPU; expected in about %d s",
errLoading, e.heldBy, int(e.eta.Seconds()))
}
return fmt.Sprintf("%s: waiting for %s to give up the GPU", errLoading, e.heldBy)
}
func (e *waitingErr) Unwrap() error { return errLoading }
// loadingErr is why a wanted model is not serving yet: waiting on another
// model's cards when the activator can say so, else simply loading.
func (r *Router) loadingErr(model string) error {
if w, ok := r.activator.(waiter); ok {
if by, eta, waiting := w.WaitFor(model); waiting {
return &waitingErr{heldBy: by, eta: eta}
}
}
return fmt.Errorf("%w: %s", errLoading, "the model is loading; retry shortly")
}
// whyUnservable is the activator's reason a model cannot be served, if any.
func (r *Router) whyUnservable(model string) (string, bool) {
u, ok := r.activator.(unservable)
+13 -1
View File
@@ -730,7 +730,19 @@ func (r *Router) finish(w http.ResponseWriter, req *parsedRequest, res dispatchR
// is told so will make it. Without the header a well-behaved client
// backs off exactly as it would from a permanent failure.
status = http.StatusServiceUnavailable
w.Header().Set("Retry-After", "30")
retry := 30
// Waiting on another model's cards: who holds them and for about
// how long, which a job keeps as its progress.
var we *waitingErr
if errors.As(res.err, &we) {
w.Header().Set(waitingForHeader, we.heldBy)
if we.eta > 0 {
secs := int(we.eta.Seconds())
w.Header().Set(etaHeader, strconv.Itoa(secs))
retry = min(max(secs, 5), retry)
}
}
w.Header().Set("Retry-After", strconv.Itoa(retry))
writeError(w, status, "server_error", res.err.Error())
case errors.Is(res.err, errNoCapacity):
status = http.StatusServiceUnavailable
+16
View File
@@ -283,3 +283,19 @@ func TestJobsAreTheirOwners(t *testing.T) {
t.Errorf("another project read a job's result: %d", code)
}
}
func TestAJobWaitingOnAnotherModelsCardsSaysSo(t *testing.T) {
c := newJobComfy(t)
rt, secret := jobRouter(t, c)
rt.cfg.ActivationWait = 10 * time.Millisecond
rt.SetActivator(&waitingActivator{fakeActivator: fakeActivator{onDemand: map[string]bool{"sleepy": true}},
heldBy: "flux-schnell", eta: 90 * time.Second})
j := queueJob(t, rt, "job_w", store.JobQueued, `{"model":"sleepy","prompt":"a cube"}`)
rt.wakeJobs()
got := waitJob(t, rt.Handler(), secret, j.ID, func(j map[string]any) bool { return j["progress"] != nil })
p, _ := got["progress"].(map[string]any)
if got["status"] != "running" || p["state"] != "waiting_gpu" || p["held_by"] != "flux-schnell" || p["eta_seconds"] != 90.0 {
t.Errorf("job %v", got)
}
}
+12
View File
@@ -3,6 +3,7 @@ package router
import (
"bytes"
"context"
"encoding/json"
"errors"
"fmt"
"net/http"
@@ -180,6 +181,17 @@ func (r *Router) serveJob(ctx context.Context, j *store.Job) error {
// because waiting does not help.
if wait := w.header.Get("Retry-After"); wait != "" && (status == http.StatusTooManyRequests || status >= 500) &&
time.Now().Before(j.ExpiresAt) {
if by := w.header.Get(waitingForHeader); by != "" {
// Kept as its progress: whoever asks after it is told what
// it waits on, rather than nothing at all.
p := map[string]any{"state": "waiting_gpu", "held_by": by}
if eta, err := strconv.Atoi(w.header.Get(etaHeader)); err == nil && eta > 0 {
p["eta_seconds"] = eta
}
if b, err := json.Marshal(p); err == nil {
_ = r.store.SetJobProgress(ctx, j.ID, string(b), "")
}
}
secs, _ := strconv.Atoi(wait)
select {
case <-rctx.Done():
+1 -2
View File
@@ -1023,8 +1023,7 @@ func (r *Router) dispatchOne(ctx context.Context, w http.ResponseWriter, req *pa
// Asked for, still loading. Saying "no endpoint serves this" would
// be a lie, and a client told to retry will succeed where one told
// the model does not exist gives up.
return dispatchResult{err: fmt.Errorf("%w: %s", errLoading,
"the model is loading; retry shortly")}
return dispatchResult{err: r.loadingErr(model)}
}
return dispatchResult{err: fmt.Errorf("%w: %s", errNoCapacity, noEndpointFor(model, slices.DeleteFunc(r.servedModels(), r.hiddenModel)))}
}
+33
View File
@@ -808,3 +808,36 @@ func TestABatchOfEmbeddingsIsCheckedTextByText(t *testing.T) {
t.Errorf("a passage over the context was let through: %d", rec.Code)
}
}
type waitingActivator struct {
fakeActivator
heldBy string
eta time.Duration
}
func (a *waitingActivator) WaitFor(string) (string, time.Duration, bool) {
return a.heldBy, a.eta, true
}
func TestAModelWaitingOnAnothersCardsSaysWhoAndForHowLong(t *testing.T) {
// A 3D job waiting for the picture model to give up the one card was
// told nothing at all; its client could not tell waiting from stuck.
e := newFakeEngine(t, "m")
cfg := baseConfig(config.Endpoint{ID: "a", URL: e.srv.URL, Models: []string{"m"}})
cfg.Router.ActivationWait = 50 * time.Millisecond
rt := newTestRouter(t, cfg)
rt.SetActivator(&waitingActivator{fakeActivator: fakeActivator{onDemand: map[string]bool{"sleepy": true}},
heldBy: "flux-schnell", eta: 12 * time.Second})
rec := post(t, rt.Handler(), map[string]any{"model": "sleepy", "messages": []any{}}, nil)
if rec.Code != http.StatusServiceUnavailable {
t.Fatalf("status %d, expected 503", rec.Code)
}
if rec.Header().Get(waitingForHeader) != "flux-schnell" || rec.Header().Get(etaHeader) != "12" ||
rec.Header().Get("Retry-After") != "12" {
t.Errorf("headers %v", rec.Header())
}
if !strings.Contains(rec.Body.String(), "waiting for flux-schnell") {
t.Errorf("body %s", rec.Body.String())
}
}
+8 -4
View File
@@ -11,10 +11,14 @@ import (
// project: OpenAI's metadata (at most 16 keys, as OpenAI allows) and the
// x-llmgw-tags header, which wins for a key both give.
const (
tagsHeader = "x-llmgw-tags"
maxTags = 16
maxTagValue = 512
maxEndUser = 256
tagsHeader = "x-llmgw-tags"
// waitingForHeader and etaHeader say, on a 503, which model holds the
// cards the one asked for waits on, and about how many seconds it waits.
waitingForHeader = "X-Llmgw-Waiting-For"
etaHeader = "X-Llmgw-Eta-Seconds"
maxTags = 16
maxTagValue = 512
maxEndUser = 256
)
var tagKey = regexp.MustCompile(`^[A-Za-z0-9_.-]{1,64}$`)