fix(worker): reap deleted backends and stop models that live on a worker

core/http/endpoints/localai/backend_monitor.go

@@ -1,9 +1,15 @@
 package localai
 
 import (
+	"errors"
+	"fmt"
+	"net/http"
+
 	"github.com/labstack/echo/v4"
 	"github.com/mudler/LocalAI/core/schema"
 	"github.com/mudler/LocalAI/core/services/monitoring"
+	"github.com/mudler/LocalAI/pkg/model"
+	"github.com/mudler/xlog"
 )
 
 // BackendMonitorEndpoint returns the status of the specified backend
@@ -48,7 +54,25 @@ func BackendShutdownEndpoint(bm *monitoring.BackendMonitorService) echo.HandlerF
 		if err := c.Bind(input); err != nil {
 			return err
 		}
+		if input.Model == "" {
+			return echo.NewHTTPError(http.StatusBadRequest, "model is required")
+		}
 
-		return bm.ShutdownModel(input.Model)
+		if err := bm.ShutdownModel(input.Model); err != nil {
+			// "Not loaded" is a client-side condition, not a server fault, so
+			// it should not surface as a 500. In distributed mode this branch
+			// used to be reached for models that were running on a worker —
+			// only this replica's local store was empty. The loader now
+			// consults the node registry first, so reaching it means the
+			// model is loaded neither here nor anywhere in the cluster.
+			if errors.Is(err, model.ErrModelNotFound) {
+				xlog.Info("Shutdown requested for a model that is not loaded", "model", input.Model)
+				return echo.NewHTTPError(http.StatusNotFound,
+					fmt.Sprintf("model %q is not loaded on this instance or on any worker node", input.Model))
+			}
+			xlog.Error("Failed to shut down model", "model", input.Model, "error", err)
+			return err
+		}
+		return c.JSON(http.StatusOK, map[string]string{"message": "model stopped"})
 	}
 }

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

@@ -12,6 +12,7 @@ import (
 
 	"github.com/mudler/LocalAI/core/services/galleryop"
 	"github.com/mudler/LocalAI/core/services/messaging"
+	"github.com/mudler/LocalAI/pkg/model"
 	"github.com/mudler/xlog"
 )
 
@@ -71,6 +72,17 @@ func (a *RemoteUnloaderAdapter) InstallTimeout() time.Duration {
 	return a.installTimeout
 }
 
+// Compile-time proof that the adapter still satisfies the loader's optional
+// extensions. Both are consumed via runtime type assertion in deleteProcess, so
+// a signature drift here would silently downgrade behavior — losing force
+// propagation, or making ShutdownModel unable to tell a cluster-wide miss from
+// a completed unload — rather than failing the build.
+var (
+	_ model.RemoteModelUnloader        = (*RemoteUnloaderAdapter)(nil)
+	_ model.RemoteModelContextUnloader = (*RemoteUnloaderAdapter)(nil)
+	_ model.RemoteModelPresenceChecker = (*RemoteUnloaderAdapter)(nil)
+)
+
 // UnloadRemoteModel finds the node(s) hosting the given model and tells them
 // to stop their backend process via NATS backend.stop event.
 // The worker process handles a bounded Free() followed by process termination;
@@ -80,6 +92,22 @@ func (a *RemoteUnloaderAdapter) UnloadRemoteModel(modelName string) error {
 	return a.UnloadRemoteModelContext(context.Background(), modelName, false)
 }
 
+// HasRemoteModel reports whether any node currently holds the model. It exists
+// because UnloadRemoteModel is idempotent and so cannot signal "there was
+// nothing to stop"; ShutdownModel consults this first so it can answer 404 for
+// a model loaded neither locally nor anywhere in the cluster, instead of the
+// misleading 500 "model not found" that a local-store miss used to produce.
+func (a *RemoteUnloaderAdapter) HasRemoteModel(ctx context.Context, modelName string) (bool, error) {
+	if ctx == nil {
+		ctx = context.Background()
+	}
+	nodes, err := a.registry.FindNodesWithModel(ctx, modelName)
+	if err != nil {
+		return false, fmt.Errorf("finding nodes with model %q: %w", modelName, err)
+	}
+	return len(nodes) > 0, nil
+}
+
 // UnloadRemoteModelContext is the cancellation-aware extension used by the
 // model loader to preserve forced shutdown across the distributed boundary.
 func (a *RemoteUnloaderAdapter) UnloadRemoteModelContext(ctx context.Context, modelName string, force bool) error {
@@ -91,6 +119,10 @@ func (a *RemoteUnloaderAdapter) UnloadRemoteModelContext(ctx context.Context, mo
 		return fmt.Errorf("finding nodes with model %q: %w", modelName, err)
 	}
 	if len(nodes) == 0 {
+		// Unloading is idempotent by contract: cleanup paths (model deletion,
+		// config edits, watchdog eviction) legitimately run against an
+		// already-unloaded model and must not fail. Callers that need to tell
+		// this case apart use HasRemoteModel before unloading.
 		xlog.Debug("No remote nodes found with model", "model", modelName)
 		return nil
 	}
@@ -238,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
 }
 
@@ -305,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())
+		})
+	})
+})