diff --git a/CHANGELOG.md b/CHANGELOG.md index 2548e38..0a396bd 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -37,6 +37,10 @@ APIs change between minors without a deprecation period. ### Fixed +- **The judge init installs is no longer replaced by the leader's own.** Just after the control plane started, the + leader saw a policy naming a judge that init was still fetching, and served one itself. Its entry replaced init's, so + a second judge ran with the leader's sizing, and with judge machines it ran off them. The leader now waits ten minutes + after it starts before it serves a judge itself. If someone adds one while it fetches, theirs stays. - **A judge the leader serves by itself is sized as init's is.** When a guard policy names a judge the catalog lacks, the leader serves it. That judge reserved 3 GB and no cores, and ran llama.cpp's prompt cache, which grew one to 3.2 GB. Without cores of its own, a judgment on a busy node missed its 5 s budget, and `fail: closed` refused the diff --git a/internal/cli/judgekeeper.go b/internal/cli/judgekeeper.go index 648a70b..b1778da 100644 --- a/internal/cli/judgekeeper.go +++ b/internal/cli/judgekeeper.go @@ -32,6 +32,12 @@ type judgeKeeper struct { // tried is when each judge was last asked for, so one that cannot be // fetched is not asked for again on every look. tried map[string]time.Time + // grace is how long after its first look the keeper serves no judge + // itself: init adds its own judge just after the control plane starts, + // sized and placed as the install says, and one the leader added first + // was what init then kept. Zero waits for nothing. + grace time.Duration + first time.Time // What refitting an earlier install's judge needs: the catalog to change // its profile in, the replicas to restart, and the nodes' CPUs. Any nil @@ -51,6 +57,9 @@ type judgeKeeper struct { // asked for again. const judgeRetry = 10 * time.Minute +// judgeGrace is the keeper's grace on a control plane just started. +const judgeGrace = 10 * time.Minute + func (k *judgeKeeper) run(ctx context.Context, every time.Duration) { t := time.NewTicker(every) defer t.Stop() @@ -84,6 +93,12 @@ func (k *judgeKeeper) check() { have[m.Name] = true k.refit(m) } + if k.first.IsZero() { + k.first = now() + } + if now().Sub(k.first) < k.grace { + return + } busy := map[string]bool{} for _, j := range k.serve.Jobs() { if j.State == modelserve.Running { @@ -128,6 +143,8 @@ func (k *judgeKeeper) check() { } req := kj.serveRequest(opts) req.By = "llmgw (a guard policy names it)" + // init, or an operator, may add it while this fetches: theirs stays. + req.IfAbsent = true job, err := k.serve.Start(req) if err != nil { k.log.Error("could not start serving a judge a policy names", "judge", name, "err", err) diff --git a/internal/cli/judgekeeper_test.go b/internal/cli/judgekeeper_test.go index d115772..5ffea1d 100644 --- a/internal/cli/judgekeeper_test.go +++ b/internal/cli/judgekeeper_test.go @@ -52,7 +52,8 @@ func TestAJudgeAPolicyNamesIsFetchedWhenTheCatalogLacksIt(t *testing.T) { // The GPU judge alone: the CPU one is there, and one llmgw does not know // is not guessed at. if len(fs.started) != 1 || fs.started[0].Name != "qwen3guard-4b" || fs.started[0].GPUClass != "6000-24g" || - fs.started[0].Class != catalog.ClassStandard || fs.started[0].GuardFormat != "qwen3guard" || fs.started[0].CPU { + fs.started[0].Class != catalog.ClassStandard || fs.started[0].GuardFormat != "qwen3guard" || fs.started[0].CPU || + !fs.started[0].IfAbsent { t.Fatalf("started %+v", fs.started) } @@ -207,3 +208,30 @@ func TestAJudgeIsRefittedToTheCoresInitWouldGive(t *testing.T) { } } } + +// Just after the control plane starts, the judge init is adding is left to +// init: the keeper serves one itself only after its grace. +func TestTheKeeperLeavesAJustStartedInstallsJudgeToInit(t *testing.T) { + lib := guard.Config{Policies: map[string]guard.Policy{"b": {Judge: &guard.JudgePolicy{Model: "qwen3guard-0.6b", Input: "block"}}}} + now := time.Date(2026, 10, 19, 12, 0, 0, 0, time.UTC) + fs := &fakeServe{} + k := &judgeKeeper{ + library: func() guard.Config { return lib }, + models: func() []catalog.Model { return nil }, + serve: fs, grace: judgeGrace, + gpus: func() []adminapi.ClassUse { return nil }, + log: slog.New(slog.DiscardHandler), + now: func() time.Time { return now }, + } + k.check() + now = now.Add(judgeGrace / 2) + k.check() + if len(fs.started) != 0 { + t.Fatalf("served within its grace: %+v", fs.started) + } + now = now.Add(judgeGrace) + k.check() + if len(fs.started) != 1 { + t.Errorf("never served after its grace: %d", len(fs.started)) + } +} diff --git a/internal/cli/server.go b/internal/cli/server.go index 182b521..cef8cea 100644 --- a/internal/cli/server.go +++ b/internal/cli/server.go @@ -607,6 +607,7 @@ func newServerCmd() *cobra.Command { // A judge a policy names and the catalog lacks is // fetched and served, on the leader. keeper := &judgeKeeper{ + grace: judgeGrace, library: gate.Library, models: func() []catalog.Model { return admin.Catalog().Models() }, serve: serve, diff --git a/internal/control/catalog/catalog.go b/internal/control/catalog/catalog.go index 1fcd39d..528c838 100644 --- a/internal/control/catalog/catalog.go +++ b/internal/control/catalog/catalog.go @@ -6,6 +6,7 @@ import ( "crypto/sha256" "encoding/hex" "encoding/json" + "errors" "fmt" "maps" "net/url" @@ -23,6 +24,9 @@ import ( "gitea.mynodes.xyz/ilker/llmgw/internal/guard" ) +// ErrExists is a model added only if absent that was there already. +var ErrExists = errors.New("already in the catalog") + // Scaling is how many replicas a model should have. type Scaling struct { Min int `json:"min,omitempty" koanf:"min"` diff --git a/internal/control/modelserve/modelserve.go b/internal/control/modelserve/modelserve.go index 87fc2a8..eb773b8 100644 --- a/internal/control/modelserve/modelserve.go +++ b/internal/control/modelserve/modelserve.go @@ -163,6 +163,10 @@ type Request struct { // CPUCores is what a replica computes on; zero takes half the smallest // node it fits on. CPUCores int `json:"cpu_cores,omitempty"` + // IfAbsent adds the model only if nobody has meanwhile: the leader's own + // serving of a judge a policy names gives way to the one init or an + // operator added while it fetched, rather than replacing it. + IfAbsent bool `json:"-"` // MemoryGB is what a replica on CPUs reserves, in place of the figure // worked out from its weights and context: a judge's, whose own sizing // is known. @@ -691,7 +695,18 @@ func (s *Server) addAndStart(ctx context.Context, j *Job, repo hf.Repo, set hf.F } m.Class = req.Class s.step(j, "catalog", Running, "adding "+name) - if err := s.deps.Catalog.Add(ctx, m, req.By); err != nil { + add := s.deps.Catalog.Add + if adder, ok := s.deps.Catalog.(interface { + AddNew(ctx context.Context, m catalog.Model, by string) error + }); ok && req.IfAbsent { + add = adder.AddNew + } + if err := add(ctx, m, req.By); err != nil { + if req.IfAbsent && errors.Is(err, catalog.ErrExists) { + s.step(j, "catalog", Skipped, name+" was added meanwhile; left as it is") + s.finish(j) + return + } s.fail(j, "catalog", err) return } diff --git a/internal/control/modelserve/modelserve_test.go b/internal/control/modelserve/modelserve_test.go index 1bd9311..e38539c 100644 --- a/internal/control/modelserve/modelserve_test.go +++ b/internal/control/modelserve/modelserve_test.go @@ -452,3 +452,26 @@ func TestAJudgeIsServedWithTheMemoryAskedAndNoPromptCache(t *testing.T) { t.Error("a model that is not a judge lost its prompt cache") } } + +// takenCatalog has the model by the time a job comes to add it. +type takenCatalog struct{ fakeCatalog } + +func (c *takenCatalog) AddNew(context.Context, catalog.Model, string) error { + return fmt.Errorf("model %q: %w", "x", catalog.ErrExists) +} + +func TestAJobAddingOnlyIfAbsentGivesWayToAModelAddedMeanwhile(t *testing.T) { + files := []hf.File{{Path: "Qwen3Guard-Gen-0.6B-Q8_0.gguf", Bytes: 600 << 20, SHA256: "aa"}} + s, _, _, _ := setup(t, files, nil) + taken := &takenCatalog{} + s.deps.Catalog = taken + job, err := s.Start(Request{Repo: "example/Qwen3Guard-Gen-0.6B-GGUF", File: "Qwen3Guard-Gen-0.6B-Q8_0.gguf", + GuardFormat: "qwen3guard", CPU: true, CPUCores: 2, MemoryGB: 2, Replicas: 2, IfAbsent: true}) + if err != nil { + t.Fatal(err) + } + j := wait(t, s, job.ID) + if j.State != Done || len(taken.added) != 0 { + t.Errorf("job %s, added %d: %+v", j.State, len(taken.added), j.Steps) + } +} diff --git a/internal/control/modelsvc/modelsvc.go b/internal/control/modelsvc/modelsvc.go index c05e38d..e1bb55c 100644 --- a/internal/control/modelsvc/modelsvc.go +++ b/internal/control/modelsvc/modelsvc.go @@ -106,6 +106,17 @@ func (s *Service) Sources(ctx context.Context) (map[string][]catalog.Model, erro // than persisted and then skipped on every future start — which would be a // model that looks added and never appears. func (s *Service) Add(ctx context.Context, m catalog.Model, by string) error { + return s.add(ctx, m, by, false) +} + +// AddNew is Add for a model nobody has added meanwhile: one that is in the +// catalog by now is left as it is, and catalog.ErrExists said. The leader serving a +// judge a policy names gives way so to the judge init added while it fetched. +func (s *Service) AddNew(ctx context.Context, m catalog.Model, by string) error { + return s.add(ctx, m, by, true) +} + +func (s *Service) add(ctx context.Context, m catalog.Model, by string, onlyNew bool) error { s.mu.Lock() defer s.mu.Unlock() @@ -135,6 +146,9 @@ func (s *Service) Add(ctx context.Context, m catalog.Model, by string) error { next := s.base(stored) for _, e := range stored { if e.Name == m.Name { + if onlyNew { + return fmt.Errorf("model %q: %w", m.Name, catalog.ErrExists) + } continue // this call replaces it } next = append(next, e) diff --git a/internal/control/modelsvc/modelsvc_test.go b/internal/control/modelsvc/modelsvc_test.go index 9a1f626..810806e 100644 --- a/internal/control/modelsvc/modelsvc_test.go +++ b/internal/control/modelsvc/modelsvc_test.go @@ -3,6 +3,7 @@ package modelsvc import ( "context" "encoding/json" + "errors" "log/slog" "strings" "testing" @@ -377,3 +378,25 @@ func TestResourcesAreSetOnAStoredModelAndSurviveARestart(t *testing.T) { t.Error("the cores were lost on reload") } } + +// The leader serving a judge gives way to the one init added while it +// fetched: AddNew leaves a model already there as it is. +func TestAddNewLeavesAModelAddedMeanwhile(t *testing.T) { + s, _, rec := newService(t) + first := model("judge") + first.Description = "init's" + if err := s.Add(t.Context(), first, "init"); err != nil { + t.Fatal(err) + } + second := model("judge") + second.Description = "the leader's" + if err := s.AddNew(t.Context(), second, "llmgw"); !errors.Is(err, catalog.ErrExists) { + t.Fatalf("err %v", err) + } + if m, _ := rec.last.Resolve("judge"); m.Description != "init's" { + t.Errorf("replaced: %q", m.Description) + } + if err := s.AddNew(t.Context(), model("fresh"), "llmgw"); err != nil { + t.Errorf("a new model: %v", err) + } +}