fix(distributed): bound the health monitor's miss streaks to live rows

HealthMonitor.misses holds one consecutive-failed-probe count per (node,
model, replica) and nothing ever removed an entry whose row had gone. It is
the only per-node state in a frontend that grows on model churn rather than on
fleet size, so a deployment that loads and unloads models for months
accumulates an integer per tuple it ever probed and gives none back.

There are four ways a row stops being visible to the pass, not one. A node
departs and the pass skips its probes; a node goes offline or unhealthy on a
stale heartbeat and the pass skips it entirely; an operator sets a node
draining; or the row is removed by an unload, a scale-down or an eviction, and
nothing tells this monitor.

So the bound is the pass itself, and not a subscription on the departure
notifier. The notifier evicts the caches a DEPARTURE invalidates and it keeps
that one meaning; this reads a different fact, that there is no longer a row
to count misses against, and covers all four cases with one rule. A row the
pass could not probe is marked seen before the probe, so an unreachable worker
still leaves its streak exactly as it was rather than having it forgiven; a
pass that could not list the fleet prunes nothing, since it observed nothing.
Forgetting only ever delays a reap by up to the miss threshold and can never
cause one.

Assisted-by: Claude Opus 5 [claude-code]
Signed-off-by: Ettore Di Giacinto <mudler@localai.io>

core/services/nodes/health.go

@@ -51,9 +51,13 @@ type HealthMonitor struct {
 	// See DepartureNotifier: this monitor is its only caller.
 	departures *DepartureNotifier
 	missesMu   sync.Mutex
-	misses     map[modelKey]int // consecutive failed-probe counts; reset on success or model removal
-	cancel     context.CancelFunc
-	cancelMu   sync.Mutex
+	// misses holds one consecutive-failed-probe count per (node, model,
+	// replica). It is bounded to the rows the LAST completed pass actually
+	// walked; see forgetUnseenMisses for why that, and not a departure
+	// subscription, is what bounds it.
+	misses   map[modelKey]int
+	cancel   context.CancelFunc
+	cancelMu sync.Mutex
 }
 
 // NewHealthMonitor creates a new HealthMonitor.
