diff --git a/core/application/distributed.go b/core/application/distributed.go index 1fceee1317d2..4d9f3e376377 100644 --- a/core/application/distributed.go +++ b/core/application/distributed.go @@ -365,8 +365,10 @@ func initDistributed(cfg *config.ApplicationConfig, authDB *gorm.DB, configLoade // All dependencies ready — build SmartRouter with all options at once var conflictResolver nodes.ConcurrencyConflictResolver + var pinnedResolver nodes.PinnedModelResolver if configLoader != nil { conflictResolver = configLoader + pinnedResolver = configLoader } modelCleanup := nodes.NewModelCleanupService(registry, remoteUnloader) router := nodes.NewSmartRouter(registry, nodes.SmartRouterOptions{ @@ -377,6 +379,7 @@ func initDistributed(cfg *config.ApplicationConfig, authDB *gorm.DB, configLoade AuthToken: routerAuthToken, DB: authDB, ConflictResolver: conflictResolver, + PinnedResolver: pinnedResolver, PrefixProvider: prefixProvider, PrefixConfig: prefixCfg, Pressure: pressure, @@ -439,6 +442,7 @@ func initDistributed(cfg *config.ApplicationConfig, authDB *gorm.DB, configLoade ProbeStaleAfter: 2 * time.Minute, Pressure: pressure, PressureThreshold: prefixCfg.PressureScaleThreshold, + PinnedResolver: pinnedResolver, }) // Create ModelRouterAdapter to wire into ModelLoader diff --git a/core/config/model_config_loader.go b/core/config/model_config_loader.go index 9120cc2685ca..7733a73922f4 100644 --- a/core/config/model_config_loader.go +++ b/core/config/model_config_loader.go @@ -412,6 +412,26 @@ func (bcl *ModelConfigLoader) GetModelsConflictingWith(name string) []string { return conflicts } +// GetPinnedModelNames returns the names of every configured, not-disabled +// model with `pinned: true`. The distributed router and replica reconciler +// consult this so cluster-wide eviction honours the same pin contract the +// local watchdog enforces (#11101) — without it, a pinned model becomes +// eviction-eligible the moment its in-flight count drops to zero. +func (bcl *ModelConfigLoader) GetPinnedModelNames() []string { + bcl.Lock() + defer bcl.Unlock() + var pinned []string + for n, cfg := range bcl.configs { + if cfg.IsDisabled() { + continue + } + if cfg.IsPinned() { + pinned = append(pinned, n) + } + } + return pinned +} + // UpdateModelConfig updates an existing model config in the loader. // This is useful for updating runtime-detected properties like thinking support. func (bcl *ModelConfigLoader) UpdateModelConfig(m string, updater func(*ModelConfig)) { diff --git a/core/config/model_config_loader_test.go b/core/config/model_config_loader_test.go index d654226efb46..1a3e9b03ae0e 100644 --- a/core/config/model_config_loader_test.go +++ b/core/config/model_config_loader_test.go @@ -267,6 +267,40 @@ var _ = Describe("ModelConfigLoader.GetModelsConflictingWith", func() { }) }) +var _ = Describe("ModelConfigLoader.GetPinnedModelNames", func() { + var bcl *ModelConfigLoader + + BeforeEach(func() { + bcl = NewModelConfigLoader("/tmp/pinned-test-models") + }) + + insert := func(cfg ModelConfig) { + bcl.Lock() + bcl.configs[cfg.Name] = cfg + bcl.Unlock() + } + + boolPtr := func(b bool) *bool { return &b } + + It("returns nil when nothing is pinned", func() { + insert(ModelConfig{Name: "a"}) + insert(ModelConfig{Name: "b", Pinned: boolPtr(false)}) + Expect(bcl.GetPinnedModelNames()).To(BeNil()) + }) + + It("returns only pinned, enabled models", func() { + insert(ModelConfig{Name: "a", Pinned: boolPtr(true)}) + insert(ModelConfig{Name: "b"}) + insert(ModelConfig{Name: "c", Pinned: boolPtr(true)}) + Expect(bcl.GetPinnedModelNames()).To(ConsistOf("a", "c")) + }) + + It("ignores disabled pinned models", func() { + insert(ModelConfig{Name: "a", Pinned: boolPtr(true), Disabled: boolPtr(true)}) + Expect(bcl.GetPinnedModelNames()).To(BeNil()) + }) +}) + var _ = Describe("ModelConfigLoader alias resolution", func() { var loader *ModelConfigLoader diff --git a/core/services/nodes/interfaces.go b/core/services/nodes/interfaces.go index be4dbc25d916..aafa0e47f4d2 100644 --- a/core/services/nodes/interfaces.go +++ b/core/services/nodes/interfaces.go @@ -46,7 +46,7 @@ type ModelRouter interface { FindIdleNode(ctx context.Context) (*BackendNode, error) FindLeastLoadedNode(ctx context.Context) (*BackendNode, error) FindGlobalLRUModelWithZeroInFlight(ctx context.Context) (*NodeModel, error) - FindLRUModel(ctx context.Context, nodeID string) (*NodeModel, error) + FindLRUModel(ctx context.Context, nodeID string, excludeModels []string) (*NodeModel, error) Get(ctx context.Context, nodeID string) (*BackendNode, error) GetModelScheduling(ctx context.Context, modelName string) (*ModelSchedulingConfig, error) GetGoverningScheduling(ctx context.Context, modelName string) (*ModelSchedulingConfig, error) @@ -84,6 +84,15 @@ type ConcurrencyConflictResolver interface { GetModelsConflictingWith(modelName string) []string } +// PinnedModelResolver reports which configured models are pinned. Satisfied +// by *config.ModelConfigLoader. The router's eviction paths and the +// reconciler's idle scale-down exclude these models so `pinned: true` holds +// cluster-wide, not just against the per-node watchdog (#11101). Deliberate +// teardown (admin unload, model delete, node drain) intentionally bypasses it. +type PinnedModelResolver interface { + GetPinnedModelNames() []string +} + // NodeHealthStore is used by HealthMonitor for node status management. type NodeHealthStore interface { List(ctx context.Context) ([]BackendNode, error) diff --git a/core/services/nodes/model_router_test.go b/core/services/nodes/model_router_test.go index 9a77d96ae2bb..fe19884070f5 100644 --- a/core/services/nodes/model_router_test.go +++ b/core/services/nodes/model_router_test.go @@ -114,7 +114,7 @@ func (f *fakeModelRouterForSmartRouter) FindLeastLoadedNode(_ context.Context) ( func (f *fakeModelRouterForSmartRouter) FindGlobalLRUModelWithZeroInFlight(_ context.Context) (*NodeModel, error) { return nil, nil } -func (f *fakeModelRouterForSmartRouter) FindLRUModel(_ context.Context, _ string) (*NodeModel, error) { +func (f *fakeModelRouterForSmartRouter) FindLRUModel(_ context.Context, _ string, _ []string) (*NodeModel, error) { return nil, nil } func (f *fakeModelRouterForSmartRouter) Get(_ context.Context, nodeID string) (*BackendNode, error) { diff --git a/core/services/nodes/pinned_eviction_test.go b/core/services/nodes/pinned_eviction_test.go new file mode 100644 index 000000000000..64ba0cdca3d2 --- /dev/null +++ b/core/services/nodes/pinned_eviction_test.go @@ -0,0 +1,235 @@ +package nodes + +// Regression tests for #11101: `pinned: true` must hold cluster-wide, not +// just against the per-node watchdog. Before the fix, every distributed +// eviction path (EvictLRU, evictLRUAndFreeNode, scaleDownIdle) was +// pinned-blind, so a pinned model became eviction-eligible the moment its +// in-flight count dropped to zero — observable as the backend being freed +// immediately after every request under capacity pressure. + +import ( + "context" + "time" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + + "github.com/mudler/LocalAI/core/services/testutil" + "gorm.io/gorm" + "runtime" +) + +type fakePinnedResolver struct { + names []string +} + +func (f *fakePinnedResolver) GetPinnedModelNames() []string { return f.names } + +var _ = Describe("Pinned models vs distributed eviction (#11101)", func() { + Describe("EvictLRU (mock-based)", func() { + It("excludes pinned models at the query, evicting nothing when only a pinned model is idle", func() { + reg := &fakeModelRouter{ + findLRUModel: &NodeModel{NodeID: "n1", ModelName: "pinned-model"}, + } + unloader := &fakeUnloader{} + + router := NewSmartRouter(reg, SmartRouterOptions{ + Unloader: unloader, + PinnedResolver: &fakePinnedResolver{names: []string{"pinned-model"}}, + }) + + _, err := router.EvictLRU(context.Background(), "n1") + Expect(err).To(HaveOccurred()) + // The exclusion must reach the registry query: filtering after + // selection would just burn the attempt instead of picking the + // next-oldest unpinned model. + Expect(reg.findLRUExclude).To(ContainElement("pinned-model")) + Expect(unloader.stopCalls).To(BeEmpty()) + }) + + It("still evicts unpinned models when a resolver is wired", func() { + reg := &fakeModelRouter{ + findLRUModel: &NodeModel{NodeID: "n1", ModelName: "plain-model"}, + } + unloader := &fakeUnloader{} + + router := NewSmartRouter(reg, SmartRouterOptions{ + Unloader: unloader, + PinnedResolver: &fakePinnedResolver{names: []string{"some-other-model"}}, + }) + + evicted, err := router.EvictLRU(context.Background(), "n1") + Expect(err).ToNot(HaveOccurred()) + Expect(evicted).To(Equal("plain-model")) + Expect(unloader.stopCalls).To(ContainElement("n1:plain-model")) + }) + }) + + Describe("evictLRUAndFreeNode (integration)", func() { + var ( + db *gorm.DB + registry *NodeRegistry + node *BackendNode + ) + + BeforeEach(func() { + if runtime.GOOS == "darwin" { + Skip("testcontainers requires Docker, not available on macOS CI") + } + db = testutil.SetupTestDB() + var err error + registry, err = NewNodeRegistry(db) + Expect(err).ToNot(HaveOccurred()) + + node = &BackendNode{ + Name: "pinned-evict-node", + NodeType: NodeTypeBackend, + Address: "10.0.0.200:50051", + } + Expect(registry.Register(context.Background(), node, true)).To(Succeed()) + Expect(registry.MarkHealthy(context.Background(), node.ID)).To(Succeed()) + }) + + // setLoadedModel creates an idle loaded replica row with a controlled + // last_used so the specs can dictate LRU order. + setLoadedModel := func(name string, lastUsed time.Time) { + Expect(registry.SetNodeModel(context.Background(), node.ID, name, 0, "loaded", "", 0)).To(Succeed()) + Expect(db.Model(&NodeModel{}). + Where("node_id = ? AND model_name = ?", node.ID, name). + Update("last_used", lastUsed).Error).ToNot(HaveOccurred()) + } + + It("skips the pinned LRU model and evicts the next-oldest unpinned one", func() { + // The pinned model is the older (= LRU) candidate. Without the + // exclusion it would be selected first. + setLoadedModel("pinned-old", time.Now().Add(-2*time.Hour)) + setLoadedModel("plain-new", time.Now().Add(-1*time.Hour)) + + router := NewSmartRouter(registry, SmartRouterOptions{ + DB: db, + PinnedResolver: &fakePinnedResolver{names: []string{"pinned-old"}}, + }) + + freed, err := router.evictLRUAndFreeNode(context.Background()) + Expect(err).ToNot(HaveOccurred()) + Expect(freed.ID).To(Equal(node.ID)) + + var remaining []NodeModel + Expect(db.Where("node_id = ?", node.ID).Find(&remaining).Error).ToNot(HaveOccurred()) + Expect(remaining).To(HaveLen(1)) + Expect(remaining[0].ModelName).To(Equal("pinned-old")) + }) + + It("returns ErrEvictionBusy instead of evicting when every idle model is pinned", func() { + setLoadedModel("pinned-only", time.Now().Add(-2*time.Hour)) + + router := NewSmartRouter(registry, SmartRouterOptions{ + DB: db, + PinnedResolver: &fakePinnedResolver{names: []string{"pinned-only"}}, + }) + + _, err := router.evictLRUAndFreeNode(context.Background()) + Expect(err).To(MatchError(ErrEvictionBusy)) + + var count int64 + Expect(db.Model(&NodeModel{}).Where("model_name = ?", "pinned-only").Count(&count).Error).ToNot(HaveOccurred()) + Expect(count).To(Equal(int64(1)), "the pinned model's replica row must survive") + }) + + It("still evicts the LRU model when no resolver is wired (embedder/back-compat path)", func() { + setLoadedModel("plain-old", time.Now().Add(-2*time.Hour)) + + router := NewSmartRouter(registry, SmartRouterOptions{DB: db}) + + freed, err := router.evictLRUAndFreeNode(context.Background()) + Expect(err).ToNot(HaveOccurred()) + Expect(freed.ID).To(Equal(node.ID)) + }) + }) + + Describe("scaleDownIdle (integration)", func() { + var ( + db *gorm.DB + registry *NodeRegistry + ) + + BeforeEach(func() { + if runtime.GOOS == "darwin" { + Skip("testcontainers requires Docker, not available on macOS CI") + } + db = testutil.SetupTestDB() + var err error + registry, err = NewNodeRegistry(db) + Expect(err).ToNot(HaveOccurred()) + }) + + registerNode := func(name, address string) *BackendNode { + node := &BackendNode{ + Name: name, + NodeType: NodeTypeBackend, + Address: address, + MaxReplicasPerModel: 4, + } + Expect(registry.Register(context.Background(), node, true)).To(Succeed()) + return node + } + + It("does not scale down idle replicas of a pinned model", func() { + node1 := registerNode("pin-idle-1", "10.0.1.1:50051") + node2 := registerNode("pin-idle-2", "10.0.1.2:50051") + node3 := registerNode("pin-idle-3", "10.0.1.3:50051") + Expect(registry.SetModelScheduling(context.Background(), &ModelSchedulingConfig{ + ModelName: "pinned-replicated", MinReplicas: 1, MaxReplicas: 4, + })).To(Succeed()) + + // Three idle replicas above the floor of one — prime scale-down bait. + pastTime := time.Now().Add(-10 * time.Minute) + for _, n := range []*BackendNode{node1, node2, node3} { + Expect(registry.SetNodeModel(context.Background(), n.ID, "pinned-replicated", 0, "loaded", "", 0)).To(Succeed()) + db.Model(&NodeModel{}).Where("node_id = ? AND model_name = ?", n.ID, "pinned-replicated"). + Update("last_used", pastTime) + } + + unloader := &fakeUnloader{} + reconciler := NewReplicaReconciler(ReplicaReconcilerOptions{ + Registry: registry, + Unloader: unloader, + DB: db, + ScaleDownDelay: 1 * time.Minute, + PinnedResolver: &fakePinnedResolver{names: []string{"pinned-replicated"}}, + }) + + reconciler.reconcile(context.Background()) + + Expect(unloader.unloadCalls).To(BeEmpty()) + }) + + It("scales down unpinned models normally with a resolver wired", func() { + node1 := registerNode("unpin-idle-1", "10.0.1.4:50051") + node2 := registerNode("unpin-idle-2", "10.0.1.5:50051") + Expect(registry.SetModelScheduling(context.Background(), &ModelSchedulingConfig{ + ModelName: "plain-replicated", MinReplicas: 1, MaxReplicas: 4, + })).To(Succeed()) + + pastTime := time.Now().Add(-10 * time.Minute) + for _, n := range []*BackendNode{node1, node2} { + Expect(registry.SetNodeModel(context.Background(), n.ID, "plain-replicated", 0, "loaded", "", 0)).To(Succeed()) + db.Model(&NodeModel{}).Where("node_id = ? AND model_name = ?", n.ID, "plain-replicated"). + Update("last_used", pastTime) + } + + unloader := &fakeUnloader{} + reconciler := NewReplicaReconciler(ReplicaReconcilerOptions{ + Registry: registry, + Unloader: unloader, + DB: db, + ScaleDownDelay: 1 * time.Minute, + PinnedResolver: &fakePinnedResolver{names: []string{"some-other-model"}}, + }) + + reconciler.reconcile(context.Background()) + + Expect(unloader.unloadCalls).To(HaveLen(1)) + }) + }) +}) diff --git a/core/services/nodes/reconciler.go b/core/services/nodes/reconciler.go index 62cc73e1545e..a14a3fa70d5a 100644 --- a/core/services/nodes/reconciler.go +++ b/core/services/nodes/reconciler.go @@ -5,6 +5,7 @@ import ( "encoding/json" "errors" "fmt" + "slices" "sync" "time" @@ -155,6 +156,9 @@ type ReplicaReconciler struct { // what a backend inside a request cannot do. inFlightIdleMu sync.Mutex inFlightIdle map[string]int + // pinnedResolver exempts pinned models from idle scale-down (see + // ReplicaReconcilerOptions.PinnedResolver). nil disables the exemption. + pinnedResolver PinnedModelResolver } // ModelScheduler abstracts the scheduling logic needed by the reconciler. @@ -192,6 +196,12 @@ type ReplicaReconcilerOptions struct { // PressureThreshold is the forced-disturb count within PressureWindow that // triggers a scale-up. Default prefixcache.DefaultConfig().PressureScaleThreshold (1). PressureThreshold int + // PinnedResolver, when set, exempts `pinned: true` models from idle + // scale-down so the pin contract holds cluster-wide (#11101). nil + // disables the exemption. Dead-row reaping is unaffected: it removes + // registry rows for processes that are already gone, which is state + // correction, not eviction. + PinnedResolver PinnedModelResolver } // NewReplicaReconciler creates a new ReplicaReconciler. @@ -235,6 +245,7 @@ func NewReplicaReconciler(opts ReplicaReconcilerOptions) *ReplicaReconciler { probeStaleAfter: probeStaleAfter, pressure: opts.Pressure, pressureThreshold: pressureThreshold, + pinnedResolver: opts.PinnedResolver, } } @@ -1083,11 +1094,18 @@ func (rc *ReplicaReconciler) scaleUp(ctx context.Context, cfg ModelSchedulingCon return scheduled > 0 } -// scaleDownIdle removes idle replicas above the floor. +// scaleDownIdle removes idle replicas above the floor. Pinned models are +// exempt entirely: `pinned: true` promises the operator the model stays +// resident, and trimming to a floor of one still means every request beyond +// the survivor's capacity pays a cold reload (#11101). func (rc *ReplicaReconciler) scaleDownIdle(ctx context.Context, cfg ModelSchedulingConfig, current, floor int) { if rc.unloader == nil { return } + if rc.pinnedResolver != nil && slices.Contains(rc.pinnedResolver.GetPinnedModelNames(), cfg.ModelName) { + xlog.Debug("Reconciler: skipping idle scale-down for pinned model", "model", cfg.ModelName) + return + } // Find idle replicas that have been unused for longer than scaleDownDelay. // Order by replica_index DESC first, then last_used ASC: trim the diff --git a/core/services/nodes/reconciler_inflight_leak_test.go b/core/services/nodes/reconciler_inflight_leak_test.go index f3574fab7e0c..4898d8852f38 100644 --- a/core/services/nodes/reconciler_inflight_leak_test.go +++ b/core/services/nodes/reconciler_inflight_leak_test.go @@ -149,14 +149,14 @@ var _ = Describe("ReplicaReconciler — leaked in_flight sweeper", func() { seed("pinned", 1, 2*inFlightLeakIdleAfter) rc := newReconciler(&fakeProber{outcomes: map[string]ProbeOutcome{addr: ProbeAlive}}) - _, err := registry.FindLRUModel(context.Background(), node.ID) + _, err := registry.FindLRUModel(context.Background(), node.ID, nil) Expect(err).To(HaveOccurred(), "precondition: the leak hides the row from LRU") for range inFlightLeakConfirmations { rc.sweepLeakedInFlight(context.Background()) } - lru, err := registry.FindLRUModel(context.Background(), node.ID) + lru, err := registry.FindLRUModel(context.Background(), node.ID, nil) Expect(err).ToNot(HaveOccurred()) Expect(lru.ModelName).To(Equal("pinned")) }) diff --git a/core/services/nodes/registry.go b/core/services/nodes/registry.go index f799f291f902..19f2795ce0a6 100644 --- a/core/services/nodes/registry.go +++ b/core/services/nodes/registry.go @@ -2000,10 +2000,13 @@ func (r *NodeRegistry) FindNodeForModel(ctx context.Context, modelName string) ( } // FindLRUModel returns the least-recently-used model on a node. -func (r *NodeRegistry) FindLRUModel(ctx context.Context, nodeID string) (*NodeModel, error) { +func (r *NodeRegistry) FindLRUModel(ctx context.Context, nodeID string, excludeModels []string) (*NodeModel, error) { var nm NodeModel - err := currentModelRevision(r.db.WithContext(ctx)).Where("node_models.node_id = ? AND node_models.state = ? AND node_models.in_flight = 0", nodeID, "loaded"). - Order("last_used ASC").First(&nm).Error + q := currentModelRevision(r.db.WithContext(ctx)).Where("node_models.node_id = ? AND node_models.state = ? AND node_models.in_flight = 0", nodeID, "loaded") + if len(excludeModels) > 0 { + q = q.Where("node_models.model_name NOT IN ?", excludeModels) + } + err := q.Order("last_used ASC").First(&nm).Error if err != nil { return nil, fmt.Errorf("finding LRU model on node %s: %w", nodeID, err) } diff --git a/core/services/nodes/revision_eligibility_test.go b/core/services/nodes/revision_eligibility_test.go index 96ea5010b704..01f3bfd37475 100644 --- a/core/services/nodes/revision_eligibility_test.go +++ b/core/services/nodes/revision_eligibility_test.go @@ -107,7 +107,7 @@ var _ = Describe("revision eligibility consumers", func() { }), Entry("FindLRUModel", func() []string { Expect(db.Model(&NodeModel{}).Where("id IN ?", []string{"empty", "mismatch", "unloading"}).Update("node_id", nodes["current"].ID).Error).To(Succeed()) - row, err := registry.FindLRUModel(ctx, nodes["current"].ID) + row, err := registry.FindLRUModel(ctx, nodes["current"].ID, nil) Expect(err).NotTo(HaveOccurred()) return []string{row.ID} }), diff --git a/core/services/nodes/router.go b/core/services/nodes/router.go index 44acede1159a..ef5cdd3f3b29 100644 --- a/core/services/nodes/router.go +++ b/core/services/nodes/router.go @@ -53,6 +53,12 @@ type SmartRouterOptions struct { // anti-affinity is disabled at the scheduler layer; the per-node // watchdog still enforces the rule on arrival. ConflictResolver ConcurrencyConflictResolver + // PinnedResolver, when set, excludes `pinned: true` models from the + // automatic eviction paths (EvictLRU, evictLRUAndFreeNode) so the pin + // contract holds cluster-wide, mirroring the per-node watchdog (#11101). + // nil disables the exclusion. Deliberate teardown (UnloadModel, admin + // endpoints, node drain) is unaffected. + PinnedResolver PinnedModelResolver // PrefixProvider, when set, enables prefix-cache-aware routing: requests // carrying a prompt prefix chain (distributedhdr.PrefixChain) are biased // toward the node that already holds the longest matching prefix, subject @@ -161,6 +167,9 @@ type SmartRouter struct { db *gorm.DB // for advisory locks during routing stagingTracker *StagingTracker // tracks file staging progress for UI visibility conflictResolver ConcurrencyConflictResolver + // pinnedResolver feeds the eviction paths the set of pinned model names + // (see SmartRouterOptions.PinnedResolver). nil disables the exclusion. + pinnedResolver PinnedModelResolver // prefixProvider is the prefix-cache routing seam (nil disables it; see // SmartRouterOptions.PrefixProvider). prefixConfig holds the global policy // and thresholds. @@ -244,6 +253,7 @@ func NewSmartRouter(registry ModelRouter, opts SmartRouterOptions) *SmartRouter db: opts.DB, stagingTracker: NewStagingTracker(), conflictResolver: opts.ConflictResolver, + pinnedResolver: opts.PinnedResolver, probeCache: newProbeCache(probeCacheTTL), prefixProvider: opts.PrefixProvider, prefixConfig: opts.PrefixConfig, @@ -1996,10 +2006,20 @@ func (r *SmartRouter) UnloadModel(ctx context.Context, nodeID, modelName string) return nil } +// pinnedModelNames returns the pinned set for eviction exclusion, or nil when +// no resolver is wired (embedders, tests, deployments without a config loader). +func (r *SmartRouter) pinnedModelNames() []string { + if r.pinnedResolver == nil { + return nil + } + return r.pinnedResolver.GetPinnedModelNames() +} + // EvictLRU evicts the least-recently-used model from a node to make room. -// Returns the name of the evicted model, or empty string if nothing could be evicted. +// Returns the name of the evicted model, or empty string if nothing could be +// evicted. Pinned models are never candidates (#11101). func (r *SmartRouter) EvictLRU(ctx context.Context, nodeID string) (string, error) { - lru, err := r.registry.FindLRUModel(ctx, nodeID) + lru, err := r.registry.FindLRUModel(ctx, nodeID, r.pinnedModelNames()) if err != nil { return "", fmt.Errorf("finding LRU model on node %s: %w", nodeID, err) } @@ -2072,6 +2092,12 @@ func (r *SmartRouter) evictLRUAndFreeNodeFrom(ctx context.Context, candidateNode if len(candidateNodeIDs) > 0 { q = q.Where("node_models.node_id IN ?", candidateNodeIDs) } + // Pinned models are protected from automatic eviction (#11101). + // Filtered in the query, not after selection, so the next-oldest + // unpinned model is chosen instead of the attempt being wasted. + if pinned := r.pinnedModelNames(); len(pinned) > 0 { + q = q.Where("node_models.model_name NOT IN ?", pinned) + } if err := q. Order("node_models.last_used ASC"). First(&lru).Error; err != nil { @@ -2101,9 +2127,10 @@ func (r *SmartRouter) evictLRUAndFreeNodeFrom(ctx context.Context, candidateNode return node, nil } - // gorm.ErrRecordNotFound means all models have in-flight requests + // gorm.ErrRecordNotFound means every candidate is either mid-request + // or excluded as pinned if attempt == 0 { - xlog.Info("All models have in-flight requests, waiting for capacity") + xlog.Info("No evictable model (all busy or pinned), waiting for capacity") } select { case <-ctx.Done(): diff --git a/core/services/nodes/router_test.go b/core/services/nodes/router_test.go index 10c64632911d..b36f915b042e 100644 --- a/core/services/nodes/router_test.go +++ b/core/services/nodes/router_test.go @@ -5,6 +5,7 @@ import ( "errors" "fmt" "runtime" + "slices" "sync" "time" @@ -92,6 +93,9 @@ type fakeModelRouter struct { // FindLRUModel returns findLRUModel *NodeModel findLRUErr error + // findLRUExclude records the exclusion list EvictLRU passed, so specs can + // assert pinned models were filtered at the query, not post-hoc. + findLRUExclude []string // NextFreeReplicaIndex returns nextFreeReplicaIdx int @@ -344,7 +348,11 @@ func (f *fakeModelRouter) FindGlobalLRUModelWithZeroInFlight(_ context.Context) return f.findGlobalLRUModel, f.findGlobalLRUErr } -func (f *fakeModelRouter) FindLRUModel(_ context.Context, _ string) (*NodeModel, error) { +func (f *fakeModelRouter) FindLRUModel(_ context.Context, _ string, excludeModels []string) (*NodeModel, error) { + f.findLRUExclude = excludeModels + if f.findLRUModel != nil && slices.Contains(excludeModels, f.findLRUModel.ModelName) { + return nil, fmt.Errorf("finding LRU model: record not found") + } return f.findLRUModel, f.findLRUErr } diff --git a/docs/content/advanced/vram-management.md b/docs/content/advanced/vram-management.md index 020f8b63e6aa..c24683d7599a 100644 --- a/docs/content/advanced/vram-management.md +++ b/docs/content/advanced/vram-management.md @@ -214,6 +214,14 @@ choosing where to load a new model, it prefers nodes that don't already host a same-group model, falling back to eviction only if every candidate has a conflict. +`pinned: true` also holds cluster-wide: the distributed scheduler's LRU +eviction (used to free capacity for a new model) and the replica reconciler's +idle scale-down both exclude pinned models, matching what the per-node +watchdog already guarantees. Deliberate teardown - an admin unload, deleting +the model, draining a node - still applies to pinned models. Note that if +every idle model on the cluster is pinned, a new model's load waits for +capacity instead of evicting one. + `concurrency_groups` composes with `NodeSelector` (which decides *which nodes* a model is eligible for) - the two filters apply in sequence. Use `NodeSelector` to target hardware classes; use `concurrency_groups` to keep