From bebd812e7dbf520cfcc620135cb106d1cca1ec8f Mon Sep 17 00:00:00 2001 From: Ettore Di Giacinto Date: Mon, 24 Aug 2026 12:48:23 +0000 Subject: [PATCH] fix(distributed): stop flapping agent nodes on backend listing Only backend workers subscribe to backend.list. ListBackends asked every node that was not pending, offline or draining, so an agent worker could only answer "no responders", which the error handling reads as a node that has gone away. Every poll of the backends view therefore marked each agent node unhealthy, and its next heartbeat marked it healthy again. While unhealthy the node is not schedulable, so this also cost agent capacity for as long as each flap lasted. Skip non-backend workers, as the backend-op fan-out already does for the same reason. A backend worker that does not answer is still marked unhealthy: that one really is gone. Signed-off-by: Ettore Di Giacinto Assisted-by: Claude Code:claude-opus-5 [golangci-lint] --- .../nodes/managers_agent_node_test.go | 84 +++++++++++++++++++ core/services/nodes/managers_distributed.go | 13 ++- 2 files changed, 95 insertions(+), 2 deletions(-) create mode 100644 core/services/nodes/managers_agent_node_test.go diff --git a/core/services/nodes/managers_agent_node_test.go b/core/services/nodes/managers_agent_node_test.go new file mode 100644 index 000000000..8ee95c083 --- /dev/null +++ b/core/services/nodes/managers_agent_node_test.go @@ -0,0 +1,84 @@ +package nodes + +import ( + "context" + "runtime" + "time" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + "gorm.io/gorm" + + "github.com/mudler/LocalAI/core/services/messaging" + "github.com/mudler/LocalAI/core/services/testutil" +) + +// Agent workers do not subscribe to the backend.* subjects, so asking one to +// list its backends can only answer "no responders". ListBackends read that as +// a node that had gone away and marked it unhealthy; the node's next heartbeat +// marked it healthy again. Every poll of the backends view therefore flapped +// every agent node in the cluster, and while it was unhealthy the router would +// not schedule onto it. +var _ = Describe("Backend listing across mixed node types", func() { + var ( + db *gorm.DB + registry *NodeRegistry + mc *scriptedMessagingClient + mgr *DistributedBackendManager + ctx context.Context + ) + + 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()) + mc = newScriptedMessagingClient() + mgr = &DistributedBackendManager{ + local: stubLocalBackendManager{}, + adapter: NewRemoteUnloaderAdapter(nil, mc, 3*time.Minute, 15*time.Minute), + registry: registry, + } + ctx = context.Background() + }) + + register := func(name, nodeType string) *BackendNode { + node := &BackendNode{Name: name, NodeType: nodeType, Address: name + ":50051"} + Expect(registry.Register(ctx, node, true)).To(Succeed()) + fetched, err := registry.GetByName(ctx, name) + Expect(err).ToNot(HaveOccurred()) + Expect(fetched.Status).To(Equal(StatusHealthy)) + return fetched + } + + statusOf := func(id string) string { + n, err := registry.Get(ctx, id) + Expect(err).ToNot(HaveOccurred()) + return n.Status + } + + It("leaves an agent node healthy instead of flapping it", func() { + agent := register("agent-worker-1", NodeTypeAgent) + mc.scriptNoResponders(messaging.SubjectNodeBackendList(agent.ID)) + + _, err := mgr.ListBackends() + Expect(err).ToNot(HaveOccurred()) + + Expect(statusOf(agent.ID)).To(Equal(StatusHealthy), + "an agent node cannot answer backend.list and must not be judged on it") + }) + + It("still marks a backend node unhealthy when it does not answer", func() { + backendNode := register("worker-a", NodeTypeBackend) + mc.scriptNoResponders(messaging.SubjectNodeBackendList(backendNode.ID)) + + _, err := mgr.ListBackends() + Expect(err).ToNot(HaveOccurred()) + + Expect(statusOf(backendNode.ID)).To(Equal(StatusUnhealthy), + "a backend worker that does not answer is genuinely gone") + }) +}) diff --git a/core/services/nodes/managers_distributed.go b/core/services/nodes/managers_distributed.go index 127425b1a..4132eca79 100644 --- a/core/services/nodes/managers_distributed.go +++ b/core/services/nodes/managers_distributed.go @@ -331,8 +331,9 @@ func (d *DistributedBackendManager) DeleteBackendDetailed(ctx context.Context, n // populated from the first node seen so single-node-minded callers still work. // // Pending/offline/draining nodes are skipped because they aren't expected to -// answer NATS requests; unhealthy nodes are still queried — ErrNoResponders -// then marks them unhealthy and the loop continues. +// answer NATS requests, and so are non-backend workers, which do not subscribe +// to backend.list at all; unhealthy backend nodes are still queried — +// ErrNoResponders then marks them unhealthy and the loop continues. func (d *DistributedBackendManager) ListBackends() (gallery.SystemBackends, error) { result := make(gallery.SystemBackends) allNodes, err := d.registry.List(context.Background()) @@ -344,6 +345,14 @@ func (d *DistributedBackendManager) ListBackends() (gallery.SystemBackends, erro if node.Status == StatusPending || node.Status == StatusOffline || node.Status == StatusDraining { continue } + // Only backend workers subscribe to backend.list. Asking an agent + // worker can only answer "no responders", which the error handling + // below reads as a node that has gone away, so every poll of this view + // marked every agent node unhealthy and its next heartbeat marked it + // healthy again. The backend-op fan-out skips them for the same reason. + if node.NodeType != "" && node.NodeType != NodeTypeBackend { + continue + } reply, err := d.adapter.ListBackends(node.ID) if err != nil { if errors.Is(err, nats.ErrNoResponders) {