[pull] master from mudler:master

backend/cpp/ds4/Makefile

@@ -1,10 +1,10 @@
 # ds4 backend Makefile.
 #
-# Upstream pin lives below as DS4_VERSION?=84cc882352757baf628a1776badf7cc54d584e28
+# Upstream pin lives below as DS4_VERSION?=c1d4597a80e300b803dc642519718f2c999589da
 # (.github/bump_deps.sh) can find and update it - matches the
 # llama-cpp / ik-llama-cpp / turboquant convention.
 
-DS4_VERSION?=84cc882352757baf628a1776badf7cc54d584e28
+DS4_VERSION?=c1d4597a80e300b803dc642519718f2c999589da
 DS4_REPO?=https://github.com/antirez/ds4
 
 CURRENT_MAKEFILE_DIR := $(dir $(abspath $(lastword $(MAKEFILE_LIST))))

core/application/startup.go

@@ -373,6 +373,16 @@ func New(opts ...config.AppOption) (*Application, error) {
 			cfgLoaderOpts := options.ToConfigLoaderOptions()
 			modelRevisionLifecycle := modeladmin.NewDistributedModelRevisionLifecycle(distSvc.Registry, distSvc.ModelCleanup)
 			gs.SetModelRevisionLifecycle(modelRevisionLifecycle)
+			// Bring the controller's stored revisions back in line with the
+			// configuration on disk. An inference request may only establish a
+			// revision, never replace one, so a model whose stored value had
+			// drifted stayed unroutable until someone deleted the row.
+			if err := modeladmin.ResyncModelConfigRevisions(options.Context,
+				application.ModelConfigLoader(),
+				modeladmin.NewRevisionStore(distSvc.Registry, modelRevisionLifecycle),
+			); err != nil {
+				xlog.Warn("Failed to resync model config revisions", "error", err)
+			}
 			gs.OnModelsChanged = func(evt messaging.CacheInvalidateEvent) {
 				// ApplyRemoteChange honors the op: a "delete" prunes the element
 				// (a reload-from-path is additive and cannot drop it), anything

core/services/modeladmin/revision_resync.go

@@ -0,0 +1,118 @@
+package modeladmin
+
+import (
+	"context"
+	"errors"
+	"fmt"
+
+	"github.com/mudler/xlog"
+	"gorm.io/gorm"
+
+	"github.com/mudler/LocalAI/core/config"
+)
+
+// ErrNoStoredRevision reports that the controller holds no revision for a
+// model, which is the normal state for one that has never been served.
+var ErrNoStoredRevision = gorm.ErrRecordNotFound
+
+// RevisionStore is the controller state this resync reads and corrects.
+type RevisionStore interface {
+	GetModelConfigRevision(ctx context.Context, modelName string) (string, error)
+	ApplyConfigRevisions(ctx context.Context, transitions []ModelRevisionTransition) (int, error)
+}
+
+// RevisionReader is the read half, satisfied by the node registry.
+type RevisionReader interface {
+	GetModelConfigRevision(ctx context.Context, modelName string) (string, error)
+}
+
+type revisionStore struct {
+	RevisionReader
+	lifecycle ModelRevisionLifecycle
+}
+
+func (s revisionStore) ApplyConfigRevisions(ctx context.Context, t []ModelRevisionTransition) (int, error) {
+	return s.lifecycle.ApplyConfigRevisions(ctx, t)
+}
+
+// NewRevisionStore pairs the registry that holds the stored revisions with the
+// lifecycle that publishes new ones. Returns nil when either half is missing,
+// which ResyncModelConfigRevisions treats as "nothing to reconcile".
+func NewRevisionStore(reader RevisionReader, lifecycle ModelRevisionLifecycle) RevisionStore {
+	if reader == nil || lifecycle == nil {
+		return nil
+	}
+	return revisionStore{RevisionReader: reader, lifecycle: lifecycle}
+}
+
+// ResyncModelConfigRevisions makes the controller's stored revision for each
+// model agree with what this build computes from the configuration on disk.
+//
+// The stored revision is what every inference request is checked against, but
+// nothing ever re-derived it from the persisted configuration: it moved only on
+// an edit, a gallery install, or a peer's change broadcast. Any other way for
+// the two to diverge left the model permanently unroutable, because an
+// inference request may only establish a revision, never replace one. A
+// configuration edited while this frontend was down, or a change in what the
+// revision is computed over, both landed there, and the only recovery was
+// deleting the row by hand.
+//
+// Running this at startup makes that self-correcting. Only a model whose stored
+// revision disagrees is republished, so replicas of models that did not drift
+// keep serving: republishing is not free, it quarantines every replica loaded
+// under the old revision.
+//
+// A model with no stored revision is left alone. It has never been served, and
+// inventing controller state for it here would quarantine nothing and describe
+// a model that may never be requested.
+func ResyncModelConfigRevisions(ctx context.Context, loader *config.ModelConfigLoader, store RevisionStore) error {
+	if loader == nil || store == nil {
+		return nil
+	}
+
+	var transitions []ModelRevisionTransition
+	for _, cfg := range loader.GetAllModelsConfigs() {
+		want, err := config.ModelConfigRevision(&cfg)
+		if err != nil {
+			return fmt.Errorf("compute config revision for %q: %w", cfg.Name, err)
+		}
+
+		stored, err := store.GetModelConfigRevision(ctx, cfg.Name)
+		if errors.Is(err, ErrNoStoredRevision) {
+			continue
+		}
+		if err != nil {
+			return fmt.Errorf("read stored config revision for %q: %w", cfg.Name, err)
+		}
+		if stored == want {
+			continue
+		}
+
+		xlog.Warn("Stored model config revision disagrees with the configuration on disk, republishing",
+			"model", cfg.Name, "stored", shortRevision(stored), "computed", shortRevision(want))
+		transitions = append(transitions, ModelRevisionTransition{
+			ModelName: cfg.Name, ConfigRevision: want, Disabled: cfg.IsDisabled(),
+		})
+	}
+
+	if len(transitions) == 0 {
+		return nil
+	}
+	if _, err := store.ApplyConfigRevisions(ctx, transitions); err != nil {
+		return fmt.Errorf("republish model config revisions: %w", err)
+	}
+	xlog.Info("Republished model config revisions to match the configuration on disk", "models", len(transitions))
+	return nil
+}
+
+// shortRevision trims a revision for log output; the leading bytes identify it
+// well enough to tell two apart.
+func shortRevision(revision string) string {
+	if revision == "" {
+		return "(none)"
+	}
+	if len(revision) > 12 {
+		return revision[:12]
+	}
+	return revision
+}

core/services/modeladmin/revision_resync_test.go

@@ -0,0 +1,142 @@
+package modeladmin
+
+import (
+	"context"
+	"errors"
+	"os"
+	"path/filepath"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+
+	"github.com/mudler/LocalAI/core/config"
+	"github.com/mudler/LocalAI/pkg/system"
+)
+
+// stubRevisionStore stands in for the controller's stored revisions.
+type stubRevisionStore struct {
+	stored  map[string]string
+	getErr  error
+	applied []ModelRevisionTransition
+	applyEr error
+}
+
+func (s *stubRevisionStore) GetModelConfigRevision(_ context.Context, name string) (string, error) {
+	if s.getErr != nil {
+		return "", s.getErr
+	}
+	rev, ok := s.stored[name]
+	if !ok {
+		return "", ErrNoStoredRevision
+	}
+	return rev, nil
+}
+
+func (s *stubRevisionStore) ApplyConfigRevisions(_ context.Context, t []ModelRevisionTransition) (int, error) {
+	s.applied = append(s.applied, t...)
+	return 0, s.applyEr
+}
+
+// The controller pins a model's replicas to a stored revision and rejects any
+// request carrying a different one. Nothing ever re-derived that stored value
+// from the configuration on disk: it only moved on an edit, a gallery install
+// or a peer's change event. So whenever the stored value stopped matching what
+// this build computes for an unchanged file, every request for that model was
+// rejected until an operator deleted the row by hand.
+var _ = Describe("ResyncModelConfigRevisions", func() {
+	var (
+		dir       string
+		loader    *config.ModelConfigLoader
+		store     *stubRevisionStore
+		appConfig *config.ApplicationConfig
+	)
+
+	write := func(name, body string) {
+		Expect(os.WriteFile(filepath.Join(dir, name+".yaml"), []byte(body), 0o600)).To(Succeed())
+	}
+
+	revisionOf := func(name string) string {
+		cfg, ok := loader.GetModelConfig(name)
+		Expect(ok).To(BeTrue())
+		rev, err := config.ModelConfigRevision(&cfg)
+		Expect(err).ToNot(HaveOccurred())
+		return rev
+	}
+
+	BeforeEach(func() {
+		dir = GinkgoT().TempDir()
+		appConfig = config.NewApplicationConfig()
+		appConfig.SystemState = &system.SystemState{Model: system.Model{ModelsPath: dir}}
+		loader = config.NewModelConfigLoader(dir)
+		store = &stubRevisionStore{stored: map[string]string{}}
+	})
+
+	load := func() {
+		Expect(loader.LoadModelConfigsFromPath(dir, appConfig.ToConfigLoaderOptions()...)).To(Succeed())
+	}
+
+	It("republishes the revision when the stored one no longer matches the config on disk", func() {
+		write("drifted", "name: drifted\nbackend: llama-cpp\ncontext_size: 4096\n")
+		load()
+		store.stored["drifted"] = "a-revision-from-an-earlier-build"
+
+		Expect(ResyncModelConfigRevisions(context.Background(), loader, store)).To(Succeed())
+
+		Expect(store.applied).To(HaveLen(1))
+		Expect(store.applied[0].ModelName).To(Equal("drifted"))
+		Expect(store.applied[0].ConfigRevision).To(Equal(revisionOf("drifted")))
+	})
+
+	It("leaves a model alone when the stored revision already matches", func() {
+		write("agreed", "name: agreed\nbackend: llama-cpp\n")
+		load()
+		store.stored["agreed"] = revisionOf("agreed")
+
+		Expect(ResyncModelConfigRevisions(context.Background(), loader, store)).To(Succeed())
+
+		Expect(store.applied).To(BeEmpty(), "republishing an unchanged revision would quarantine live replicas for nothing")
+	})
+
+	// A model nobody has served has no stored revision. Creating one here would
+	// invent controller state for a model that may never be requested; the first
+	// request establishes it.
+	It("does not create state for a model that has never been served", func() {
+		write("never-served", "name: never-served\nbackend: llama-cpp\n")
+		load()
+
+		Expect(ResyncModelConfigRevisions(context.Background(), loader, store)).To(Succeed())
+
+		Expect(store.applied).To(BeEmpty())
+	})
+
+	It("republishes only the models that actually drifted", func() {
+		write("drifted", "name: drifted\nbackend: llama-cpp\n")
+		write("agreed", "name: agreed\nbackend: llama-cpp\ncontext_size: 2048\n")
+		load()
+		store.stored["drifted"] = "stale"
+		store.stored["agreed"] = revisionOf("agreed")
+
+		Expect(ResyncModelConfigRevisions(context.Background(), loader, store)).To(Succeed())
+
+		Expect(store.applied).To(HaveLen(1))
+		Expect(store.applied[0].ModelName).To(Equal("drifted"))
+	})
+
+	It("reports a store failure instead of continuing silently", func() {
+		write("drifted", "name: drifted\nbackend: llama-cpp\n")
+		load()
+		store.stored["drifted"] = "stale"
+		store.applyEr = errors.New("database is down")
+
+		Expect(ResyncModelConfigRevisions(context.Background(), loader, store)).ToNot(Succeed())
+	})
+
+	It("skips a model whose stored revision cannot be read rather than guessing", func() {
+		write("unreadable", "name: unreadable\nbackend: llama-cpp\n")
+		load()
+		store.getErr = errors.New("connection reset")
+
+		Expect(ResyncModelConfigRevisions(context.Background(), loader, store)).ToNot(Succeed())
+		Expect(store.applied).To(BeEmpty())
+	})
+})

core/services/nodes/interfaces.go

@@ -60,6 +60,7 @@ type ModelRouter interface {
 	GetNodeLabels(ctx context.Context, nodeID string) ([]NodeLabel, error)
 	FindNodesWithModel(ctx context.Context, modelName string) ([]BackendNode, error)
 	LoadedReplicaStats(ctx context.Context, modelName string, candidateNodeIDs []string) ([]ReplicaCandidate, error)
+	MarkUnhealthy(ctx context.Context, nodeID string) error
 	LoadJobStore
 }
 

core/services/nodes/model_router_test.go

@@ -250,3 +250,7 @@ var _ = Describe("ModelRouterAdapter", func() {
 		})
 	})
 })
