From 3338d7bc56a15d87c8970e81a08668698fa46ff0 Mon Sep 17 00:00:00 2001 From: Ettore Di Giacinto Date: Tue, 1 Sep 2026 19:36:35 +0000 Subject: [PATCH] fix(distributed): refuse a worker that cannot tunnel, and say why it was refused Review round 1 on the change that stopped workers listening. One blocking item and seven notes. LOCALAI_WORKER_TUNNEL=false was the blocking one, and the ruling was to make it fatal rather than to correct the comment that still promised it fell back to the advertised address. There is no fallback left: a worker on this branch advertises nothing and binds only loopback, so turning the tunnel off leaves it reachable by nothing while it registers, heartbeats and reports healthy, and the scheduler keeps placing models on it. That is the worst available failure shape, so a new Config.validateStartup refuses it before prefetch, registration and NATS, while the worker is still invisible to the cluster. It absorbs the pre-existing empty-registration-token check, which had the same shape and no spec. The flag is kept rather than deleted so an operator who set it is told the promise is gone instead of having the setting ignored, and the guard around StartTunnel is removed, because a branch nothing can take reads as a supported no-tunnel mode that does not exist. The justification for erroring on an install that names no address was wrong, and the review is right that this is the dangerous form of overclaiming, because the conclusion holds and the mechanism does not. It said the resulting empty target would be refused as an invalid stream and that the refusal would read as the worker answering about its backend. Nothing in this repo branches on cluster.ErrNoRoute, and nodes.unroutable treats any recorded dial error as unroutable, so that refusal reaches every reap guard as ProbeUnknown and deletes nothing. The site now stands on what holds, that an install naming no port produced nothing routable and the failure belongs to the install rather than to a later probe, and records the retracted claim so nobody re-derives it. This retracts the same paragraph in the body of 1cf847f29. The reviewer deleted the whole tryWarmPath unnamed-replica guard and the suite stayed green, including the reservation release. It is specced now, and the asymmetry the review asked about is decided at the site: the row stays, unlike the sibling !alive branch which removes it. That branch has observed a backend dead; this one has observed only that the row is unreadable, which says nothing about whether a process is running, and the row is the last record that one might be, since the acknowledged stop path refuses a stop whose ExpectedAddress does not match and an empty one cannot be cleaned up through it either. The cross-version wire claim rested on two struct tags nobody asserted: renaming only the json keys survived mutation while the gorm column rename went red through raw SQL. Both keys are pinned now, marshal and unmarshal, per struct. A worker-first upgrade showed the operator a status code and not the reason. The registration client discarded the body, so "address is required for backend workers" was read off the socket and thrown away, and the ladder then spent four minutes on a verdict the frontend reached instantly. Refusals now quote the body and carry ErrRegistrationRejected, and both the ladder and the credential manager's Acquire stop on the first one. Acquire matters more than the ladder: it is the default path and its bound is 100 attempts, not 10. 408 and 429 are deliberately not refusals, since both are the frontend asking for the same request again. Also: the stale "not blocked by firewalls" troubleshooting line, which now names the real cause and the knobs that move the port range; and the inert address fields on the MCP Node DTO, which the Assistant was still being handed. The review named http_address there and I removed address too, because it is inert by the same argument and leaving one of a pair is arbitrary. Five mutations, all red. Deleting the warm-path guard reddens four specs and falsifying only its reservation release reddens one, so the two halves are pinned separately. Renaming only the json keys reddens both wire suites. Discarding the refusal body reddens two. Dropping the rejection classification does not fail the suite, it hangs it, which is the operator-visible symptom, so it is recorded red under a ginkgo timeout. The verify list is now derived from the diff rather than from the brief, which is what let the previous round ship a spec asserting 200 where the endpoint returns 201: nine ginkgo suites, the e2e vet, route auth coverage, the leaf check, build, the healthcheck shell suite and lint. The two jsx files have no harness in this worktree and are recorded as the one unverified surface. Assisted-by: Claude Opus 5 [claude-code] Signed-off-by: Ettore Di Giacinto --- core/cli/workerregistry/client.go | 62 +++++++- core/cli/workerregistry/client_test.go | 138 ++++++++++++++++++ core/cli/workerregistry/credentials.go | 6 + core/services/messaging/subjects_wire_test.go | 46 ++++++ core/services/nodes/model_router_test.go | 14 +- core/services/nodes/registry_wire_test.go | 39 +++++ core/services/nodes/router.go | 46 ++++-- .../nodes/router_unnamed_replica_test.go | 97 ++++++++++++ core/services/worker/addr_test.go | 45 ++++++ core/services/worker/config.go | 43 +++++- core/services/worker/worker.go | 53 +++---- docs/content/features/distributed-mode.md | 4 +- pkg/mcp/localaitools/dto.go | 16 +- pkg/mcp/localaitools/dto_test.go | 2 +- pkg/mcp/localaitools/httpapi/client.go | 14 +- 15 files changed, 569 insertions(+), 56 deletions(-) create mode 100644 core/cli/workerregistry/client_test.go create mode 100644 core/services/messaging/subjects_wire_test.go create mode 100644 core/services/nodes/registry_wire_test.go create mode 100644 core/services/nodes/router_unnamed_replica_test.go diff --git a/core/cli/workerregistry/client.go b/core/cli/workerregistry/client.go index f8728166f..fb00fb3f1 100644 --- a/core/cli/workerregistry/client.go +++ b/core/cli/workerregistry/client.go @@ -8,7 +8,9 @@ import ( "bytes" "context" "encoding/json" + "errors" "fmt" + "io" "net/http" "strings" "sync" @@ -93,7 +95,7 @@ func (c *RegistrationClient) RegisterFull(ctx context.Context, body map[string]a defer resp.Body.Close() if resp.StatusCode < 200 || resp.StatusCode >= 300 { - return nil, fmt.Errorf("registration failed with status %d", resp.StatusCode) + return nil, registrationStatusError(resp) } var result RegisterResponse @@ -103,6 +105,58 @@ func (c *RegistrationClient) RegisterFull(ctx context.Context, body map[string]a return &result, nil } +// ErrRegistrationRejected marks a registration the frontend REFUSED, as opposed +// to one it could not answer. +// +// Retrying a refusal cannot change it: the request is wrong, or this worker is +// not allowed to make it. The one that matters in practice is a worker of this +// release registering against a frontend that predates it, which answers +// "address is required for backend workers" with 400, because a worker no +// longer has an address to send. Without this the retry ladder spends four +// minutes on a verdict the frontend reached instantly, and the operator watches +// it before being told anything. +// +// 408 and 429 are deliberately NOT rejections. Both are the frontend asking for +// the same request again later, which is exactly what a retry does. +var ErrRegistrationRejected = errors.New("the frontend refused this registration") + +// maxRegistrationErrorBody bounds how much of a refusal's body is quoted back. +// Enough for a message, not enough for an HTML error page to bury the log line +// it is meant to explain. +const maxRegistrationErrorBody = 512 + +// registrationStatusError turns a non-2xx response into an error that says WHY. +// +// The body is the point. The frontend explains its refusals there +// ("address is required for backend workers", "invalid registration token"), +// and discarding it left an operator with a bare status code: the one line that +// would tell them which of several possible mistakes they made was read off the +// socket and thrown away. +func registrationStatusError(resp *http.Response) error { + detail, err := io.ReadAll(io.LimitReader(resp.Body, maxRegistrationErrorBody)) + if err != nil { + xlog.Debug("Could not read the frontend's registration error body", "status", resp.StatusCode, "error", err) + } + msg := strings.Join(strings.Fields(string(detail)), " ") + base := fmt.Sprintf("registration failed with status %d", resp.StatusCode) + if msg != "" { + base = fmt.Sprintf("%s: %s", base, msg) + } + if isRegistrationRejection(resp.StatusCode) { + return fmt.Errorf("%s: %w", base, ErrRegistrationRejected) + } + return errors.New(base) +} + +// isRegistrationRejection reports whether a status is a verdict rather than a +// condition that may pass. +func isRegistrationRejection(status int) bool { + if status == http.StatusRequestTimeout || status == http.StatusTooManyRequests { + return false + } + return status >= 400 && status < 500 +} + // Register sends a single registration request and returns the node ID and // optional credentials (API token for agent workers, NATS JWT when configured). func (c *RegistrationClient) Register(ctx context.Context, body map[string]any) (nodeID, apiToken, natsJWT, natsSeed string, err error) { @@ -138,6 +192,12 @@ func (c *RegistrationClient) RegisterFullWithRetry(ctx context.Context, body map if err == nil { return res, nil } + if errors.Is(err, ErrRegistrationRejected) { + // A verdict, not an outage. Reported on the first attempt so the + // reason the frontend gave is the first thing in the log rather + // than the last, after the ladder. + return nil, err + } if attempt == maxRetries { return nil, fmt.Errorf("failed after %d attempts: %w", maxRetries, err) } diff --git a/core/cli/workerregistry/client_test.go b/core/cli/workerregistry/client_test.go new file mode 100644 index 000000000..5870d2524 --- /dev/null +++ b/core/cli/workerregistry/client_test.go @@ -0,0 +1,138 @@ +package workerregistry + +import ( + "context" + "errors" + "net/http" + "net/http/httptest" + "sync/atomic" + "time" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +// The case these specs exist for: a worker of this release registering against +// a frontend that predates it. A worker no longer sends an address, the old +// frontend requires one, and it answers 400 with the reason in the body. Two +// things used to go wrong there at once. The reason was discarded, so the +// operator saw only "status 400" and had to guess which of several mistakes +// they had made; and the retry ladder spent four minutes on a verdict the +// frontend reached instantly. +var _ = Describe("Registration client refusals", func() { + var ( + attempts atomic.Int32 + status atomic.Int32 + body atomic.Value // string + server *httptest.Server + client *RegistrationClient + // seen carries one token per request the handler served, so a spec can + // wait for the Nth attempt instead of sleeping for however long the + // ladder's backoff happens to be. + seen chan struct{} + ) + + BeforeEach(func() { + attempts.Store(0) + status.Store(int32(http.StatusBadRequest)) + body.Store(`{"error":{"code":400,"message":"address is required for backend workers"}}`) + seen = make(chan struct{}, 64) + server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + attempts.Add(1) + select { + case seen <- struct{}{}: + default: + } + w.WriteHeader(int(status.Load())) + _, _ = w.Write([]byte(body.Load().(string))) + })) + client = &RegistrationClient{FrontendURL: server.URL, HTTPTimeout: 2 * time.Second} + }) + + AfterEach(func() { server.Close() }) + + It("quotes what the frontend said", func() { + _, err := client.RegisterFull(context.Background(), map[string]any{"name": "w1"}) + Expect(err).To(HaveOccurred()) + Expect(err.Error()).To(ContainSubstring("status 400")) + Expect(err.Error()).To(ContainSubstring("address is required for backend workers")) + }) + + It("marks a 4xx as a refusal", func() { + _, err := client.RegisterFull(context.Background(), map[string]any{"name": "w1"}) + Expect(err).To(MatchError(ErrRegistrationRejected)) + }) + + It("does not mark a 5xx as a refusal", func() { + // A frontend that is restarting or wedged has not judged anything, and + // retrying it is the whole reason the ladder exists. + status.Store(int32(http.StatusBadGateway)) + body.Store("bad gateway") + _, err := client.RegisterFull(context.Background(), map[string]any{"name": "w1"}) + Expect(err).To(HaveOccurred()) + Expect(errors.Is(err, ErrRegistrationRejected)).To(BeFalse()) + }) + + DescribeTable("treats a status that asks for the same request again as retryable", + func(code int) { + status.Store(int32(code)) + body.Store("later") + _, err := client.RegisterFull(context.Background(), map[string]any{"name": "w1"}) + Expect(err).To(HaveOccurred()) + Expect(errors.Is(err, ErrRegistrationRejected)).To(BeFalse()) + }, + Entry("408 Request Timeout", http.StatusRequestTimeout), + Entry("429 Too Many Requests", http.StatusTooManyRequests), + ) + + It("stops the retry ladder on the first refusal", func() { + // Ten attempts on a 400 is roughly four minutes of backoff before the + // operator is told anything, and the answer is the same one the + // frontend gave immediately. + _, err := client.RegisterFullWithRetry(context.Background(), map[string]any{"name": "w1"}, 10) + Expect(err).To(MatchError(ErrRegistrationRejected)) + Expect(err.Error()).To(ContainSubstring("address is required for backend workers")) + Expect(attempts.Load()).To(Equal(int32(1))) + }) + + It("still retries something that is not a refusal", func() { + // The control. Without it, a change that returned on EVERY error would + // pass the spec above and silently delete the retry behaviour a worker + // booting alongside its frontend depends on. + status.Store(int32(http.StatusServiceUnavailable)) + body.Store("starting up") + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + done := make(chan error, 1) + go func() { + _, err := client.RegisterFullWithRetry(ctx, map[string]any{"name": "w1"}, 10) + done <- err + }() + + // Two tokens is the whole assertion: the ladder came back for a second + // attempt on a status that is not a verdict. Waiting on the handler + // rather than on a duration makes it exact instead of tolerant. + Eventually(seen).Should(Receive()) + Eventually(seen, "10s").Should(Receive()) + cancel() + + var ladderErr error + Eventually(done).Should(Receive(&ladderErr)) + Expect(ladderErr).To(HaveOccurred()) + Expect(errors.Is(ladderErr, ErrRegistrationRejected)).To(BeFalse()) + Expect(attempts.Load()).To(BeNumerically(">=", 2)) + }) + + It("stops the credential manager's acquire loop on a refusal", func() { + // The default worker path goes through Acquire, not the ladder above, + // and its bound is 100 attempts rather than 10. A refusal there is the + // same verdict and has to end the same way. + mgr := NewNATSCredentialManager(func(ctx context.Context) (*RegisterResponse, error) { + return client.RegisterFull(ctx, map[string]any{"name": "w1"}) + }, true) + _, err := mgr.Acquire(context.Background()) + Expect(err).To(MatchError(ErrRegistrationRejected)) + Expect(attempts.Load()).To(Equal(int32(1))) + }) +}) diff --git a/core/cli/workerregistry/credentials.go b/core/cli/workerregistry/credentials.go index f9d8c2231..b023b9916 100644 --- a/core/cli/workerregistry/credentials.go +++ b/core/cli/workerregistry/credentials.go @@ -2,6 +2,7 @@ package workerregistry import ( "context" + "errors" "fmt" "sync" "time" @@ -146,6 +147,11 @@ func (m *NATSCredentialManager) Acquire(ctx context.Context) (*RegisterResponse, for attempt := 1; m.maxAttempts <= 0 || attempt <= m.maxAttempts; attempt++ { res, err := m.register(ctx) switch { + case errors.Is(err, ErrRegistrationRejected): + // The frontend refused rather than failed. Waiting through the full + // attempt ladder would delay the operator's only explanation by the + // length of the ladder and change nothing about the answer. + return nil, err case err != nil: lastReason = err xlog.Warn("Registration failed, retrying", "attempt", attempt, "next_retry", backoff, "error", err) diff --git a/core/services/messaging/subjects_wire_test.go b/core/services/messaging/subjects_wire_test.go new file mode 100644 index 000000000..2c90254e8 --- /dev/null +++ b/core/services/messaging/subjects_wire_test.go @@ -0,0 +1,46 @@ +package messaging + +import ( + "encoding/json" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +// BackendInstallReply.WorkerLocalAddress was called Address until workers +// stopped advertising. The Go field was renamed so no reader takes it for a +// dial target; the wire key was deliberately NOT renamed, because a worker and +// a frontend from different releases have to keep understanding each other +// across a rolling upgrade. +// +// That is a cross-version compatibility property resting on one struct tag, and +// a struct tag nobody asserts is a property nobody has. Renaming just the tags +// left the whole suite green when this was written. +var _ = Describe("backend.install reply wire format", func() { + It("writes the address under the key an older frontend reads", func() { + out, err := json.Marshal(BackendInstallReply{Success: true, WorkerLocalAddress: "127.0.0.1:50052"}) + Expect(err).ToNot(HaveOccurred()) + + var raw map[string]any + Expect(json.Unmarshal(out, &raw)).To(Succeed()) + Expect(raw).To(HaveKeyWithValue("address", "127.0.0.1:50052")) + Expect(raw).ToNot(HaveKey("worker_local_address"), + "renaming the wire key would make every install reply unreadable to a frontend of another release") + }) + + It("reads the address an older worker sends", func() { + // An older worker puts its ADVERTISED host here. Only the port is used, + // and the port is the same, so accepting it is both harmless and the + // thing that keeps a mixed fleet working. + var reply BackendInstallReply + Expect(json.Unmarshal([]byte(`{"success":true,"address":"worker-1:50052"}`), &reply)).To(Succeed()) + Expect(reply.Success).To(BeTrue()) + Expect(reply.WorkerLocalAddress).To(Equal("worker-1:50052")) + }) + + It("omits the address when the install failed", func() { + out, err := json.Marshal(BackendInstallReply{Success: false, Error: "boom"}) + Expect(err).ToNot(HaveOccurred()) + Expect(string(out)).ToNot(ContainSubstring("address")) + }) +}) diff --git a/core/services/nodes/model_router_test.go b/core/services/nodes/model_router_test.go index a0250c9d1..193c9d545 100644 --- a/core/services/nodes/model_router_test.go +++ b/core/services/nodes/model_router_test.go @@ -22,6 +22,7 @@ type fakeModelRouterForSmartRouter struct { nodeModel *NodeModel findErr error decrementCalled map[string]int // "nodeID:model" -> count + removed []string // "nodeID:model:replica" per RemoveNodeModel } func newFakeModelRouterForSmartRouter() *fakeModelRouterForSmartRouter { @@ -46,9 +47,20 @@ func (f *fakeModelRouterForSmartRouter) DecrementInFlight(_ context.Context, nod func (f *fakeModelRouterForSmartRouter) IncrementInFlight(_ context.Context, _, _ string, _ int) error { return nil } -func (f *fakeModelRouterForSmartRouter) RemoveNodeModel(_ context.Context, _, _ string, _ int) error { +func (f *fakeModelRouterForSmartRouter) RemoveNodeModel(_ context.Context, nodeID, modelName string, replicaIndex int) error { + f.mu.Lock() + defer f.mu.Unlock() + f.removed = append(f.removed, fmt.Sprintf("%s:%s:%d", nodeID, modelName, replicaIndex)) return nil } + +// removedModels lists the replica rows the code under test deleted, so a spec +// can assert a branch left a row alone rather than only that it returned nil. +func (f *fakeModelRouterForSmartRouter) removedModels() []string { + f.mu.Lock() + defer f.mu.Unlock() + return append([]string(nil), f.removed...) +} func (f *fakeModelRouterForSmartRouter) RemoveAllNodeModelReplicas(_ context.Context, _, _ string) error { return nil } diff --git a/core/services/nodes/registry_wire_test.go b/core/services/nodes/registry_wire_test.go new file mode 100644 index 000000000..34eec32e4 --- /dev/null +++ b/core/services/nodes/registry_wire_test.go @@ -0,0 +1,39 @@ +package nodes + +import ( + "encoding/json" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +// NodeModel.WorkerLocalAddress was called Address. The Go field was renamed so +// no reader takes it for a frontend-dialable endpoint; the json key was kept so +// no API consumer breaks, and the gorm column was kept so no migration is +// needed. +// +// The column half is enforced by the raw-SQL fragments in this package, which +// fail loudly against a renamed column. The json half had nothing enforcing it: +// renaming only the tag left this package, messaging and endpoints/localai all +// green. These specs are that half. +var _ = Describe("NodeModel wire format", func() { + It("serves the address under the key API consumers already read", func() { + out, err := json.Marshal(NodeModel{ + NodeID: "node-1", ModelName: "m", ReplicaIndex: 1, + WorkerLocalAddress: "127.0.0.1:50052", + }) + Expect(err).ToNot(HaveOccurred()) + + var raw map[string]any + Expect(json.Unmarshal(out, &raw)).To(Succeed()) + Expect(raw).To(HaveKeyWithValue("address", "127.0.0.1:50052")) + Expect(raw).ToNot(HaveKey("worker_local_address"), + "GET /api/nodes/{id}/models and /api/nodes/models both serve this struct verbatim") + }) + + It("round-trips a body written against the documented key", func() { + var nm NodeModel + Expect(json.Unmarshal([]byte(`{"node_id":"node-1","model_name":"m","address":"127.0.0.1:50052"}`), &nm)).To(Succeed()) + Expect(nm.WorkerLocalAddress).To(Equal("127.0.0.1:50052")) + }) +}) diff --git a/core/services/nodes/router.go b/core/services/nodes/router.go index e356d40ea..4e33c0981 100644 --- a/core/services/nodes/router.go +++ b/core/services/nodes/router.go @@ -724,12 +724,28 @@ func (r *SmartRouter) tryWarmPath(ctx context.Context, att *routeAttempt) *Route modelAddr := nm.WorkerLocalAddress replicaIdx := nm.ReplicaIndex - // A replica row that does not name its backend process cannot be routed - // to. There is no node address left to stand in for it, and naming an - // empty target would make the worker refuse the stream as invalid: a - // refusal that reads as the worker answering about its backend, which is - // evidence this row is not entitled to produce. Fall through to a cold - // load, which either replaces the row or reports a real failure. + // A replica row that does not name its backend process cannot be routed to. + // There is no node address left to stand in for it, and an empty target + // names no process, so the request would open a stream the worker refuses + // as an invalid request rather than one that reaches a backend. Fall + // through to a cold load, which either replaces the row or reports a real + // failure. + // + // The row is left in place, unlike the !alive branch below which removes + // it. That branch has OBSERVED a backend dead; this one has observed only + // that the row is unreadable, which says nothing about whether a process is + // running on that worker. The row is also the last record that one might + // be: the acknowledged stop path matches on ExpectedAddress and a worker + // refuses a stop whose address does not match, so an empty one cannot be + // cleaned up through it either. Keeping the row costs a lock and a + // decrement per request before the cold load and leaves something an + // operator can see; removing it would free the replica slot for a second + // copy of the model while the first one, if it exists, keeps its VRAM with + // nothing left pointing at it. + // + // Defensive rather than reachable: installBackendOnNode below refuses an + // install that names no address, so no row written by this release can look + // like this. if modelAddr == "" { if err := r.registry.DecrementInFlight(ctx, node.ID, att.trackingKey, replicaIdx); err != nil { xlog.Warn("Failed to release a reservation for an unnamed replica", @@ -1374,9 +1390,21 @@ func (r *SmartRouter) installBackendOnNode(ctx context.Context, node *BackendNod // Where the backend process listens on that worker. There is no node // address to fall back to any more, and there should not be: a worker // that reports success without naming the port it started the process - // on has told us nothing routable, and inventing a target would surface - // later as the worker refusing an invalid stream, which reads as - // evidence about the backend rather than about this install. + // on has produced nothing routable, and the failure belongs to THIS + // install rather than to whatever later step first tries to use the + // address. Substituting one would push a known-bad value into a replica + // row and defer the error to a probe, where its cause is no longer + // visible. + // + // An earlier version of this comment justified it by saying the worker + // would refuse the resulting empty target as an invalid stream and that + // the refusal would read as the worker answering about its backend. The + // first half is true (see cluster.isWorkerAnswer) and the second is + // not: nothing in this package branches on cluster.ErrNoRoute, and + // `unroutable` treats ANY recorded dial error as unroutable, so such a + // refusal reaches every reap guard as ProbeUnknown and deletes nothing. + // The decision stands on the grounds above, which do not depend on a + // classification the frontend does not currently make. if reply.WorkerLocalAddress == "" { return "", fmt.Errorf("worker %s reported backend %q installed but named no address for the process", node.ID, backendType) } diff --git a/core/services/nodes/router_unnamed_replica_test.go b/core/services/nodes/router_unnamed_replica_test.go new file mode 100644 index 000000000..a6a01fda7 --- /dev/null +++ b/core/services/nodes/router_unnamed_replica_test.go @@ -0,0 +1,97 @@ +package nodes + +import ( + "context" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +// A replica row carries the address of the backend process it names, and that +// address is how the frontend says WHICH process on a worker it means. A row +// without one names nothing, and the warm path has to decline it rather than +// route with an empty target. +// +// This is defensive: installBackendOnNode now guarantees a non-empty address +// before any row is written, so the only rows that can look like this are ones +// an older frontend wrote. It is specced anyway because the branch touches an +// in-flight reservation, and a reservation that is taken and not returned pins +// the replica against every eviction query for the life of the row. +var _ = Describe("SmartRouter warm path with an unnamed replica", func() { + var ( + registry *fakeModelRouterForSmartRouter + clients *fakeBackendClientFactory + router *SmartRouter + node *BackendNode + ) + + BeforeEach(func() { + node = &BackendNode{ID: "node-1", Name: "node-1", Status: StatusHealthy} + registry = newFakeModelRouterForSmartRouter() + registry.node = node + clients = newFakeBackendClientFactory() + router = NewSmartRouter(registry, SmartRouterOptions{ClientFactory: clients}) + }) + + warm := func() *RouteResult { + return router.tryWarmPath(context.Background(), &routeAttempt{trackingKey: "m", modelName: "m"}) + } + + Context("when the row names no backend process", func() { + BeforeEach(func() { + registry.nodeModel = &NodeModel{NodeID: node.ID, ModelName: "m", ReplicaIndex: 0} + }) + + It("declines the warm path so the caller cold-loads", func() { + Expect(warm()).To(BeNil()) + }) + + It("never asks for a client, so the empty target reaches no dialler", func() { + // The failure this prevents: an empty target opens a stream the + // worker refuses as an invalid request. That refusal is an answer + // FROM the worker, so it is the one failure on the whole path that + // is real evidence about a backend, and this row is not entitled to + // produce evidence about anything. + Expect(warm()).To(BeNil()) + Expect(clients.nodesSeen()).To(BeEmpty()) + Expect(clients.addressesSeen()).To(BeEmpty()) + }) + + It("returns the reservation FindAndLockNodeWithModel took", func() { + // Held rather than returned, the row's in_flight never reaches 0 + // and no eviction query can ever select it, so the replica slot and + // its VRAM are pinned for the life of the row. + Expect(warm()).To(BeNil()) + registry.mu.Lock() + defer registry.mu.Unlock() + Expect(registry.decrementCalled).To(HaveKeyWithValue("node-1:m", 1)) + }) + + It("leaves the row in place", func() { + // Deliberately unlike the sibling !alive branch, which removes the + // row. A dead backend has been observed dead; this row has been + // observed to be unreadable, which says nothing about whether a + // process is running on that worker. It is also the last record + // that one may be: the acknowledged stop path matches on + // ExpectedAddress, so a stop for an empty one is refused by the + // worker, and deleting the row here would free the replica slot for + // a second copy of the same model while the first one, if it + // exists, keeps its VRAM. The cost of keeping it is one lock and + // decrement per request before the cold load, and a row an operator + // can see; the cost of removing it is an orphan nothing points at. + Expect(warm()).To(BeNil()) + Expect(registry.removedModels()).To(BeEmpty()) + }) + }) + + It("routes normally once the row names one", func() { + // The control. Without it every assertion above would also pass on a + // warm path that declined everything. + registry.nodeModel = &NodeModel{NodeID: node.ID, ModelName: "m", ReplicaIndex: 0, WorkerLocalAddress: "127.0.0.1:50052"} + Expect(warm()).ToNot(BeNil()) + Expect(clients.addressesSeen()).To(ContainElement("127.0.0.1:50052")) + registry.mu.Lock() + defer registry.mu.Unlock() + Expect(registry.decrementCalled).ToNot(HaveKey("node-1:m")) + }) +}) diff --git a/core/services/worker/addr_test.go b/core/services/worker/addr_test.go index 633d183b2..447880653 100644 --- a/core/services/worker/addr_test.go +++ b/core/services/worker/addr_test.go @@ -90,3 +90,48 @@ var _ = Describe("Worker address resolution", func() { }) }) }) + +var _ = Describe("Worker startup validation", func() { + // A Config as kong would hand it over with nothing unusual set: the tunnel + // on by its default, no auth enforcement. + newConfig := func() *Config { + return &Config{WorkerTunnel: true} + } + + It("accepts the default configuration", func() { + Expect(newConfig().validateStartup()).To(Succeed()) + }) + + It("refuses to start with the tunnel turned off", func() { + // Not a warning and not a degraded mode. A worker without its tunnel + // advertises nothing, binds only loopback, and has no frontend path + // that dials it, yet it would register, heartbeat and report healthy, + // so the scheduler would keep placing models on it and every one would + // fail. Refusing at boot is the only outcome that is visible. + cfg := newConfig() + cfg.WorkerTunnel = false + err := cfg.validateStartup() + Expect(err).To(HaveOccurred()) + Expect(err.Error()).To(ContainSubstring("LOCALAI_WORKER_TUNNEL")) + Expect(err.Error()).To(ContainSubstring("nothing can reach it")) + }) + + It("refuses enforcement without a registration token", func() { + cfg := newConfig() + cfg.RegistrationRequireAuth = true + Expect(cfg.validateStartup()).To(MatchError(ContainSubstring("LOCALAI_REGISTRATION_TOKEN is empty"))) + }) + + It("refuses the umbrella switch without a registration token", func() { + cfg := newConfig() + cfg.DistributedRequireAuth = true + Expect(cfg.validateStartup()).To(MatchError(ContainSubstring("LOCALAI_REGISTRATION_TOKEN is empty"))) + }) + + It("accepts enforcement once a token is set", func() { + cfg := newConfig() + cfg.DistributedRequireAuth = true + cfg.RegistrationToken = "shared" + Expect(cfg.validateStartup()).To(Succeed()) + }) +}) diff --git a/core/services/worker/config.go b/core/services/worker/config.go index 375ba1822..a3bb09fa0 100644 --- a/core/services/worker/config.go +++ b/core/services/worker/config.go @@ -1,5 +1,7 @@ package worker +import "fmt" + // Config is the configuration for the distributed agent worker. // // Field tags are kong/kong-env metadata read by core/cli/worker.go's WorkerCMD, @@ -64,9 +66,22 @@ type Config struct { HeartbeatInterval string `env:"LOCALAI_HEARTBEAT_INTERVAL" default:"10s" help:"Interval between heartbeats" group:"registration"` // WorkerTunnel holds one outbound multiplexed connection to the frontend // and serves the frontend's requests over it, so the worker needs no - // inbound port. Turning it off leaves the worker reachable only at the - // addresses it advertises, which is the pre-tunnel behaviour. - WorkerTunnel bool `env:"LOCALAI_WORKER_TUNNEL" default:"true" help:"Hold one outbound multiplexed tunnel to the frontend and serve its requests over it, so this worker needs no inbound port." group:"distributed"` + // inbound port. + // + // Turning it off is now a fatal misconfiguration and validateStartup + // refuses to boot on it, which is a behaviour change from when this flag + // had a working "off" position. It no longer has one: this worker + // advertises no address and binds only loopback, and no frontend path + // dials a worker's address, so a worker without its tunnel is reachable by + // nothing. Left running it would be the worst available failure shape, + // because it registers, heartbeats and reports healthy, so the scheduler + // keeps placing models on it and every one of them fails. + // + // The flag is kept rather than deleted so that an operator who set it, on + // the old promise that it fell back to the advertised address, is told + // exactly that the promise is gone instead of having their setting quietly + // ignored. + WorkerTunnel bool `env:"LOCALAI_WORKER_TUNNEL" default:"true" help:"Hold one outbound multiplexed tunnel to the frontend and serve its requests over it, so this worker needs no inbound port. Setting it false is refused: a worker has no other way to be reached." group:"distributed"` NodeLabels string `env:"LOCALAI_NODE_LABELS" help:"Comma-separated key=value labels for this node (e.g. tier=fast,gpu=a100)" group:"registration"` // MaxReplicasPerModel caps how many replicas of any one model can run on // this worker concurrently. Default 1 = historical single-replica @@ -110,3 +125,25 @@ func (c Config) NatsAuthRequired() bool { func (c Config) RegistrationAuthRequired() bool { return c.RegistrationRequireAuth || c.DistributedRequireAuth } + +// validateStartup reports a configuration this worker must refuse to boot on, +// as opposed to one it can degrade under. +// +// It runs before prefetch, registration and NATS, so a refusal happens while +// the worker is still invisible to the cluster. That ordering is the point of +// checking here at all: both conditions below produce a worker that would +// register, heartbeat and be scheduled onto, so discovering them later means +// discovering them as failed inferences on a node the frontend believes is +// healthy. +func (c Config) validateStartup() error { + // The file-transfer server fails open on an empty token (see + // nodes.checkBearerToken), so enforcement plus no token is a request to + // serve the models directory unauthenticated. + if c.RegistrationAuthRequired() && c.RegistrationToken == "" { + return fmt.Errorf("registration auth is required (LOCALAI_REGISTRATION_REQUIRE_AUTH or LOCALAI_DISTRIBUTED_REQUIRE_AUTH) but LOCALAI_REGISTRATION_TOKEN is empty: refusing to start an unauthenticated file-transfer server") + } + if !c.WorkerTunnel { + return fmt.Errorf("LOCALAI_WORKER_TUNNEL is false, but this worker advertises no address and binds only loopback, and no frontend path dials a worker's address: without its tunnel nothing can reach it. Remove the setting, or run the pre-tunnel release on both the worker and the frontend") + } + return nil +} diff --git a/core/services/worker/worker.go b/core/services/worker/worker.go index 7184a63e4..a208eeb43 100644 --- a/core/services/worker/worker.go +++ b/core/services/worker/worker.go @@ -30,12 +30,12 @@ import ( func Run(ctx *cliContext.Context, cfg *Config) error { xlog.Info("Starting worker", "basePort", cfg.effectiveBasePort()) - // Fail fast (before prefetch/registration/NATS) when enforcement is on but no - // registration token is set: the worker's HTTP file-transfer server fails - // open on an empty token (see nodes.checkBearerToken), so refuse to start - // rather than register and then die mid-boot. - if cfg.RegistrationAuthRequired() && cfg.RegistrationToken == "" { - return fmt.Errorf("registration auth is required (LOCALAI_REGISTRATION_REQUIRE_AUTH or LOCALAI_DISTRIBUTED_REQUIRE_AUTH) but LOCALAI_REGISTRATION_TOKEN is empty — refusing to start an unauthenticated file-transfer server") + // Fail fast, before prefetch, registration and NATS, on any configuration + // that would produce a worker the cluster believes in and cannot use. See + // validateStartup for what those are and why each is fatal rather than + // degraded. + if err := cfg.validateStartup(); err != nil { + return err } systemState, err := system.GetSystemState( @@ -209,26 +209,29 @@ func Run(ctx *cliContext.Context, cfg *Config) error { // frontend URL or this node's identity is unusable, and a worker that // silently ran without its tunnel would look healthy while being // unreachable to everything that dials through it. - if cfg.WorkerTunnel { - tunnel, terr := StartTunnel(shutdownCtx, TunnelConfig{ - FrontendURL: cfg.RegisterTo, - NodeID: nodeID, - Token: tunnelToken, - // Built by tunnelServices rather than inline, so the routing - // table, which is this feature's security boundary, is reachable - // from a spec without starting a worker. - Services: tunnelServices(cfg, httpAddr), - }) - if terr != nil { - nodes.ShutdownFileTransferServer(httpServer) - return fmt.Errorf("starting the worker tunnel: %w", terr) - } - defer func() { - if err := tunnel.Close(); err != nil { - xlog.Warn("Closing the worker tunnel failed", "error", err) - } - }() + // + // Unconditional: LOCALAI_WORKER_TUNNEL=false is refused by validateStartup + // before this point, so there is no configuration that reaches here without + // one. A guard here would be a branch nothing can take, which reads as a + // supported no-tunnel mode that does not exist. + tunnel, terr := StartTunnel(shutdownCtx, TunnelConfig{ + FrontendURL: cfg.RegisterTo, + NodeID: nodeID, + Token: tunnelToken, + // Built by tunnelServices rather than inline, so the routing + // table, which is this feature's security boundary, is reachable + // from a spec without starting a worker. + Services: tunnelServices(cfg, httpAddr), + }) + if terr != nil { + nodes.ShutdownFileTransferServer(httpServer) + return fmt.Errorf("starting the worker tunnel: %w", terr) } + defer func() { + if err := tunnel.Close(); err != nil { + xlog.Warn("Closing the worker tunnel failed", "error", err) + } + }() // Connect to NATS xlog.Info("Connecting to NATS", "url", sanitize.URL(cfg.NatsURL)) diff --git a/docs/content/features/distributed-mode.md b/docs/content/features/distributed-mode.md index ff8354214..bbdb45777 100644 --- a/docs/content/features/distributed-mode.md +++ b/docs/content/features/distributed-mode.md @@ -1243,9 +1243,9 @@ Notes: **Port conflicts on workers:** - Each model gets its own gRPC process on an incrementing port (50051, 50052, ...) - The HTTP file transfer server runs on the base port - 1 (default: 50050) -- Ensure the port range is not blocked by firewalls or used by other services +- All of those bind loopback, so a firewall cannot be the cause. What can is another service on the same host already holding a port in the range: move the worker's range with `LOCALAI_ADDR` (see [Worker Port Configuration](#worker-port-configuration)) or bound it with `LOCALAI_GRPC_MAX_PORT` - Verify the backend gallery configuration is correct -- The worker needs network access to download backends from the gallery +- The worker needs OUTBOUND network access to the gallery, to `LOCALAI_REGISTER_TO` and to `LOCALAI_NATS_URL`. It needs no inbound access at all ## Roadmap: Routing and Caching Enhancements diff --git a/pkg/mcp/localaitools/dto.go b/pkg/mcp/localaitools/dto.go index 1055f86d9..dd9113df9 100644 --- a/pkg/mcp/localaitools/dto.go +++ b/pkg/mcp/localaitools/dto.go @@ -97,13 +97,17 @@ type SystemInfo struct { } // Node is one entry in list_nodes. +// +// It carries no address. A worker holds one outbound tunnel to a frontend +// replica and advertises no endpoint, so both `address` and `http_address` are +// empty on every node registered by a current worker. They were dropped rather +// than left empty because this struct is read by the LocalAI Assistant, and an +// always-blank field an operator can ask about invites an answer built on it. type Node struct { - ID string `json:"id"` - Address string `json:"address,omitempty"` - HTTPAddress string `json:"http_address,omitempty"` - TotalVRAM uint64 `json:"total_vram,omitempty"` - Healthy bool `json:"healthy"` - LastSeen string `json:"last_seen,omitempty"` + ID string `json:"id"` + TotalVRAM uint64 `json:"total_vram,omitempty"` + Healthy bool `json:"healthy"` + LastSeen string `json:"last_seen,omitempty"` } // SetNodeVRAMBudgetRequest is the input for set_node_vram_budget. It PUTs diff --git a/pkg/mcp/localaitools/dto_test.go b/pkg/mcp/localaitools/dto_test.go index 865d00e8c..2d807d0b7 100644 --- a/pkg/mcp/localaitools/dto_test.go +++ b/pkg/mcp/localaitools/dto_test.go @@ -31,7 +31,7 @@ var _ = Describe("DTOs round-trip through JSON", func() { roundTripDTO(InstallBackendRequest{GalleryName: "g", BackendName: "b"}) roundTripDTO(Backend{Name: "n", Installed: true}) roundTripDTO(SystemInfo{Version: "v1", Distributed: false, ModelsPath: "/tmp", LoadedModels: []string{"a"}, InstalledBackends: []string{"x"}}) - roundTripDTO(Node{ID: "n", Address: "a", HTTPAddress: "h", TotalVRAM: 100, Healthy: true, LastSeen: "now"}) + roundTripDTO(Node{ID: "n", TotalVRAM: 100, Healthy: true, LastSeen: "now"}) roundTripDTO(VRAMEstimateRequest{ModelName: "m", ContextSize: 4096, GPULayers: -1, KVQuantBits: 8}) roundTripDTO(ImportModelURIRequest{URI: "u", BackendPreference: "llama-cpp", Overrides: map[string]any{"k": "v"}}) roundTripDTO(ImportModelURIResponse{JobID: "j", DiscoveredModelName: "m", AmbiguousBackend: true, Modality: "tts", BackendCandidates: []string{"a", "b"}, Hint: "h"}) diff --git a/pkg/mcp/localaitools/httpapi/client.go b/pkg/mcp/localaitools/httpapi/client.go index 923f35eba..791ddc1db 100644 --- a/pkg/mcp/localaitools/httpapi/client.go +++ b/pkg/mcp/localaitools/httpapi/client.go @@ -458,11 +458,11 @@ func (c *Client) SystemInfo(ctx context.Context) (*localaitools.SystemInfo, erro } func (c *Client) ListNodes(ctx context.Context) ([]localaitools.Node, error) { + // address / http_address are deliberately not decoded: a worker advertises + // no endpoint, so both are empty on every current node. var raw []struct { - ID string `json:"id"` - Address string `json:"address"` - HTTPAddress string `json:"http_address"` - Status string `json:"status"` + ID string `json:"id"` + Status string `json:"status"` } if err := c.do(ctx, http.MethodGet, routeNodes, nil, &raw); err != nil { // Treat 404/disabled as "no nodes" to keep parity with single-process. @@ -474,10 +474,8 @@ func (c *Client) ListNodes(ctx context.Context) ([]localaitools.Node, error) { out := make([]localaitools.Node, 0, len(raw)) for _, n := range raw { out = append(out, localaitools.Node{ - ID: n.ID, - Address: n.Address, - HTTPAddress: n.HTTPAddress, - Healthy: n.Status == "healthy", + ID: n.ID, + Healthy: n.Status == "healthy", }) } return out, nil