From 1cf847f29ef92f29c940558070c9908b00490b36 Mon Sep 17 00:00:00 2001 From: Ettore Di Giacinto Date: Tue, 1 Sep 2026 18:47:37 +0000 Subject: [PATCH] feat(distributed): stop workers listening, and stop them advertising A worker now opens no listener on a routable interface and states no endpoint at registration. Backend processes and the file-transfer server bind loopback, and the frontend reaches both through the tunnel the worker dials. The bind address is built from loopbackHost, the same constant the tunnel's grpc tag dials, so "the worker binds where its tunnel dials" is one fact in one place rather than two literals that can drift. All three advertisement sites are closed, not one: the registration body, RegisterNodeRequest, and the per-backend address in the install reply. That third one was hiding a live bug. stopModelExact refuses a stop whose ExpectedAddress does not match what the worker recorded for the process. The worker recorded 127.0.0.1:port; handleBackendInstall reported advertiseHost:port; the router stored the reported one and sent it straight back. On any worker whose advertise host was not 127.0.0.1, every acknowledged model stop failed with an address mismatch. Nothing caught it because the e2e harness set LOCALAI_ADVERTISE_ADDR=127.0.0.1, which made the rewrite a no-op. Removing the rewrite makes the two strings the same by construction. The brief was wrong about two of the four functions it called dead. effectiveBasePort is the base of the backend port allocator and resolveHTTPAddr is the file server's bind address; deleting them would have deleted the port allocator and the file server. Only the two advertise* helpers were dead, and addr_test.go is rewritten rather than deleted, because the port arithmetic it pinned still needs pinning. NodeModel.Address survives with a narrowed meaning and is renamed WorkerLocalAddress, along with the install reply field that feeds it. The frontend still has to say WHICH backend process on a worker it means, and the port in this string is how it says it: it travels as a stream target and the worker dials its own loopback. The gorm column and the json key stay "address", so neither a migration nor an API break rides along. Every fall-back to the node's address is gone. installBackendOnNode now errors when a worker reports success without naming one, because substituting the now-always-empty node address would name an empty target, and the worker refuses that as an invalid stream, which is classified as the worker answering about its backend. That is the "a present worker reads as something it is not" class this phase forbids. DistributedModelStore.Range had the same shape and was already wrong: it built each remote model's client from the node's base gRPC port, never the port a backend process listens on, so Free and Status went to the wrong place. It uses the replica's address now. BackendNode.Address and HTTPAddress are kept but made provably inert: no writer, no reader that acts on them, and Register force-clears both on re-registration so an upgraded worker's stale advertisement does not outlive its own upgrade in the API and the Nodes page. Dropping the columns is a ~90-site edit across the specs, the e2e suite, the MCP dto and the UI; it is recorded as a follow-up rather than folded in here. A persistent tunnel 401 still does not trigger re-registration, and now for a reason rather than a deferral. Register CLEARS the node's replica rows, so re-registering on a 401 would delete a live worker's rows on every retry, and under the name collision that causes the 401 the two workers would take turns doing it forever: a credential failure causing model reclamation. It also cannot fix the named cause, since a collision is indistinguishable from a restart. The 401 log now names both causes and says nothing can reach this worker, which is true only now that it has no listener. The container healthcheck did not break the way the brief expected, since the listener still exists on loopback and the probe runs inside the container. It did have a real #10987 defect that this change makes the common case: it read LOCALAI_SERVE_ADDR only, while effectiveBasePort reads LOCALAI_ADDR first, so a worker on a non-default base port was probed on 50050 and reported unhealthy while working. It follows the same precedence now. Docs, the compose file and the e2e harness are updated in step: no inbound rule or published port is needed for a worker, the two advertise variables are gone, the remaining address variables are read for their port only, the firewall-the-file-transfer-port warning is narrowed to the LOCALAI_HTTP_ADDR opt-out, and the upgrade-order note no longer claims the worker still listens. The Nodes page showed node.address, which is now always blank, so it shows the node id instead. Eight mutations, all red on a named spec, including reverting the loopback bind, re-adding the address to the registration body, restoring both node-address fall-backs, dropping the force-clear, storing the endpoint's address again, and un-fixing the healthcheck. One of them caught a defect in a spec I had just written: it asserted 200 where the endpoint returns 201, which went unnoticed because core/http/endpoints/localai is not on the task's verify list. It is run here. Assisted-by: Claude Opus 5 [claude-code] Signed-off-by: Ettore Di Giacinto --- core/http/endpoints/localai/nodes.go | 27 ++--- core/http/endpoints/localai/nodes_test.go | 40 +++++-- .../src/components/nodes/NodePanel.jsx | 6 +- core/http/react-ui/src/pages/NodeDetail.jsx | 2 +- core/services/messaging/subjects.go | 15 ++- core/services/nodes/disk_headroom_test.go | 2 +- core/services/nodes/distributed_store.go | 16 ++- core/services/nodes/distributed_store_test.go | 51 +++++++-- core/services/nodes/health.go | 16 ++- core/services/nodes/health_mock_test.go | 12 ++ core/services/nodes/health_test.go | 14 +-- .../nodes/managers_distributed_test.go | 16 +-- core/services/nodes/model_router.go | 4 +- core/services/nodes/model_router_test.go | 14 ++- core/services/nodes/reconciler.go | 12 +- .../nodes/reconciler_busy_probe_test.go | 14 +-- .../nodes/reconciler_inflight_leak_test.go | 16 +-- core/services/nodes/reconciler_test.go | 36 +++--- .../nodes/reconciler_worker_processes_test.go | 14 +-- core/services/nodes/registry.go | 64 ++++++++--- core/services/nodes/registry_test.go | 36 +++++- .../nodes/revision_eligibility_test.go | 6 +- core/services/nodes/router.go | 58 +++++++--- .../nodes/router_eviction_alias_test.go | 2 +- .../nodes/router_eviction_selector_test.go | 2 +- .../services/nodes/router_load_budget_test.go | 2 +- core/services/nodes/router_load_job_test.go | 2 +- .../nodes/router_load_timeout_test.go | 2 +- core/services/nodes/router_reap_load_test.go | 2 +- .../services/nodes/router_reservation_test.go | 2 +- .../nodes/router_revision_lifecycle_test.go | 2 +- .../nodes/router_staging_context_test.go | 4 +- .../nodes/router_staging_deadline_test.go | 4 +- core/services/nodes/router_test.go | 70 +++++++++--- .../nodes/router_unreachable_worker_test.go | 4 +- core/services/nodes/unloader.go | 2 +- core/services/nodes/unloader_test.go | 6 +- core/services/worker/addr_test.go | 103 ++++++++++-------- core/services/worker/config.go | 22 ++-- core/services/worker/lifecycle.go | 26 ++--- core/services/worker/registration.go | 51 +++------ core/services/worker/supervisor.go | 39 +++++-- core/services/worker/tunnel.go | 9 +- core/services/worker/worker.go | 25 +++-- docker-compose.distributed.yaml | 21 ++-- docs/content/features/distributed-mode.md | 53 +++++---- scripts/build/healthcheck.sh | 23 +++- scripts/build/healthcheck_test.sh | 15 +++ tests/e2e/distributed/cluster/cluster.go | 6 +- .../distributed/model_config_revision_test.go | 2 +- 50 files changed, 635 insertions(+), 357 deletions(-) diff --git a/core/http/endpoints/localai/nodes.go b/core/http/endpoints/localai/nodes.go index f404555ac..0571e7d7f 100644 --- a/core/http/endpoints/localai/nodes.go +++ b/core/http/endpoints/localai/nodes.go @@ -77,10 +77,14 @@ func GetNodeEndpoint(registry *nodes.NodeRegistry) echo.HandlerFunc { // RegisterNodeRequest is the request body for registering a new worker node. type RegisterNodeRequest struct { - Name string `json:"name"` - NodeType string `json:"node_type,omitempty"` // "backend" (default) or "agent" - Address string `json:"address"` - HTTPAddress string `json:"http_address,omitempty"` + Name string `json:"name"` + NodeType string `json:"node_type,omitempty"` // "backend" (default) or "agent" + // No address and no http_address. A worker has no inbound endpoint to + // register: it holds one outbound tunnel to a frontend replica and every + // protocol the frontend speaks to it travels on that. An older worker still + // sends both keys and they are ignored, which is the intended outcome: + // storing them would put a dialable-looking endpoint back in the API for + // something nothing dials. Token string `json:"token,omitempty"` TotalVRAM uint64 `json:"total_vram,omitempty"` AvailableVRAM uint64 `json:"available_vram,omitempty"` @@ -142,22 +146,15 @@ func RegisterNodeEndpoint(registry *nodes.NodeRegistry, expectedToken string, au fmt.Sprintf("invalid node_type %q; must be %q or %q", nodeType, nodes.NodeTypeBackend, nodes.NodeTypeAgent))) } - // Backend workers require address; agent workers don't serve gRPC + // A backend worker no longer has to state an address; the tunnel it + // dials is what makes it reachable, and requiring one here would refuse + // exactly the workers this design is for. if req.Name == "" { return c.JSON(http.StatusBadRequest, nodeError(http.StatusBadRequest, "name is required")) } - if nodeType == nodes.NodeTypeBackend && req.Address == "" { - return c.JSON(http.StatusBadRequest, nodeError(http.StatusBadRequest, "address is required for backend workers")) - } if len(req.Name) > 255 { return c.JSON(http.StatusBadRequest, nodeError(http.StatusBadRequest, "name exceeds 255 characters")) } - if len(req.Address) > 512 { - return c.JSON(http.StatusBadRequest, nodeError(http.StatusBadRequest, "address exceeds 512 characters")) - } - if len(req.HTTPAddress) > 512 { - return c.JSON(http.StatusBadRequest, nodeError(http.StatusBadRequest, "http_address exceeds 512 characters")) - } // Hash the token for storage (if provided) var tokenHash string @@ -177,8 +174,6 @@ func RegisterNodeEndpoint(registry *nodes.NodeRegistry, expectedToken string, au node := &nodes.BackendNode{ Name: req.Name, NodeType: nodeType, - Address: req.Address, - HTTPAddress: req.HTTPAddress, TokenHash: tokenHash, TotalVRAM: req.TotalVRAM, AvailableVRAM: req.AvailableVRAM, diff --git a/core/http/endpoints/localai/nodes_test.go b/core/http/endpoints/localai/nodes_test.go index dababff38..2255116ab 100644 --- a/core/http/endpoints/localai/nodes_test.go +++ b/core/http/endpoints/localai/nodes_test.go @@ -289,7 +289,11 @@ var _ = Describe("Node HTTP handlers", func() { Expect(errObj["message"]).To(ContainSubstring("exceeds 255 characters")) }) - It("returns 400 when address is missing for backend node type", func() { + It("registers a backend worker that states no address", func() { + // This used to be a 400. It is the shape every worker now + // registers with: it has no inbound endpoint, it holds one outbound + // tunnel, and refusing it here would refuse exactly the workers the + // tunnel exists for. e := echo.New() body := `{"name":"worker-no-addr"}` req := httptest.NewRequest(http.MethodPost, "/", strings.NewReader(body)) @@ -299,13 +303,35 @@ var _ = Describe("Node HTTP handlers", func() { handler := RegisterNodeEndpoint(registry, "", true, nil, "", natsauth.Config{}) Expect(handler(c)).To(Succeed()) - Expect(rec.Code).To(Equal(http.StatusBadRequest)) + Expect(rec.Code).To(Equal(http.StatusCreated)) - var resp map[string]any - Expect(json.Unmarshal(rec.Body.Bytes(), &resp)).To(Succeed()) - errObj, ok := resp["error"].(map[string]any) - Expect(ok).To(BeTrue()) - Expect(errObj["message"]).To(ContainSubstring("address is required")) + stored, err := registry.GetByName(context.Background(), "worker-no-addr") + Expect(err).ToNot(HaveOccurred()) + Expect(stored.NodeType).To(Equal(nodes.NodeTypeBackend)) + Expect(stored.Address).To(BeEmpty()) + Expect(stored.HTTPAddress).To(BeEmpty()) + }) + + It("stores no address even when a worker still sends one", func() { + // An older worker keeps sending both keys. Storing them would put a + // dialable-looking endpoint back into the API and the Nodes page for + // something nothing dials, and would leave a reader of either one + // unsure which workers are reached how. + e := echo.New() + body := `{"name":"worker-legacy-addr","address":"10.0.0.9:50051","http_address":"10.0.0.9:50050"}` + req := httptest.NewRequest(http.MethodPost, "/", strings.NewReader(body)) + req.Header.Set(echo.HeaderContentType, echo.MIMEApplicationJSON) + rec := httptest.NewRecorder() + c := e.NewContext(req, rec) + + handler := RegisterNodeEndpoint(registry, "", true, nil, "", natsauth.Config{}) + Expect(handler(c)).To(Succeed()) + Expect(rec.Code).To(Equal(http.StatusCreated)) + + stored, err := registry.GetByName(context.Background(), "worker-legacy-addr") + Expect(err).ToNot(HaveOccurred()) + Expect(stored.Address).To(BeEmpty()) + Expect(stored.HTTPAddress).To(BeEmpty()) }) It("returns 400 when node_type is invalid", func() { diff --git a/core/http/react-ui/src/components/nodes/NodePanel.jsx b/core/http/react-ui/src/components/nodes/NodePanel.jsx index 623db0093..b73714ab9 100644 --- a/core/http/react-ui/src/components/nodes/NodePanel.jsx +++ b/core/http/react-ui/src/components/nodes/NodePanel.jsx @@ -19,7 +19,11 @@ export default function NodePanel({ node, models = [], onApprove, onDrain, onRes
{node.name} - {node.address} + {/* A worker has no address to show: it holds an outbound tunnel and + binds nothing routable. Its id is what identifies it in routing + logs, so that is what an operator needs here. Pre-tunnel nodes + may still carry an address until they re-register. */} + {node.address || node.id}
e.stopPropagation()}> {node.status === 'pending' && ( diff --git a/core/http/react-ui/src/pages/NodeDetail.jsx b/core/http/react-ui/src/pages/NodeDetail.jsx index bff7db526..52d3fea1d 100644 --- a/core/http/react-ui/src/pages/NodeDetail.jsx +++ b/core/http/react-ui/src/pages/NodeDetail.jsx @@ -78,7 +78,7 @@ export default function NodeDetail() { navigate('/app/nodes')} className="link-plain">