+
+func (f *fakeModelRouterForSmartRouter) MarkUnhealthy(_ context.Context, _ string) error {
+	return nil
+}

core/services/nodes/reconciler.go

@@ -278,8 +278,9 @@ func (rc *ReplicaReconciler) reconcileOnce(ctx context.Context) {
 
 // reconcileState runs the state-reconciliation passes: drain pending backend
 // ops for freshly-healthy nodes, reconcile registry rows against what workers
-// report they are running, then port-probe whatever is left. All passes are
-// best-effort: a failure on one node doesn't stop the rest.
+// report they are running, port-probe whatever is left, then reclaim replica
+// slots held by loads nobody is driving. All passes are best-effort: a failure
+// on one node doesn't stop the rest.
 //
 // Order matters. The worker pass runs first and refreshes updated_at for every
 // model a worker vouches for, which takes those rows out of the port prober's
@@ -292,6 +293,9 @@ func (rc *ReplicaReconciler) reconcileState(ctx context.Context) {
 	rc.reconcileNodeProcesses(ctx)
 	rc.probeLoadedModels(ctx)
 	rc.sweepLeakedInFlight(ctx)
+	// Runs last: the passes above can move a row into a serving state, and a
+	// row that just became loaded is no longer this sweeper's business.
+	rc.reclaimAbandonedLoads(ctx)
 }
 
 // drainPendingBackendOps retries queued backend ops whose next_retry_at has