fix(distributed): drop replica rows when a worker stops a backend

A worker returns a stopped backend's gRPC port to its allocator as soon as
the process is confirmed dead, and hands it to the next backend that starts.
The controller's NodeModel row for the old address survives, and both
SmartRouter.probeHealth and the HealthMonitor per-model probe verify
liveness, not identity, so once an unrelated backend binds the recycled port
the stale row passes every check and the request is served by the wrong
backend instead of failing.

backend.delete is newly able to trigger this: before #10956 a delete never
actually stopped a process, so it never recycled a port. backend.upgrade has
the identical gap and always did — upgradeBackend force-stops every process
using the binary and starts none back up, while
DistributedBackendManager.UpgradeBackend never removes rows. model.unload is
the one path that gets this right today: it calls RemoveAllNodeModelReplicas
straight after StopBackend.

Report the process keys the worker terminated on the delete and upgrade
replies, and drop the matching rows in RemoteUnloaderAdapter, which already
holds a ModelLocator with RemoveNodeModel. All three call sites funnel
through that adapter, so no new interface, DB migration, or proto change is
needed. A key is reported only once its process is confirmed gone, so the
list stays trustworthy on the partial-failure replies too.

Old workers never populate the new fields. ReportsStoppedProcesses tells
"stopped nothing" apart from "does not report", so an old worker's silence
falls back to the pre-existing probe-based staleness recovery instead of
being mistaken for a completed cleanup.

Quarantine released ports for a short window as an interlock covering the
NATS round-trip between the worker freeing the port and the controller
dropping the row. It is deliberately not derived from HealthCheckInterval:
that cadence is operator-tunable and the per-model reaper can be disabled
outright, so coupling a worker-local constant to it would be silently wrong
on some clusters. Eager row removal is the fix; the delay only closes the
handoff gap.

Identity verification in probeHealth was considered and rejected: Health and
Status carry no backend identity, so it needs a proto change plus an
implementation in 36 Python and 4 C++ Health servicers, it is fail-open for
any backend not yet rebuilt, and the probeCache short-circuit means it would
not even execute during the 30s window where the misroute happens.

Fixes #10952
Refs #10954, #10956

Assisted-by: Claude:claude-opus-4-8 golangci-lint
Signed-off-by: Ettore Di Giacinto <mudler@localai.io>

Author: mudler

core/services/messaging/subjects.go

