diff --git a/core/backend/options.go b/core/backend/options.go index 4f7c81483d14..93a6eadc00a3 100644 --- a/core/backend/options.go +++ b/core/backend/options.go @@ -202,18 +202,18 @@ func ModelOptions(c config.ModelConfig, so *config.ApplicationConfig, opts ...mo model.WithContext(so.Context), model.WithModelID(c.ModelID()), } - // Prefer the revision stamped when the configuration was loaded. c has since - // been merged with this request's prediction parameters (temperature, top_p, - // stop, ...), and hashing it here would produce a different revision for - // every distinct request body — which the controller reads as a config - // change and rejects as stale. Recomputing is the fallback for a config that - // never passed through the loader. + // Use the revision stamped when the configuration was parsed, and only + // that. By this point c has been merged with the request's prediction + // parameters and had SetDefaults applied, so hashing it here would produce + // a revision that depends on the request body and on whether the model file + // parsed, which the controller reads as a config change and rejects. Every + // config the loader hands out is stamped; an unstamped one was synthesized + // elsewhere and is routed without a revision rather than with a wrong one. if revision := c.PersistedConfigRevision(); revision != "" { defOpts = append(defOpts, model.WithConfigRevision(revision)) - } else if revision, err := config.ModelConfigRevision(&c); err == nil { - defOpts = append(defOpts, model.WithConfigRevision(revision)) } else { - xlog.Warn("Failed to compute model configuration revision", "model", c.ModelID(), "error", err) + xlog.Warn("Model configuration carries no revision stamp; routing without one", + "model", c.ModelID()) } managedPrimary := len(c.Artifacts) > 0 && c.Artifacts[0].Resolved != nil if managedPrimary { diff --git a/core/config/model_config.go b/core/config/model_config.go index c6121eb8cf36..600519c7fc4c 100644 --- a/core/config/model_config.go +++ b/core/config/model_config.go @@ -1864,7 +1864,7 @@ func (c *ModelConfig) PersistedConfigRevision() string { // persisted. It is computed from the receiver as-is, so callers must invoke it // only on a configuration that has not been merged with request overrides. func (c *ModelConfig) StampPersistedConfigRevision() error { - revision, err := ModelConfigRevision(c) + revision, err := modelConfigRevision(c) if err != nil { return err } diff --git a/core/config/model_config_loader.go b/core/config/model_config_loader.go index 2a062c39dfcf..b91449ff01bc 100644 --- a/core/config/model_config_loader.go +++ b/core/config/model_config_loader.go @@ -965,3 +965,38 @@ func hasAnyMappingKey(mapping *yaml.Node, keys ...string) bool { func nonemptyScalar(node *yaml.Node) bool { return node != nil && node.Kind == yaml.ScalarNode && node.Tag == "!!str" && strings.TrimSpace(node.Value) != "" } + +// RevisionFor returns the config revision for modelName: the one an inference +// request for that model will carry. +// +// This is the only way to obtain a revision outside this package. Every +// publisher must use it, so that what is published and what is checked are +// the same value by construction rather than by two implementations happening +// to agree. Hashing a ModelConfig directly is not available to callers, because +// a config that has been through SetDefaults or the request middleware hashes +// to something no request will ever present. +func (bcl *ModelConfigLoader) RevisionFor(modelName string, appConfig *ApplicationConfig) (string, error) { + cfg, err := bcl.LoadModelConfigFileByNameDefaultOptions(modelName, appConfig) + if err != nil { + return "", fmt.Errorf("resolving config revision for %q: %w", modelName, err) + } + return stampedRevision(cfg, modelName) +} + +// RevisionForPath is RevisionFor for callers that hold loader options and a +// models path rather than an ApplicationConfig. +func (bcl *ModelConfigLoader) RevisionForPath(modelName, modelPath string, opts ...ConfigLoaderOption) (string, error) { + cfg, err := bcl.LoadModelConfigFileByName(modelName, modelPath, opts...) + if err != nil { + return "", fmt.Errorf("resolving config revision for %q: %w", modelName, err) + } + return stampedRevision(cfg, modelName) +} + +func stampedRevision(cfg *ModelConfig, modelName string) (string, error) { + revision := cfg.PersistedConfigRevision() + if revision == "" { + return "", fmt.Errorf("no config revision stamped for %q", modelName) + } + return revision, nil +} diff --git a/core/config/model_config_revision.go b/core/config/model_config_revision.go index e8dd1bee5d1f..8cbad2d6b16d 100644 --- a/core/config/model_config_revision.go +++ b/core/config/model_config_revision.go @@ -10,10 +10,17 @@ import ( "google.golang.org/protobuf/proto" ) -// ModelConfigRevision returns a stable revision of the persisted semantic +// modelConfigRevision returns a stable revision of the persisted semantic // configuration. ModelConfig's JSON tags exclude runtime-derived state and // source bookkeeping, while encoding/json orders map keys deterministically. -func ModelConfigRevision(cfg *ModelConfig) (string, error) { +// +// Deliberately unexported. It must only ever be called on a configuration as +// parsed from disk, before SetDefaults folds in the GGUF guess, the hardware +// defaults and app-level options. Callers outside this package cannot tell +// which they hold, and every time one hashed a defaulted or request-merged +// config it published a revision no inference request would carry, which makes +// the model unroutable. Use ModelConfigLoader.RevisionFor instead. +func modelConfigRevision(cfg *ModelConfig) (string, error) { if cfg == nil { return "", errors.New("model config is nil") } diff --git a/core/config/model_config_revision_stability_test.go b/core/config/model_config_revision_stability_test.go index b353f3d1440a..433b8e401549 100644 --- a/core/config/model_config_revision_stability_test.go +++ b/core/config/model_config_revision_stability_test.go @@ -51,9 +51,8 @@ template: Expect(loader.LoadModelConfigsFromPath(dir, appConfig.ToConfigLoaderOptions()...)).To(Succeed()) cfg, ok := loader.GetModelConfig("example") Expect(ok).To(BeTrue()) - revision, err := config.ModelConfigRevision(&cfg) - Expect(err).ToNot(HaveOccurred()) - return revision + Expect(cfg.PersistedConfigRevision()).ToNot(BeEmpty()) + return cfg.PersistedConfigRevision() } It("does not change when the same file is loaded repeatedly", func() { diff --git a/core/config/model_config_revision_test.go b/core/config/model_config_revision_test.go index 0b95d3549514..b7593727178e 100644 --- a/core/config/model_config_revision_test.go +++ b/core/config/model_config_revision_test.go @@ -19,10 +19,11 @@ var _ = Describe("Model configuration revisions", func() { return cfg } + // The raw hash is unexported on purpose, so these specs exercise it the way + // every caller now must: by stamping the parsed config. revision := func(cfg *config.ModelConfig) string { - value, err := config.ModelConfigRevision(cfg) - Expect(err).NotTo(HaveOccurred()) - return value + Expect(cfg.StampPersistedConfigRevision()).To(Succeed()) + return cfg.PersistedConfigRevision() } It("is stable across equivalent YAML formatting and map order", func() { diff --git a/core/http/endpoints/localai/edit_model_test.go b/core/http/endpoints/localai/edit_model_test.go index 17f7c4a7c8f3..223943e4619a 100644 --- a/core/http/endpoints/localai/edit_model_test.go +++ b/core/http/endpoints/localai/edit_model_test.go @@ -294,9 +294,9 @@ var _ = Describe("Edit Model test", func() { Expect(client.published[0]).To(Equal(messaging.CacheInvalidateEvent{ Element: "old", Op: "delete", ConfigRevision: modeladmin.DeletedModelConfigRevision("old"), })) - newConfig, ok := loader.GetModelConfig("new") + _, ok := loader.GetModelConfig("new") Expect(ok).To(BeTrue()) - newRevision, err := config.ModelConfigRevision(&newConfig) + newRevision, err := loader.RevisionForPath("new", tempDir) Expect(err).ToNot(HaveOccurred()) Expect(client.published[1]).To(Equal(messaging.CacheInvalidateEvent{ Element: "new", Op: "install", ConfigRevision: newRevision, @@ -313,9 +313,9 @@ var _ = Describe("Edit Model test", func() { } _, oldOnPeer := peerLoader.GetModelConfig("old") Expect(oldOnPeer).To(BeFalse()) - peerConfig, newOnPeer := peerLoader.GetModelConfig("new") + _, newOnPeer := peerLoader.GetModelConfig("new") Expect(newOnPeer).To(BeTrue()) - peerRevision, err := config.ModelConfigRevision(&peerConfig) + peerRevision, err := peerLoader.RevisionForPath("new", tempDir) Expect(err).ToNot(HaveOccurred()) Expect(peerRevision).To(Equal(newRevision)) Expect(peerLifecycle.batches).To(Equal([][]modeladmin.ModelRevisionTransition{ diff --git a/core/services/modeladmin/config.go b/core/services/modeladmin/config.go index 515d014f7b53..2cadfcaedcc5 100644 --- a/core/services/modeladmin/config.go +++ b/core/services/modeladmin/config.go @@ -185,13 +185,9 @@ func (s *ConfigService) patchConfig(ctx context.Context, name string, patch map[ // because SetDefaults runs again on the request path and is not // idempotent for every model, and the edit would leave the model // unroutable. - resolved, err := s.Loader.LoadModelConfigFileByNameDefaultOptions(updated.Name, s.AppConfig) + revision, err := s.Loader.RevisionFor(updated.Name, s.AppConfig) if err != nil { - return fmt.Errorf("resolve config revision: %w", err) - } - revision := resolved.PersistedConfigRevision() - if revision == "" { - return fmt.Errorf("no config revision stamped for %q", updated.Name) + return err } _ = s.Loader.Preload(s.modelsPath()) pending, err := s.applyRevision(ctx, name, updated.Name, revision, updated.IsDisabled()) @@ -351,13 +347,12 @@ func (s *ConfigService) editYAML(ctx context.Context, name string, body []byte) if err := s.Loader.LoadModelConfigsFromPath(modelsPath, s.AppConfig.ToConfigLoaderOptions()...); err != nil { return fmt.Errorf("reload configs: %w", err) } - loaded, ok := s.Loader.GetModelConfig(req.Name) - if !ok { + if _, ok := s.Loader.GetModelConfig(req.Name); !ok { return fmt.Errorf("reload configs: model %q missing", req.Name) } - revision, err := config.ModelConfigRevision(&loaded) + revision, err := s.Loader.RevisionFor(req.Name, s.AppConfig) if err != nil { - return fmt.Errorf("compute config revision: %w", err) + return err } if err := s.Loader.Preload(modelsPath); err != nil { return fmt.Errorf("preload after edit: %w", err) diff --git a/core/services/modeladmin/remote_sync.go b/core/services/modeladmin/remote_sync.go index 844239a8829d..9eec4fa5e710 100644 --- a/core/services/modeladmin/remote_sync.go +++ b/core/services/modeladmin/remote_sync.go @@ -49,9 +49,9 @@ func applyRemoteChange(ctx context.Context, cl *config.ModelConfigLoader, models disabled := true if exists { var err error - revision, err = config.ModelConfigRevision(&cfg) + revision, err = authoritative.RevisionForPath(name, modelsPath, opts...) if err != nil { - return fmt.Errorf("compute authoritative model config revision for %q: %w", name, err) + return fmt.Errorf("resolve authoritative model config revision for %q: %w", name, err) } disabled = cfg.IsDisabled() } @@ -83,15 +83,9 @@ func changedConfigNames(current, snapshot map[string]config.ModelConfig, named s changed[name] = struct{}{} continue } - previousRevision, err := config.ModelConfigRevision(&previous) - if err != nil { - return nil, fmt.Errorf("compute current model config revision for %q: %w", name, err) - } - revision, err := config.ModelConfigRevision(&cfg) - if err != nil { - return nil, fmt.Errorf("compute authoritative model config revision for %q: %w", name, err) - } - if previousRevision != revision { + // Both sides come from a loader, so both carry the revision stamped + // when their file was parsed. Comparing the stamps compares the files. + if previous.PersistedConfigRevision() != cfg.PersistedConfigRevision() { changed[name] = struct{}{} } } diff --git a/core/services/modeladmin/remote_sync_test.go b/core/services/modeladmin/remote_sync_test.go index 32289429fa3a..d5a2b4664f2f 100644 --- a/core/services/modeladmin/remote_sync_test.go +++ b/core/services/modeladmin/remote_sync_test.go @@ -58,9 +58,9 @@ var _ = Describe("ApplyRemoteChange", func() { Expect(ApplyRemoteChange(context.Background(), loader, dir, evt, lifecycle)).To(Succeed()) Expect(ApplyRemoteChange(context.Background(), loader, dir, evt, lifecycle)).To(Succeed()) Expect(lifecycle.calls).To(HaveLen(2)) - loaded, ok := loader.GetModelConfig("peer-alias") + _, ok := loader.GetModelConfig("peer-alias") Expect(ok).To(BeTrue()) - revision, err := config.ModelConfigRevision(&loaded) + revision, err := loader.RevisionForPath("peer-alias", dir) Expect(err).ToNot(HaveOccurred()) Expect(lifecycle.calls[0].revision).To(Equal(revision)) Expect(lifecycle.calls[1].revision).To(Equal(revision)) @@ -84,7 +84,7 @@ var _ = Describe("ApplyRemoteChange", func() { loaded, ok := loader.GetModelConfig("peer-alias") Expect(ok).To(BeTrue()) Expect(loaded.ContextSize).To(HaveValue(Equal(10000))) - revision, err := config.ModelConfigRevision(&loaded) + revision, err := loader.RevisionForPath(loaded.Name, dir) Expect(err).ToNot(HaveOccurred()) Expect(lifecycle.calls).To(HaveLen(3)) Expect(lifecycle.calls[1].revision).To(Equal(revision)) @@ -100,7 +100,7 @@ var _ = Describe("ApplyRemoteChange", func() { loaded, ok := loader.GetModelConfig("reinstalled") Expect(ok).To(BeTrue()) - revision, err := config.ModelConfigRevision(&loaded) + revision, err := loader.RevisionForPath(loaded.Name, dir) Expect(err).ToNot(HaveOccurred()) Expect(lifecycle.calls).To(HaveLen(1)) Expect(lifecycle.calls[0].revision).To(Equal(revision)) @@ -172,7 +172,7 @@ var _ = Describe("ApplyRemoteChange", func() { Expect(loaded.ContextSize).To(HaveValue(Equal(10000))) _, ok = loader.GetModelConfig("deleted") Expect(ok).To(BeFalse()) - changedRevision, err := config.ModelConfigRevision(&loaded) + changedRevision, err := loader.RevisionForPath(loaded.Name, dir) Expect(err).ToNot(HaveOccurred()) Expect(lifecycle.calls).To(ConsistOf( revisionLifecycleCall{oldName: "changed", newName: "changed", revision: changedRevision}, @@ -228,7 +228,7 @@ var _ = Describe("ApplyRemoteChange", func() { loaded, ok := loader.GetModelConfig("ordered") Expect(ok).To(BeTrue()) Expect(loaded.ContextSize).To(HaveValue(Equal(10000))) - revision, err := config.ModelConfigRevision(&loaded) + revision, err := loader.RevisionForPath(loaded.Name, dir) Expect(err).ToNot(HaveOccurred()) Expect(lifecycle.revisions()).To(HaveLen(2)) Expect(lifecycle.revisions()[1]).To(Equal(revision)) @@ -263,7 +263,7 @@ var _ = Describe("ApplyRemoteChange", func() { Expect(ok).To(BeTrue()) Expect(loaded.ContextSize).To(HaveValue(Equal(10000))) Expect(readMap(filepath.Join(dir, "ordered.yaml"))).To(HaveKeyWithValue("context_size", 10000)) - revision, err := config.ModelConfigRevision(&loaded) + revision, err := loader.RevisionForPath(loaded.Name, dir) Expect(err).ToNot(HaveOccurred()) Expect(lifecycle.revisions()).To(HaveLen(2)) Expect(lifecycle.revisions()[1]).To(Equal(revision)) diff --git a/core/services/modeladmin/revision_agreement_test.go b/core/services/modeladmin/revision_agreement_test.go new file mode 100644 index 000000000000..fcc9616b9a90 --- /dev/null +++ b/core/services/modeladmin/revision_agreement_test.go @@ -0,0 +1,102 @@ +package modeladmin + +import ( + "os" + "path/filepath" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + + "github.com/mudler/LocalAI/core/config" + "github.com/mudler/LocalAI/pkg/system" +) + +// A model's revision is published by administration and checked against on +// every inference request. Those were computed by different code, and each time +// they drifted the model became unroutable until someone deleted the row by +// hand: the request path resolves through the loader, while publishers hashed +// whatever ModelConfig they were holding, which by then had SetDefaults applied. +// +// There is now one resolver, ModelConfigLoader.RevisionFor, and the raw hash is +// unexported so a new publisher cannot reintroduce the split. This pins the +// property that mattered: whatever a publisher writes is what a request brings. +var _ = Describe("Published and requested revisions agree", func() { + var ( + dir string + appConfig *config.ApplicationConfig + loader *config.ModelConfigLoader + ) + + // Several shapes, because the divergence only ever showed up on configs + // rich enough for SetDefaults to change something: a model file to guess + // from, several derived usecases, explicit options. + models := map[string]string{ + "plain": "name: plain\nbackend: llama-cpp\nparameters:\n model: plain.gguf\n", + "multimodal": "name: multimodal\nbackend: llama-cpp\ncontext_size: 50000\nknown_usecases:\n - chat\nmmproj: mm/mmproj.gguf\noptions:\n - use_jinja:true\n - parallel:2\nparameters:\n model: mm/model.gguf\n", + "auto-ctx": "name: auto-ctx\nbackend: llama-cpp\ncontext_size: -1\nparameters:\n model: auto.gguf\n", + "no-backend": "name: no-backend\nparameters:\n model: bare.gguf\n", + "with-thread": "name: with-thread\nbackend: llama-cpp\nthreads: 3\nparameters:\n model: t.gguf\n", + } + + BeforeEach(func() { + dir = GinkgoT().TempDir() + for name, body := range models { + Expect(os.WriteFile(filepath.Join(dir, name+".yaml"), []byte(body), 0o600)).To(Succeed()) + } + appConfig = config.NewApplicationConfig() + appConfig.SystemState = &system.SystemState{Model: system.Model{ModelsPath: dir}} + appConfig.Threads = 8 + loader = config.NewModelConfigLoader(dir) + Expect(loader.LoadModelConfigsFromPath(dir, appConfig.ToConfigLoaderOptions()...)).To(Succeed()) + }) + + // requestRevision mirrors what core/backend.ModelOptions forwards to the + // router: the stamp on the config the request pipeline resolved. + requestRevision := func(name string) string { + cfg, err := loader.LoadModelConfigFileByNameDefaultOptions(name, appConfig) + Expect(err).ToNot(HaveOccurred()) + return cfg.PersistedConfigRevision() + } + + It("resolves the same revision a request will carry, for every model shape", func() { + for name := range models { + published, err := loader.RevisionFor(name, appConfig) + Expect(err).ToNot(HaveOccurred(), "model %s", name) + Expect(published).To(Equal(requestRevision(name)), "model %s: publisher and request disagree", name) + } + }) + + It("resolves the same revision through the path-based form", func() { + for name := range models { + byAppConfig, err := loader.RevisionFor(name, appConfig) + Expect(err).ToNot(HaveOccurred()) + byPath, err := loader.RevisionForPath(name, dir, appConfig.ToConfigLoaderOptions()...) + Expect(err).ToNot(HaveOccurred()) + Expect(byPath).To(Equal(byAppConfig), "model %s", name) + } + }) + + It("does not move when the app-level defaults change", func() { + before := map[string]string{} + for name := range models { + r, err := loader.RevisionFor(name, appConfig) + Expect(err).ToNot(HaveOccurred()) + before[name] = r + } + + other := config.NewApplicationConfig() + other.SystemState = &system.SystemState{Model: system.Model{ModelsPath: dir}} + other.Threads = 1 + other.F16 = true + other.ContextSize = 4096 + fresh := config.NewModelConfigLoader(dir) + Expect(fresh.LoadModelConfigsFromPath(dir, other.ToConfigLoaderOptions()...)).To(Succeed()) + + for name := range models { + r, err := fresh.RevisionFor(name, other) + Expect(err).ToNot(HaveOccurred()) + Expect(r).To(Equal(before[name]), + "model %s: changing an app-level setting must not make every model unroutable", name) + } + }) +}) diff --git a/core/services/modeladmin/revision_resync.go b/core/services/modeladmin/revision_resync.go index 5d84831a2ab1..7d92f52a4b3b 100644 --- a/core/services/modeladmin/revision_resync.go +++ b/core/services/modeladmin/revision_resync.go @@ -87,13 +87,9 @@ func ResyncModelConfigRevisions(ctx context.Context, loader *config.ModelConfigL // (it re-runs the GGUF guess and hardware defaults), so hashing the // stored config yields a value no request will ever carry, and // publishing it would wedge the model this resync exists to unwedge. - resolved, err := loader.LoadModelConfigFileByNameDefaultOptions(cfg.Name, appConfig) + want, err := loader.RevisionFor(cfg.Name, appConfig) if err != nil { - return fmt.Errorf("resolve config for %q: %w", cfg.Name, err) - } - want := resolved.PersistedConfigRevision() - if want == "" { - return fmt.Errorf("no config revision stamped for %q", cfg.Name) + return err } stored, err := store.GetModelConfigRevision(ctx, cfg.Name) diff --git a/core/services/modeladmin/state.go b/core/services/modeladmin/state.go index 4f84b5859e1b..d37368d9d8c9 100644 --- a/core/services/modeladmin/state.go +++ b/core/services/modeladmin/state.go @@ -68,13 +68,9 @@ func (s *ConfigService) toggleState(ctx context.Context, name string, action Act // because SetDefaults runs again on the request path and is not // idempotent for every model, and the edit would leave the model // unroutable. - resolved, err := s.Loader.LoadModelConfigFileByNameDefaultOptions(name, s.AppConfig) + revision, err := s.Loader.RevisionFor(name, s.AppConfig) if err != nil { - return fmt.Errorf("resolve config revision: %w", err) - } - revision := resolved.PersistedConfigRevision() - if revision == "" { - return fmt.Errorf("no config revision stamped for %q", name) + return err } pending, err := s.applyRevision(ctx, name, name, revision, action == ActionDisable) if err != nil { diff --git a/core/services/nodes/router_revision_lifecycle_test.go b/core/services/nodes/router_revision_lifecycle_test.go index b9b78f790436..7b1de6144b72 100644 --- a/core/services/nodes/router_revision_lifecycle_test.go +++ b/core/services/nodes/router_revision_lifecycle_test.go @@ -249,8 +249,8 @@ var _ = Describe("revision-bound load publication", func() { LLMConfig: config.LLMConfig{ContextSize: &contextSize}, } cfg.Model = "models/full-flow.gguf" - expectedRevision, err := config.ModelConfigRevision(&cfg) - Expect(err).NotTo(HaveOccurred()) + Expect(cfg.StampPersistedConfigRevision()).To(Succeed()) + expectedRevision := cfg.PersistedConfigRevision() router := NewSmartRouter(registry, SmartRouterOptions{ Unloader: unloader, diff --git a/core/services/nodes/unloader.go b/core/services/nodes/unloader.go index 3b1cd15b8955..460be8acf613 100644 --- a/core/services/nodes/unloader.go +++ b/core/services/nodes/unloader.go @@ -372,17 +372,36 @@ func (a *RemoteUnloaderAdapter) ListBackends(nodeID string) (*messaging.BackendL // scheduler could pick a node that could not be given work and the request // failed with "no responders available". // -// It reuses the models.running subject rather than a dedicated ping subject on -// purpose: a new subject would go unanswered by any worker that has not been -// upgraded yet, and this check would then report every one of them as dead. -// The worker answers out of its in-memory process table, so a live node -// replies immediately, and NATS reports no-responders without waiting out the -// timeout. +// The subject asked has to be one every worker in the fleet subscribes to, or +// this check condemns the workers that do not. models.running was the obvious +// choice and the wrong one: it arrived in 4.6, so a 4.5 worker that is alive +// and serving never answers it, and a model pinned to that node could never be +// scheduled. backend.list has been part of the worker protocol far longer, so +// it is the safer question to ask. +// +// A worker that answers anything is alive. Only when every subject reports no +// responders is the node treated as absent, so adding a newer subject here can +// never condemn an older worker. func (a *RemoteUnloaderAdapter) PingNode(nodeID string) error { - subject := messaging.SubjectNodeModelsRunning(nodeID) - _, err := messaging.RequestJSON[messaging.ModelsRunningRequest, messaging.ModelsRunningReply]( - a.nats, subject, messaging.ModelsRunningRequest{}, 5*time.Second) - return err + subjects := []string{ + messaging.SubjectNodeBackendList(nodeID), + messaging.SubjectNodeModelsRunning(nodeID), + } + var lastErr error + for _, subject := range subjects { + _, err := messaging.RequestJSON[messaging.BackendListRequest, messaging.BackendListReply]( + a.nats, subject, messaging.BackendListRequest{}, 5*time.Second) + if err == nil { + return nil + } + if !errors.Is(err, nats.ErrNoResponders) { + // Reached someone, or failed for a reason that is not absence. + // Either way the node is not proven gone. + return nil + } + lastErr = err + } + return lastErr } // ListRunningModels asks a worker node which model backend processes it diff --git a/core/services/nodes/unloader_ping_test.go b/core/services/nodes/unloader_ping_test.go new file mode 100644 index 000000000000..a9b3a5889469 --- /dev/null +++ b/core/services/nodes/unloader_ping_test.go @@ -0,0 +1,51 @@ +package nodes + +import ( + "errors" + "time" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + "github.com/nats-io/nats.go" + + "github.com/mudler/LocalAI/core/services/messaging" +) + +// The scheduler's liveness probe asks a worker a question over NATS and treats +// "no responders" as proof the worker is gone. That is only sound if every +// worker in the fleet subscribes to the subject asked. +// +// It originally asked models.running, which arrived in 4.6. A 4.5 worker is +// perfectly alive and serving, answers backend.list, and never subscribes to +// models.running, so the probe condemned it on every scheduling attempt. A +// model pinned to such a node could then never be placed at all. +var _ = Describe("Node liveness probe subject", func() { + var ( + mc *scriptedMessagingClient + adapter *RemoteUnloaderAdapter + ) + + const nodeID = "11111111-2222-3333-4444-555555555555" + + BeforeEach(func() { + mc = newScriptedMessagingClient() + adapter = NewRemoteUnloaderAdapter(nil, mc, 3*time.Minute, 15*time.Minute) + }) + + It("treats a worker that answers backend.list as alive", func() { + // A worker old enough to predate models.running: it answers the + // long-standing backend.list subject and nothing else. + mc.scriptReply(messaging.SubjectNodeBackendList(nodeID), messaging.BackendListReply{}) + mc.scriptNoResponders(messaging.SubjectNodeModelsRunning(nodeID)) + + Expect(errors.Is(adapter.PingNode(nodeID), nats.ErrNoResponders)).To(BeFalse(), + "a worker answering backend.list is alive regardless of newer subjects") + }) + + It("still reports a worker that answers nothing as absent", func() { + mc.scriptNoResponders(messaging.SubjectNodeBackendList(nodeID)) + mc.scriptNoResponders(messaging.SubjectNodeModelsRunning(nodeID)) + + Expect(errors.Is(adapter.PingNode(nodeID), nats.ErrNoResponders)).To(BeTrue()) + }) +}) diff --git a/gallery/index.yaml b/gallery/index.yaml index c980a79fdbba..1c67d799bdbb 100644 --- a/gallery/index.yaml +++ b/gallery/index.yaml @@ -558,10 +558,10 @@ files: - filename: llama-cpp/models/ornith-1.5-9b/Ornith-1.5-9B-Q4_K_M.gguf uri: huggingface://ornith-ai/Ornith-1.5-9B-GGUF/Ornith-1.5-9B-Q4_K_M.gguf - sha256: 7d791afcb31812acc88cd5aafc675391df28c6fc3d8eae002bb4e6cc3d8cfd8d + sha256: 70c112196e0b7023803c9762752e46d29e612a92c83f995bc3ba1ceb07e8fab6 - filename: llama-cpp/mmproj/ornith-1.5-9b/mmproj-BF16.gguf uri: huggingface://ornith-ai/Ornith-1.5-9B-GGUF/mmproj-Ornith-1.5-9B-BF16.gguf - sha256: d65001a94c4b6852bc7a0e7c5cc92fe8506755bb270e54483fd5feec7ae39a19 + sha256: 626f9f90627402a6bf4a999111d0fbd69b5fcca7aa8ba089d69e5f10e8858e1d - !!merge <<: *ornith-1-5-9b name: "ornith-1.5-9b-q8" variants: [] @@ -593,10 +593,10 @@ files: - filename: llama-cpp/models/ornith-1.5-9b/Ornith-1.5-9B-Q8_0.gguf uri: huggingface://ornith-ai/Ornith-1.5-9B-GGUF/Ornith-1.5-9B-Q8_0.gguf - sha256: 6874eeb25c71081dc8f0bbe88f3ebb786312447132745371cd980bce95d259b9 + sha256: 22086870b009dbe9815ee752c48a82de930118a7c5ce5599590892ae03b8b010 - filename: llama-cpp/mmproj/ornith-1.5-9b/mmproj-BF16.gguf uri: huggingface://ornith-ai/Ornith-1.5-9B-GGUF/mmproj-Ornith-1.5-9B-BF16.gguf - sha256: d65001a94c4b6852bc7a0e7c5cc92fe8506755bb270e54483fd5feec7ae39a19 + sha256: 626f9f90627402a6bf4a999111d0fbd69b5fcca7aa8ba089d69e5f10e8858e1d - &qwen3-8-27b-obliterated name: "qwen3.8-27b-obliterated-q4" variants: @@ -654,7 +654,7 @@ files: - filename: llama-cpp/models/qwen3.8-27b-obliterated/Qwen3.8-27B-OBLITERATED-Q4_K_M.gguf uri: huggingface://OBLITERATUS/Qwen3.8-27B-OBLITERATED/Qwen3.8-27B-OBLITERATED-Q4_K_M.gguf - sha256: c5e4fe705883e244a468c9e445c8d6ba37fd310b0113e25d2b8a7f2d6f1243e8 + sha256: 1f74330b211a8253c96f1bf586cba6eb56d37117c97ed9e6eec18c198a4e7fe5 - filename: llama-cpp/mmproj/qwen3.8-27b-obliterated/mmproj-model-bf16.gguf uri: huggingface://OBLITERATUS/Qwen3.8-27B-OBLITERATED/mmproj-model-bf16.gguf sha256: e484e3b7e907ed0e0644c0de56c3f5929c7ad5c9c6cc84d35a9d8dc08d461545 @@ -687,7 +687,7 @@ files: - filename: llama-cpp/models/qwen3.8-27b-obliterated/Qwen3.8-27B-OBLITERATED-Q8_0.gguf uri: huggingface://OBLITERATUS/Qwen3.8-27B-OBLITERATED/Qwen3.8-27B-OBLITERATED-Q8_0.gguf - sha256: 4ed72a101dfa7f8fd642598368c4d334f1334cedc5254b71a06b5c4a542c59fc + sha256: afa839b2fa5bc890e5735031dda2c6239d3b6bba3b6ffa29477cbc14a2e1f221 - filename: llama-cpp/mmproj/qwen3.8-27b-obliterated/mmproj-model-bf16.gguf uri: huggingface://OBLITERATUS/Qwen3.8-27B-OBLITERATED/mmproj-model-bf16.gguf sha256: e484e3b7e907ed0e0644c0de56c3f5929c7ad5c9c6cc84d35a9d8dc08d461545