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) {