@@ -224,9 +228,15 @@ func (hm *HealthMonitor) doCheckAll(ctx context.Context) {
 	nodes, err := hm.registry.List(ctx)
 	if err != nil {
 		xlog.Error("Health monitor: failed to list nodes", "error", err)
+		// No prune. A pass that could not read the fleet observed nothing, and
+		// pruning against nothing would wipe every streak in progress.
 		return
 	}
 
+	// Every model row this pass walked, whether or not it managed to probe it.
+	// See forgetUnseenMisses.
+	seen := make(map[modelKey]struct{})
+
 	for _, node := range nodes {
 		if node.Status == StatusDraining {
 			continue
@@ -326,6 +336,14 @@ func (hm *HealthMonitor) doCheckAll(ctx context.Context) {
 				if m.WorkerLocalAddress == "" {
 					continue
 				}
+				// Marked seen BEFORE the probe, so that a row this pass could
+				// not reach keeps the streak it already had. The unreachable
+				// branch below leaves that streak exactly as it was, and a
+				// prune that only counted PROBED rows would clear it instead,
+				// forgiving a backend that really has died every time a peer
+				// link blipped.
+				key := modelKey{NodeID: node.ID, ModelName: m.ModelName, ReplicaIndex: m.ReplicaIndex}
+				seen[key] = struct{}{}
 				// Through the node's tunnel, never a direct dial to m.WorkerLocalAddress:
 				// that address is a port inside the worker. A worker this
 				// replica cannot reach is not evidence that its backend died,
@@ -359,7 +377,6 @@ func (hm *HealthMonitor) doCheckAll(ctx context.Context) {
 					continue
 				}
 
-				key := modelKey{NodeID: node.ID, ModelName: m.ModelName, ReplicaIndex: m.ReplicaIndex}
 				hm.missesMu.Lock()
 				if ok {
 					// Probe succeeded — wipe any previous miss streak.
@@ -393,4 +410,55 @@ func (hm *HealthMonitor) doCheckAll(ctx context.Context) {
 			}
 		}
 	}
+
+	hm.forgetUnseenMisses(seen)
+}
+
+// forgetUnseenMisses drops every miss streak whose model row this pass did not
+// walk, so the map is bounded by the fleet's live rows rather than by every
+// (node, model, replica) the process has ever seen.
+//
+// It is level triggered and it is deliberately NOT a subscriber on the
+// departure notifier, though the notifier is where the rest of the per-node
+// caches are evicted from and it would have been the shorter change.
+//
+// A departure subscription would fix ONE of the four ways a streak is
+// stranded. The others are a node marked offline or unhealthy on a stale
+// heartbeat, a node the operator set draining, and a model row removed by
+// anything other than this loop (an unload, a scale-down, an eviction). Each
+// leaves a counter with no row behind it, and each would need its own hook.
+// What all four have in common is exactly the thing this reads: the row was
+// not there when the pass looked. So the notifier keeps its one meaning, "no
+// live replica holds this node's tunnel", and this keeps its own, "there is no
+// longer a row to count misses against".
+//
+// Forgetting is safe in one direction only, and it is the safe one: a cleared
+// streak DELAYS a reap by up to perModelMissThreshold passes and can never
+// cause one. That is what makes it sound to prune on a pass whose
+// GetNodeModels call failed, which is otherwise indistinguishable here from a
+// node with no models.
+func (hm *HealthMonitor) forgetUnseenMisses(seen map[modelKey]struct{}) {
+	hm.missesMu.Lock()
+	defer hm.missesMu.Unlock()
+	for key := range hm.misses {
+		if _, ok := seen[key]; !ok {
+			delete(hm.misses, key)
+		}
+	}
+}
+
+// missCount reports the streak recorded for one model row. It exists for the
+// specs that assert this map is bounded, which is otherwise a property with no
+// observable effect until the process runs out of memory.
+func (hm *HealthMonitor) missCount(nodeID, modelName string, replicaIndex int) int {
+	hm.missesMu.Lock()
+	defer hm.missesMu.Unlock()
+	return hm.misses[modelKey{NodeID: nodeID, ModelName: modelName, ReplicaIndex: replicaIndex}]
+}
+
+// trackedMisses reports how many model rows currently hold a miss streak.
+func (hm *HealthMonitor) trackedMisses() int {
+	hm.missesMu.Lock()
+	defer hm.missesMu.Unlock()
+	return len(hm.misses)
 }

core/services/nodes/health_miss_bounds_test.go

@@ -0,0 +1,152 @@
+package nodes
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+
+	"github.com/mudler/LocalAI/core/services/cluster"
+)
+
+// The miss map counts consecutive failed probes per (node, model, replica). It
+// is the only per-node state in this process that grows on model churn rather
+// than on fleet size, so a deployment that loads and unloads models for months
+// accumulates one integer per tuple it ever probed and never gives one back.
+//
+// The bound is the pass itself: a streak survives only while the pass can still
+// see the row it counts against. Every spec below is one way a row stops being
+// visible, and the last two are the two ways a row must NOT be forgotten.
+var _ = Describe("HealthMonitor miss-streak bounds", func() {
+	const staleThreshold = 30 * time.Second
+
+	var (
+		ctx     context.Context
+		store   *fakeNodeHealthStore
+		factory *fakeBackendClientFactory
+		hm      *HealthMonitor
+	)
+
+	// deadModel is one model row whose backend answers every probe with a
+	// failure, so a single pass leaves a streak of exactly one.
+	deadModel := func(nodeID, modelName, addr string) NodeModel {
+		factory.setClient(addr, &fakeBackendClient{healthy: false, err: fmt.Errorf("connection refused")})
+		return NodeModel{NodeID: nodeID, ModelName: modelName, WorkerLocalAddress: addr}
+	}
+
+	BeforeEach(func() {
+		ctx = context.Background()
+		store = newFakeNodeHealthStore()
+		factory = newFakeBackendClientFactory()
+		hm = newTestHealthMonitor(store, factory, true, staleThreshold)
+		hm.perModelHealthCheck = true
+	})
+
+	It("forgets the streak of a model row that is no longer registered", func() {
+		// An unload, a scale-down or an LRU eviction removes the row without
+		// telling this monitor. Nothing else in the loop ever revisits the key,
+		// so the counter is stranded for the life of the process.
+		store.addNode(makeTestNode("node-a", "worker-a", "10.0.0.1:50051", StatusHealthy, freshTime()))
+		store.setNodeModels("node-a", deadModel("node-a", "m", "10.0.0.1:50053"))
+
+		hm.doCheckAll(ctx)
+		Expect(hm.missCount("node-a", "m", 0)).To(Equal(1))
+
+		store.setNodeModels("node-a")
+		hm.doCheckAll(ctx)
+
+		Expect(hm.trackedMisses()).To(BeZero())
+	})
+
+	It("forgets the streaks of a node whose heartbeat went stale", func() {
+		// The pass skips the node entirely from this branch, so every streak it
+		// held is unreachable from here on. autoOffline deletes its rows too,
+		// which is what makes the counters count against nothing.
+		store.addNode(makeTestNode("node-b", "worker-b", "10.0.0.2:50051", StatusHealthy, freshTime()))
+		store.setNodeModels("node-b", deadModel("node-b", "m", "10.0.0.2:50053"))
+
+		hm.doCheckAll(ctx)
+		Expect(hm.missCount("node-b", "m", 0)).To(Equal(1))
+
+		store.setHeartbeat("node-b", staleTime(staleThreshold))
+		hm.doCheckAll(ctx)
+
+		Expect(hm.trackedMisses()).To(BeZero())
+	})
+
+	It("forgets the streaks of a node the operator set draining", func() {
+		store.addNode(makeTestNode("node-c", "worker-c", "10.0.0.3:50051", StatusHealthy, freshTime()))
+		store.setNodeModels("node-c", deadModel("node-c", "m", "10.0.0.3:50053"))
+
+		hm.doCheckAll(ctx)
+		Expect(hm.missCount("node-c", "m", 0)).To(Equal(1))
+
+		store.getNode("node-c").Status = StatusDraining
+		hm.doCheckAll(ctx)
+
+		Expect(hm.trackedMisses()).To(BeZero())
+	})
+
+	It("forgets the streaks of a node whose tunnel departed", func() {
+		// The departure branch skips the probes deliberately: there is no route
+		// to dial. It also announces the departure, and every other per-node
+		// cache is dropped from that announcement. This one is not, because a
+		// departure is only one of the four ways a row stops being visible.
+		store.addNode(makeTestNode("node-d", "worker-d", "10.0.0.4:50051", StatusHealthy, freshTime()))
+		store.setNodeModels("node-d", deadModel("node-d", "m", "10.0.0.4:50053"))
+
+		hm.doCheckAll(ctx)
+		Expect(hm.missCount("node-d", "m", 0)).To(Equal(1))
+
+		hm.presence = &stubPresence{answer: cluster.PresenceGone}
+		hm.reconnectGrace = time.Minute
+		hm.departures = NewDepartureNotifier()
+		hm.doCheckAll(ctx)
+
+		Expect(hm.trackedMisses()).To(BeZero())
+	})
+
+	It("keeps the streak of a row the pass could not probe", func() {
+		// An unreachable worker is not evidence about its backends. The streak
+		// is neither advanced nor cleared, so a row that was two misses from
+		// removal before the peer link blipped is still two misses from removal
+		// after it. A prune that counted only PROBED rows would forgive it.
+		store.addNode(makeTestNode("node-e", "worker-e", "10.0.0.5:50051", StatusHealthy, freshTime()))
+		store.setNodeModels("node-e", deadModel("node-e", "m", "10.0.0.5:50053"))
+
+		hm.doCheckAll(ctx)
+		Expect(hm.missCount("node-e", "m", 0)).To(Equal(1))
+
+		factory.refuseForNode = fmt.Errorf("no tunnel for you")
+		hm.doCheckAll(ctx)
+
+		Expect(hm.missCount("node-e", "m", 0)).To(Equal(1))
+	})
+
+	It("keeps every streak when the pass could not read the fleet", func() {
+		// A pass that failed to list nodes observed nothing at all, and pruning
+		// against nothing would clear every streak in the deployment on one
+		// database hiccup.
+		store.addNode(makeTestNode("node-f", "worker-f", "10.0.0.6:50051", StatusHealthy, freshTime()))
+		store.setNodeModels("node-f", deadModel("node-f", "m", "10.0.0.6:50053"))
+
+		hm.doCheckAll(ctx)
+		Expect(hm.missCount("node-f", "m", 0)).To(Equal(1))
+
+		hm.registry = &listFailsStore{NodeHealthStore: store, err: fmt.Errorf("connection reset")}
+		hm.doCheckAll(ctx)
+
+		Expect(hm.missCount("node-f", "m", 0)).To(Equal(1))
+	})
+})
+
+// listFailsStore is the fake store with its fleet listing broken, which a real
+// registry cannot be made to do on demand.
+type listFailsStore struct {
+	NodeHealthStore
+	err error
+}
+
+func (s *listFailsStore) List(context.Context) ([]BackendNode, error) { return nil, s.err }

core/services/nodes/health_mock_test.go

@@ -53,6 +53,24 @@ func (f *fakeNodeHealthStore) addNodeModel(nodeID string, nm NodeModel) {
 	f.models[nodeID] = append(f.models[nodeID], nm)
 }
 
+// setNodeModels replaces a node's model rows, so a spec can make a row
+// disappear the way an unload, a scale-down or an eviction does.
+func (f *fakeNodeHealthStore) setNodeModels(nodeID string, models ...NodeModel) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.models[nodeID] = models
+}
+
+// setHeartbeat backdates a node's heartbeat so a spec can drive the stale
+// branch without waiting.
+func (f *fakeNodeHealthStore) setHeartbeat(nodeID string, at time.Time) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	if n, ok := f.nodes[nodeID]; ok {
+		n.LastHeartbeat = at
+	}
+}
+
 func (f *fakeNodeHealthStore) getNode(id string) *BackendNode {
 	f.mu.Lock()
 	defer f.mu.Unlock()