@@ -226,6 +226,14 @@ type BackendUpgradeRequest struct {
 type BackendUpgradeReply struct {
 	Success bool   `json:"success"`
 	Error   string `json:"error,omitempty"`
+
+	// StoppedProcessKeys / ReportsStoppedProcesses carry the same
+	// stale-row-invalidation contract as on BackendDeleteReply; an upgrade
+	// force-stops every process using the binary and starts none back up, so it
+	// recycles ports exactly the way a delete does. See that type for why the
+	// boolean is not redundant with an empty list.
+	StoppedProcessKeys      []string `json:"stopped_process_keys,omitempty"`
+	ReportsStoppedProcesses bool     `json:"reports_stopped_processes,omitempty"`
 }
 
 // SubjectNodeBackendList queries a worker node for its installed backends.
@@ -290,6 +298,23 @@ type BackendDeleteRequest struct {
 type BackendDeleteReply struct {
 	Success bool   `json:"success"`
 	Error   string `json:"error,omitempty"`
+
+	// StoppedProcessKeys names every `modelID#replica` process the worker
+	// terminated while serving this delete. Stopping a process returns its gRPC
+	// port to the worker's allocator, so any NodeModel row still pointing at
+	// that address becomes a live misroute the moment an unrelated backend
+	// binds the recycled port: probeHealth verifies liveness, not identity, so
+	// the request is served by the wrong backend rather than failing. The
+	// controller uses these keys to drop the rows eagerly.
+	StoppedProcessKeys []string `json:"stopped_process_keys,omitempty"`
+
+	// ReportsStoppedProcesses distinguishes "this worker enumerates what it
+	// stopped and stopped nothing" from "this worker predates the field". Both
+	// send an empty list, and only the first is authoritative. Without this
+	// flag a controller cannot tell them apart and would eventually be tempted
+	// to read silence as a completed cleanup, which is precisely the wrong
+	// conclusion against an older worker.
+	ReportsStoppedProcesses bool `json:"reports_stopped_processes,omitempty"`
 }
 
 // SubjectNodeModelUnload tells a worker node to unload a model (gRPC Free) without killing the backend.

core/services/nodes/unloader.go

@@ -270,6 +270,9 @@ func (a *RemoteUnloaderAdapter) UpgradeBackend(nodeID, backendType, galleriesJSO
 		return nil, fmt.Errorf("%w (subject=%s nodeID=%s backend=%s): %v",
 			galleryop.ErrWorkerStillInstalling, subject, nodeID, backendType, err)
 	}
+	if err == nil {
+		a.dropStoppedReplicaRows(nodeID, "backend.upgrade", backendType, reply.StoppedProcessKeys, reply.ReportsStoppedProcesses)
+	}
 	return reply, err
 }
 
@@ -337,7 +340,55 @@ func (a *RemoteUnloaderAdapter) DeleteBackend(nodeID, backendName string) (*mess
 	subject := messaging.SubjectNodeBackendDelete(nodeID)
 	xlog.Info("Sending NATS backend.delete", "nodeID", nodeID, "backend", backendName)
 
-	return messaging.RequestJSON[messaging.BackendDeleteRequest, messaging.BackendDeleteReply](a.nats, subject, messaging.BackendDeleteRequest{Backend: backendName}, 2*time.Minute)
+	reply, err := messaging.RequestJSON[messaging.BackendDeleteRequest, messaging.BackendDeleteReply](a.nats, subject, messaging.BackendDeleteRequest{Backend: backendName}, 2*time.Minute)
+	if err != nil {
+		return reply, err
+	}
+	a.dropStoppedReplicaRows(nodeID, "backend.delete", backendName, reply.StoppedProcessKeys, reply.ReportsStoppedProcesses)
+	return reply, nil
+}
+
+// dropStoppedReplicaRows removes the NodeModel rows addressing processes a
+// worker just terminated.
+//
+// Why eagerly, rather than leaving it to the existing health checks: stopping a
+// process returns its gRPC port to the worker's allocator, and the next backend
+// started there can be handed that same port. Until the row is gone it names a
+// live address, so both SmartRouter.probeHealth and the HealthMonitor per-model
+// probe — which verify liveness, not identity — pass against whatever now
+// occupies the port, and the request is served by the wrong backend rather than
+// failing. Nothing else on the delete/upgrade path tells the controller the
+// address just became invalid, unlike model.unload which drops its rows itself.
+//
+// reported=false means the worker predates this reply field. Its empty list is
+// then indistinguishable from "stopped nothing", so it must NOT be read as a
+// completed cleanup: leave the rows alone and fall back to the probe-based
+// staleness recovery that was the only mechanism before this change.
+func (a *RemoteUnloaderAdapter) dropStoppedReplicaRows(nodeID, op, backendName string, processKeys []string, reported bool) {
+	if !reported {
+		xlog.Debug("Worker did not report stopped processes; relying on probe-based staleness recovery",
+			"nodeID", nodeID, "op", op, "backend", backendName)
+		return
+	}
+	ctx := context.Background()
+	for _, key := range processKeys {
+		modelName, replicaIndex, ok := model.ParseBackendProcessKey(key)
+		if !ok {
+			// Acting on a guess could evict the row of a healthy sibling replica.
+			xlog.Warn("Ignoring unparseable process key reported by worker",
+				"nodeID", nodeID, "op", op, "backend", backendName, "processKey", key)
+			continue
+		}
+		xlog.Info("Dropping replica row for a process the worker stopped",
+			"nodeID", nodeID, "op", op, "backend", backendName, "model", modelName, "replica", replicaIndex)
+		if err := a.registry.RemoveNodeModel(ctx, nodeID, modelName, replicaIndex); err != nil {
+			// Best-effort: probe-based recovery remains the backstop, and failing
+			// the operator's delete over a bookkeeping error would be worse than
+			// the stale row this prevents.
+			xlog.Warn("Failed to drop replica row for a stopped process",
+				"nodeID", nodeID, "op", op, "model", modelName, "replica", replicaIndex, "error", err)
+		}
+	}
 }
 
 // UnloadModelOnNode sends a model.unload request to a specific node.

core/services/nodes/unloader_stale_rows_test.go

@@ -0,0 +1,186 @@
+package nodes
+
+import (
+	"encoding/json"
+	"time"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+
+	"github.com/mudler/LocalAI/core/services/messaging"
+)
+
+// Replies are handed to the adapter as raw JSON rather than as marshalled
+// structs on purpose: these specs pin the behavior a controller must exhibit
+// when a worker on a *different* build answers it. Marshalling the current
+// struct would only ever produce the current field set and could not express
+// an old worker's reply at all.
+var _ = Describe("RemoteUnloaderAdapter stale replica rows", func() {
+	var (
+		locator *fakeModelLocator
+		mc      *fakeMessagingClient
+		adapter *RemoteUnloaderAdapter
+	)
+
+	BeforeEach(func() {
+		locator = &fakeModelLocator{}
+		mc = &fakeMessagingClient{}
+		adapter = NewRemoteUnloaderAdapter(locator, mc, 3*time.Minute, 15*time.Minute)
+	})
+
+	Describe("DeleteBackend", func() {
+		It("drops the replica rows for every process the worker stopped", func() {
+			// The worker has just terminated these processes and returned their
+			// gRPC ports to its allocator. Any row still pointing at those
+			// addresses will pass probeHealth as soon as an unrelated backend
+			// binds the recycled port, and the request is then silently served
+			// by the wrong backend.
+			mc.requestReply = []byte(`{
+				"success": true,
+				"reports_stopped_processes": true,
+				"stopped_process_keys": ["qwen3-0.6b#0", "qwen3-0.6b#2"]
+			}`)
+
+			reply, err := adapter.DeleteBackend("node-1", "llama-cpp")
+			Expect(err).NotTo(HaveOccurred())
+			Expect(reply.Success).To(BeTrue())
+
+			Expect(locator.removedReplicas).To(ConsistOf(
+				modelReplicaRef{"node-1", "qwen3-0.6b", 0},
+				modelReplicaRef{"node-1", "qwen3-0.6b", 2},
+			))
+		})
+
+		It("keeps model IDs that contain the replica separator intact", func() {
+			// Process keys are `modelID#replica` and model IDs are free to
+			// contain '#' themselves, so the split must be anchored at the last
+			// separator. Splitting at the first one addresses a row that does
+			// not exist and leaves the real stale row in place.
+			mc.requestReply = []byte(`{
+				"success": true,
+				"reports_stopped_processes": true,
+				"stopped_process_keys": ["weird#name#3"]
+			}`)
+
+			_, err := adapter.DeleteBackend("node-1", "llama-cpp")
+			Expect(err).NotTo(HaveOccurred())
+			Expect(locator.removedReplicas).To(ConsistOf(modelReplicaRef{"node-1", "weird#name", 3}))
+		})
+
+		It("removes nothing when a worker that reports stopped processes stopped none", func() {
+			// Deleting a backend that was never loaded is routine. The reply is
+			// authoritative here, so "no keys" genuinely means "no rows".
+			mc.requestReply = []byte(`{"success": true, "reports_stopped_processes": true}`)
+
+			_, err := adapter.DeleteBackend("node-1", "llama-cpp")
+			Expect(err).NotTo(HaveOccurred())
+			Expect(locator.removedReplicas).To(BeEmpty())
+		})
+
+		It("does not read an old worker's silence as a completed cleanup", func() {
+			// A worker predating this change never sets either field. Its empty
+			// list is indistinguishable from "stopped nothing", so the
+			// controller must not conclude the node is clean, and equally must
+			// not guess at rows to delete. It falls back to the probe-based
+			// self-heal in SmartRouter.probeHealth, which is exactly the
+			// pre-change behavior.
+			mc.requestReply = []byte(`{"success": true}`)
+
+			reply, err := adapter.DeleteBackend("node-1", "llama-cpp")
+			Expect(err).NotTo(HaveOccurred())
+			Expect(reply.Success).To(BeTrue())
+			Expect(reply.ReportsStoppedProcesses).To(BeFalse())
+			Expect(locator.removedReplicas).To(BeEmpty())
+			Expect(locator.removedPairs).To(BeEmpty())
+		})
+
+		It("leaves rows alone when the worker could not stop the process", func() {
+			// The worker aborts the delete without listing the key it failed to
+			// kill: that process survived, so its address is still correct and
+			// dropping the row would force a needless reload of a live replica.
+			mc.requestReply = []byte(`{
+				"success": false,
+				"error": "could not stop running process qwen3-0.6b#0",
+				"reports_stopped_processes": true
+			}`)
+
+			reply, err := adapter.DeleteBackend("node-1", "llama-cpp")
+			Expect(err).NotTo(HaveOccurred())
+			Expect(reply.Success).To(BeFalse())
+			Expect(locator.removedReplicas).To(BeEmpty())
+		})
+
+		It("drops rows for processes that died before a later step failed", func() {
+			// The worker reports a key only once that process is confirmed gone,
+			// so a failure further along (removing files, re-registering) does
+			// not make the already-recycled ports any less dangerous. Gating
+			// removal on overall success would strand exactly those rows.
+			mc.requestReply = []byte(`{
+				"success": false,
+				"error": "failed to delete backend files",
+				"reports_stopped_processes": true,
+				"stopped_process_keys": ["qwen3-0.6b#0"]
+			}`)
+
+			_, err := adapter.DeleteBackend("node-1", "llama-cpp")
+			Expect(err).NotTo(HaveOccurred())
+			Expect(locator.removedReplicas).To(ConsistOf(modelReplicaRef{"node-1", "qwen3-0.6b", 0}))
+		})
+
+		It("ignores malformed process keys instead of removing a wrong row", func() {
+			mc.requestReply = []byte(`{
+				"success": true,
+				"reports_stopped_processes": true,
+				"stopped_process_keys": ["no-replica-suffix", "qwen3-0.6b#notanumber", "good#1"]
+			}`)
+
+			_, err := adapter.DeleteBackend("node-1", "llama-cpp")
+			Expect(err).NotTo(HaveOccurred())
+			Expect(locator.removedReplicas).To(ConsistOf(modelReplicaRef{"node-1", "good", 1}))
+		})
+	})
+
+	Describe("UpgradeBackend", func() {
+		It("drops the replica rows for every process the worker stopped", func() {
+			// An upgrade force-stops every process using the binary and starts
+			// none of them back up, so it recycles ports exactly as delete does
+			// while leaving the same rows behind.
+			mc.requestReply = []byte(`{
+				"success": true,
+				"reports_stopped_processes": true,
+				"stopped_process_keys": ["whisper#0", "whisper#1"]
+			}`)
+
+			reply, err := adapter.UpgradeBackend("node-1", "whisper", "", "", "", "", 0, "", nil)
+			Expect(err).NotTo(HaveOccurred())
+			Expect(reply.Success).To(BeTrue())
+
+			Expect(locator.removedReplicas).To(ConsistOf(
+				modelReplicaRef{"node-1", "whisper", 0},
+				modelReplicaRef{"node-1", "whisper", 1},
+			))
+		})
+
+		It("does not read an old worker's silence as a completed cleanup", func() {
+			mc.requestReply = []byte(`{"success": true}`)
+
+			reply, err := adapter.UpgradeBackend("node-1", "whisper", "", "", "", "", 0, "", nil)
+			Expect(err).NotTo(HaveOccurred())
+			Expect(reply.ReportsStoppedProcesses).To(BeFalse())
+			Expect(locator.removedReplicas).To(BeEmpty())
+		})
+	})
+
+	Describe("reply wire compatibility", func() {
+		It("decodes an old worker's delete reply without the new fields", func() {
+			// Guards the rolling-upgrade direction that matters: a new
+			// controller must keep working against every worker already
+			// deployed, not just ones rebuilt from this commit.
+			var reply messaging.BackendDeleteReply
+			Expect(json.Unmarshal([]byte(`{"success": true}`), &reply)).To(Succeed())
+			Expect(reply.Success).To(BeTrue())
+			Expect(reply.ReportsStoppedProcesses).To(BeFalse())
+			Expect(reply.StoppedProcessKeys).To(BeEmpty())
+		})
+	})
+})

core/services/nodes/unloader_test.go

@@ -20,22 +20,35 @@ import (
 
 // fakeModelLocator implements ModelLocator with configurable node lists.
 type fakeModelLocator struct {
-	nodes        []BackendNode
-	findErr      error
-	removedPairs []modelNodePair // records RemoveNodeModel calls
+	nodes           []BackendNode
+	findErr         error
+	removedPairs    []modelNodePair   // records RemoveNodeModel calls
+	removedReplicas []modelReplicaRef // records RemoveNodeModel calls including the replica index
 }
 
 type modelNodePair struct {
 	nodeID    string
 	modelName string
 }
 
+// modelReplicaRef records a row removal at full replica granularity.
+// modelNodePair drops the index because RemoveAllNodeModelReplicas has none;
+// the backend delete/upgrade paths address exactly one replica row, so an
+// assertion that ignored the index could not tell a correct removal from one
+// that wiped a sibling replica still serving traffic.
+type modelReplicaRef struct {
+	nodeID       string
+	modelName    string
+	replicaIndex int
+}
+
 func (f *fakeModelLocator) FindNodesWithModel(_ context.Context, _ string) ([]BackendNode, error) {
 	return f.nodes, f.findErr
 }
 
-func (f *fakeModelLocator) RemoveNodeModel(_ context.Context, nodeID, modelName string, _ int) error {
+func (f *fakeModelLocator) RemoveNodeModel(_ context.Context, nodeID, modelName string, replicaIndex int) error {
 	f.removedPairs = append(f.removedPairs, modelNodePair{nodeID, modelName})
+	f.removedReplicas = append(f.removedReplicas, modelReplicaRef{nodeID, modelName, replicaIndex})
 	return nil
 }