From 49b3d22974824b6a77c060c94221260ebbf281fa Mon Sep 17 00:00:00 2001 From: Ettore Di Giacinto Date: Wed, 2 Sep 2026 15:51:58 +0000 Subject: [PATCH] feat(worker): serve the control plane over the tunnel, not over NATS Ten NATS subscriptions on the worker become ten HTTP routes under /v1/control/, served on the loopback HTTP server the worker already runs and reached only through the tunnel's existing `http` stream tag. The carrier is the tag that already exists rather than a new one. A new tag would have had to invent correlation, per-request deadlines, unbounded payloads and a progress stream, and each of those is a place this branch has already put a defect. It would also have added a fifth entry to the worker's stream-refusal vocabulary, which decides what a frontend reaps on and took eight fixes to settle. Riding `http` means a control RPC to a worker another replica holds takes the same relay the inference path takes, which is the path that has been measured. The request and reply DTOs are untouched, so a body on a control route is byte-for-byte what the corresponding subject carried. No subject was deleted: agent workers still subscribe to nodes..backend.stop. Install and upgrade stream. They answer application/x-ndjson: zero or more {"progress":...} lines carrying the same event the per-op NATS subject carried, then exactly one {"reply":...} line, always last. That deletes the 8000-byte notification cap structurally instead of reproducing it on a new carrier: a progress line is written into the response the caller is already reading, so there is nothing to size and no subscribe-before-request window. The debouncer is shared with the NATS publisher rather than forked, so the ~4/s tick bound is one fact. A verb's own failure is a 200 with Error set, never a 5xx. The frontend maps a transport failure onto "no route to that worker", which nothing may act on, and the worker's answer onto evidence a reap guard may act on; answering 500 for a failed install would put the worker's verdict in the bucket reserved for a broken link. Only a request that could not be read or routed is non-2xx. Control RPCs carry the caller's budget. r.Context() replaces four context.Background() calls at the gallery-install sites, and the one pre-existing fixed timeout on model.unload is now derived from the caller's context so a shorter budget is honoured. No timeout is invented. The inner `go func()` in the install and upgrade handlers is deleted rather than nested: it existed because one subscription served every install, and over HTTP each request already has its own goroutine. Per-backend serialization stays lockBackend, which is what actually prevented two requests racing the gallery directory. Bounds against a boundary the worker now serves: every body is capped at 8 MiB before any decode; the 404 echoes at most 128 bytes of the request path, cut on a rune boundary so a half rune cannot travel downstream as a replacement character; non-POST is refused before the body is read so a probe cannot fire a command; the streaming responses set nosniff. The routes mount through nodes.AuthenticatedRoutes, which hands the registrar a private mux and puts the whole prefix behind the same constant-time bearer check as the file routes. The worker's HTTP server now takes the supervisor as a required parameter, so there is no way to start it without the control plane mounted. Assisted-by: Claude Opus 5 [claude-code] Signed-off-by: Ettore Di Giacinto --- .../nodes/authenticated_routes_test.go | 92 +++ core/services/nodes/file_transfer_server.go | 47 +- .../nodes/install_progress_publisher.go | 59 +- core/services/worker/control_routes.go | 383 +++++++++++++ core/services/worker/control_routes_test.go | 524 ++++++++++++++++++ core/services/worker/install.go | 51 +- core/services/worker/lifecycle.go | 318 +++-------- core/services/worker/model_stop_test.go | 24 +- core/services/worker/models_running.go | 8 - core/services/worker/supervisor.go | 8 +- core/services/worker/worker.go | 96 ++-- core/services/workerctl/paths.go | 71 +++ core/services/workerctl/paths_test.go | 72 +++ .../workerctl/workerctl_suite_test.go | 13 + docs/content/features/distributed-mode.md | 34 +- 15 files changed, 1469 insertions(+), 331 deletions(-) create mode 100644 core/services/nodes/authenticated_routes_test.go create mode 100644 core/services/worker/control_routes.go create mode 100644 core/services/worker/control_routes_test.go create mode 100644 core/services/workerctl/paths.go create mode 100644 core/services/workerctl/paths_test.go create mode 100644 core/services/workerctl/workerctl_suite_test.go diff --git a/core/services/nodes/authenticated_routes_test.go b/core/services/nodes/authenticated_routes_test.go new file mode 100644 index 000000000..db6c2a72b --- /dev/null +++ b/core/services/nodes/authenticated_routes_test.go @@ -0,0 +1,92 @@ +package nodes + +import ( + "net" + "net/http" + "strings" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +var _ = Describe("extra authenticated routes on the worker HTTP server", func() { + const token = "s3cr3t" + + var ( + srv *http.Server + base string + ) + + BeforeEach(func() { + lis, err := net.Listen("tcp", "127.0.0.1:0") + Expect(err).NotTo(HaveOccurred()) + dir := GinkgoT().TempDir() + srv, err = StartFileTransferServerWithRoutes(lis, dir, dir, dir, token, 0, nil, + &AuthenticatedRoutes{ + Prefix: "/v1/control/", + Register: func(mux *http.ServeMux) { + mux.HandleFunc("/v1/control/ping", func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte("pong")) + }) + }, + }) + Expect(err).NotTo(HaveOccurred()) + base = "http://" + lis.Addr().String() + DeferCleanup(func() { ShutdownFileTransferServer(srv) }) + }) + + get := func(path, bearer string) *http.Response { + GinkgoHelper() + req, err := http.NewRequest(http.MethodGet, base+path, nil) + Expect(err).NotTo(HaveOccurred()) + if bearer != "" { + req.Header.Set("Authorization", "Bearer "+bearer) + } + resp, err := http.DefaultClient.Do(req) + Expect(err).NotTo(HaveOccurred()) + DeferCleanup(func() { _ = resp.Body.Close() }) + return resp + } + + It("serves a registered extra route to a caller carrying the token", func() { + Expect(get("/v1/control/ping", token).StatusCode).To(Equal(http.StatusOK)) + }) + + It("refuses an unauthenticated request on an extra route", func() { + // The control plane must not be a second authentication path: an extra + // route that forgot its own check would be an unauthenticated command + // on the boundary the tunnel exposes. + Expect(get("/v1/control/ping", "").StatusCode).To(Equal(http.StatusUnauthorized)) + }) + + It("refuses a wrong token on an extra route", func() { + Expect(get("/v1/control/ping", "wrong").StatusCode).To(Equal(http.StatusUnauthorized)) + }) + + It("checks the token before the route exists, so an unknown control path leaks nothing", func() { + Expect(get("/v1/control/no-such-verb", "").StatusCode).To(Equal(http.StatusUnauthorized)) + }) + + It("leaves the file routes reachable alongside the extra ones", func() { + Expect(get("/healthz", "").StatusCode).To(Equal(http.StatusOK)) + }) + + It("mounts nothing when no route set is given", func() { + lis, err := net.Listen("tcp", "127.0.0.1:0") + Expect(err).NotTo(HaveOccurred()) + dir := GinkgoT().TempDir() + bare, err := StartFileTransferServerWithRoutes(lis, dir, dir, dir, token, 0, nil, nil) + Expect(err).NotTo(HaveOccurred()) + DeferCleanup(func() { ShutdownFileTransferServer(bare) }) + + req, err := http.NewRequest(http.MethodGet, "http://"+lis.Addr().String()+"/v1/control/ping", nil) + Expect(err).NotTo(HaveOccurred()) + req.Header.Set("Authorization", "Bearer "+token) + resp, err := http.DefaultClient.Do(req) + Expect(err).NotTo(HaveOccurred()) + DeferCleanup(func() { _ = resp.Body.Close() }) + Expect(resp.StatusCode).To(Equal(http.StatusNotFound)) + Expect(strings.TrimSpace(resp.Status)).NotTo(BeEmpty()) + }) +}) diff --git a/core/services/nodes/file_transfer_server.go b/core/services/nodes/file_transfer_server.go index 9a3f0ca2f..8232d9f64 100644 --- a/core/services/nodes/file_transfer_server.go +++ b/core/services/nodes/file_transfer_server.go @@ -42,17 +42,35 @@ const ( targetSidecarSuffix = ".sha256.target" ) +// AuthenticatedRoutes is a set of extra routes to serve on the worker's HTTP +// server, mounted under Prefix and behind the SAME bearer check as the file +// routes. +// +// It is a mount request rather than a plain func(*http.ServeMux) so this +// package, and not its caller, owns the authentication. A caller handed the +// server's own mux would be registering handlers alongside the file routes, and +// forgetting the token check in one of them would be an unauthenticated verb +// rather than a compile error. Register is instead given a mux of its own, +// which is reachable only through the check below. +type AuthenticatedRoutes struct { + // Prefix is the single path prefix every registered route lives under. + Prefix string + // Register mounts the routes on a mux private to this route set. + Register func(*http.ServeMux) +} + // StartFileTransferServer starts a small HTTP server for file transfer in distributed mode. // It provides PUT/GET/POST endpoints for uploading, downloading, and allocating temp files, // as well as backend log REST and WebSocket endpoints when logStore is non-nil. // Auth is via Bearer token (registration token), using constant-time comparison. // A nil readiness fails open, keeping /readyz's historical always-200 answer. -func StartFileTransferServer(addr, stagingDir, modelsDir, dataDir, token string, maxUploadSize int64, readiness *WorkerReadiness, logStore ...*model.BackendLogStore) (*http.Server, error) { +// A nil extra mounts no additional routes. +func StartFileTransferServer(addr, stagingDir, modelsDir, dataDir, token string, maxUploadSize int64, readiness *WorkerReadiness, extra *AuthenticatedRoutes, logStore ...*model.BackendLogStore) (*http.Server, error) { listener, err := net.Listen("tcp", addr) if err != nil { return nil, fmt.Errorf("listen %s: %w", addr, err) } - return StartFileTransferServerWithReadiness(listener, stagingDir, modelsDir, dataDir, token, maxUploadSize, readiness, logStore...) + return StartFileTransferServerWithRoutes(listener, stagingDir, modelsDir, dataDir, token, maxUploadSize, readiness, extra, logStore...) } // StartFileTransferServerWithListener starts the server on an existing listener. @@ -66,6 +84,12 @@ func StartFileTransferServerWithListener(lis net.Listener, stagingDir, modelsDir // the probe keeps its historical always-200 behaviour for callers that have no // meaningful readiness signal to report. func StartFileTransferServerWithReadiness(lis net.Listener, stagingDir, modelsDir, dataDir, token string, maxUploadSize int64, readiness *WorkerReadiness, logStore ...*model.BackendLogStore) (*http.Server, error) { + return StartFileTransferServerWithRoutes(lis, stagingDir, modelsDir, dataDir, token, maxUploadSize, readiness, nil, logStore...) +} + +// StartFileTransferServerWithRoutes is StartFileTransferServerWithReadiness +// plus an extra authenticated route set. See AuthenticatedRoutes. +func StartFileTransferServerWithRoutes(lis net.Listener, stagingDir, modelsDir, dataDir, token string, maxUploadSize int64, readiness *WorkerReadiness, extra *AuthenticatedRoutes, logStore ...*model.BackendLogStore) (*http.Server, error) { if err := os.MkdirAll(stagingDir, 0750); err != nil { return nil, fmt.Errorf("creating staging dir %s: %w", stagingDir, err) } @@ -165,8 +189,27 @@ func StartFileTransferServerWithReadiness(lis net.Listener, stagingDir, modelsDi // Readiness: "can this worker actually accept work?" See WorkerReadiness. mux.HandleFunc("/readyz", probe(readiness.Check)) + if extra != nil && extra.Register != nil && extra.Prefix != "" { + extraMux := http.NewServeMux() + extra.Register(extraMux) + mux.Handle(extra.Prefix, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if !checkBearerToken(r, token) { + xlog.Debug("worker HTTP server: unauthorized request on an extra route", + "method", r.Method, "path", r.URL.Path, "remote", r.RemoteAddr) + http.Error(w, "unauthorized", http.StatusUnauthorized) + return + } + extraMux.ServeHTTP(w, r) + })) + } + addr := lis.Addr().String() server := &http.Server{ + // Addr is informational here: Serve takes the listener, not this + // field. It is set so a caller that asked for port 0 can learn the + // port it actually got without threading a second return value + // through every wrapper above. + Addr: addr, Handler: mux, ReadHeaderTimeout: 30 * time.Second, // prevent slowloris; does not affect body reads } diff --git a/core/services/nodes/install_progress_publisher.go b/core/services/nodes/install_progress_publisher.go index 60eacb711..88a2d16ee 100644 --- a/core/services/nodes/install_progress_publisher.go +++ b/core/services/nodes/install_progress_publisher.go @@ -8,9 +8,14 @@ import ( ) // DebouncedInstallProgressPublisher buffers backend-install download ticks -// and publishes them to the per-op NATS progress subject at most once per -// `interval`. Always publishes the final event on Flush so the UI sees the -// terminal percentage. +// and hands them to its emit sink at most once per `interval`. Always emits the +// final event on Flush so the UI sees the terminal percentage. +// +// The sink is a function rather than a NATS subject because the same debounce +// now bounds two carriers: the NATS progress subject an agent node still +// publishes on, and a line written into the streaming HTTP response a worker +// serves over its tunnel. Keeping one debouncer keeps the ~4/s tick bound a +// single fact instead of one per carrier. // // Behavior: leading-edge debounce. The first OnDownload after a quiet window // publishes immediately; subsequent ticks within `interval` only buffer the @@ -18,13 +23,12 @@ import ( // keeps the wire chatter bounded (~4 events per second at 250ms) while // still surfacing every meaningful percentage jump. // -// Lock ordering: never hold p.mu across a Publish call. Publish hits the -// NATS client which may block on a slow link, and we don't want a stalled -// network to stall the underlying gallery download loop. +// Lock ordering: never hold p.mu across an emit call. The sink writes to the +// network, which may block on a slow link, and we don't want a stalled network +// to stall the underlying gallery download loop. type DebouncedInstallProgressPublisher struct { mu sync.Mutex - client messaging.MessagingClient - subject string + emit func(messaging.BackendInstallProgressEvent) nodeID string opID string backend string @@ -34,13 +38,24 @@ type DebouncedInstallProgressPublisher struct { timer *time.Timer } -// NewDebouncedInstallProgressPublisher constructs a publisher for one -// install operation. interval is the leading-edge debounce window -// (~250ms in production). +// NewDebouncedInstallProgressPublisher constructs a publisher for one install +// operation that publishes to the per-op NATS progress subject. interval is the +// leading-edge debounce window (~250ms in production). func NewDebouncedInstallProgressPublisher(client messaging.MessagingClient, nodeID, opID, backend string, interval time.Duration) *DebouncedInstallProgressPublisher { + subject := messaging.SubjectNodeBackendInstallProgress(nodeID, opID) + return NewDebouncedInstallProgressSink(func(ev messaging.BackendInstallProgressEvent) { + _ = client.Publish(subject, ev) + }, nodeID, opID, backend, interval) +} + +// NewDebouncedInstallProgressSink constructs a publisher for one install +// operation that hands each debounced event to emit. +// +// emit is called with p.mu released, so a sink that blocks on a slow link +// cannot stall the gallery download loop that feeds it. +func NewDebouncedInstallProgressSink(emit func(messaging.BackendInstallProgressEvent), nodeID, opID, backend string, interval time.Duration) *DebouncedInstallProgressPublisher { return &DebouncedInstallProgressPublisher{ - client: client, - subject: messaging.SubjectNodeBackendInstallProgress(nodeID, opID), + emit: emit, nodeID: nodeID, opID: opID, backend: backend, @@ -70,7 +85,7 @@ func (p *DebouncedInstallProgressPublisher) OnDownload(file, current, total stri p.lastPublishedAt = now p.pending = nil p.mu.Unlock() - _ = p.client.Publish(p.subject, ev) + p.emit(ev) return } // Within the window: buffer the latest event and arm a trailing @@ -85,8 +100,8 @@ func (p *DebouncedInstallProgressPublisher) OnDownload(file, current, total stri } // flushPending is the trailing-edge publisher fired by the AfterFunc timer. -// It clears the pending slot under the lock, then publishes outside the -// lock so Publish never blocks an in-progress OnDownload call. +// It clears the pending slot under the lock, then emits outside the lock so the +// sink never blocks an in-progress OnDownload call. func (p *DebouncedInstallProgressPublisher) flushPending() { p.mu.Lock() p.timer = nil @@ -97,14 +112,14 @@ func (p *DebouncedInstallProgressPublisher) flushPending() { } p.mu.Unlock() if pending != nil { - _ = p.client.Publish(p.subject, *pending) + p.emit(*pending) } } -// Flush publishes any pending buffered event synchronously and stops the -// pending timer. Safe to call multiple times. Callers MUST defer Flush -// after constructing the publisher so the terminal percentage reaches the -// master even on error returns. +// Flush emits any pending buffered event synchronously and stops the pending +// timer. Safe to call multiple times. Callers MUST defer Flush after +// constructing the publisher so the terminal percentage reaches the master even +// on error returns. func (p *DebouncedInstallProgressPublisher) Flush() { p.mu.Lock() if p.timer != nil { @@ -115,6 +130,6 @@ func (p *DebouncedInstallProgressPublisher) Flush() { p.pending = nil p.mu.Unlock() if pending != nil { - _ = p.client.Publish(p.subject, *pending) + p.emit(*pending) } } diff --git a/core/services/worker/control_routes.go b/core/services/worker/control_routes.go new file mode 100644 index 000000000..dec8e70fc --- /dev/null +++ b/core/services/worker/control_routes.go @@ -0,0 +1,383 @@ +package worker + +import ( + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "sync" + + "github.com/mudler/LocalAI/core/services/messaging" + "github.com/mudler/LocalAI/core/services/workerctl" + "github.com/mudler/xlog" +) + +// The worker's control plane, served as ordinary HTTP on the loopback server +// the worker already runs, and reached only through the tunnel's `http` stream +// tag. It replaces ten NATS subscriptions. +// +// HTTP and not a new stream tag, and the reason is not convenience. Every one +// of correlation, per-request deadlines, unbounded payloads and a progress +// stream is something a new tag would have had to invent, and each is a place +// this branch has already put a defect. Riding the tag that already exists also +// adds nothing to the worker's stream-refusal vocabulary, which is the table a +// frontend decides what to reap on. And a control RPC to a worker another +// replica holds takes the SAME relay the inference path takes, which is the +// path that has been measured, rather than a second one that has not. +// +// A handler's own failure is a 200 carrying a reply with Error set, NOT a 5xx. +// The distinction is load-bearing: the frontend maps a transport failure onto +// "this frontend has no route to that worker", which nothing may reap on, and a +// worker's answer onto evidence a reap guard MAY act on. A handler that +// answered 500 for "the install failed" would put the worker's own verdict into +// the bucket reserved for a broken link. Only a failure to read or route the +// request is a non-2xx. + +// maxControlRequestBytes bounds a control request body. +// +// The largest real body is BackendInstallRequest.BackendGalleries, a serialized +// gallery list of a few hundred kilobytes. Eight megabytes is therefore not a +// size the protocol needs: it is a defence against a body that never ends, +// arriving on a boundary this worker now serves. +const maxControlRequestBytes = 8 << 20 + +// maxEchoedPathBytes bounds how much of an unknown control path the 404 body +// repeats back. The path is caller-controlled and the answer exists to be read +// in a log line, so a caller cannot make this worker echo a request-sized +// string into one. +const maxEchoedPathBytes = 128 + +// installFunc and upgradeFunc are the shapes of the two long-running verbs. +// They exist as named types so the fields that override them below read as one +// thing rather than as two inline signatures. +type installFunc func(ctx context.Context, req messaging.BackendInstallRequest, force bool, + onProgress func(messaging.BackendInstallProgressEvent)) (string, error) + +type upgradeFunc func(ctx context.Context, req messaging.BackendUpgradeRequest, + onProgress func(messaging.BackendInstallProgressEvent)) ([]string, error) + +// installer and upgrader return the implementation the streaming verbs call. +// +// The override fields exist so the ROUTING can be specced without a gallery, a +// registry or a real download, which is the same argument tunnelServices was +// extracted under: the part of this file that can put a worker's verdict in the +// wrong bucket is the part that has nothing to do with installing anything. +func (s *backendSupervisor) installer() installFunc { + if s.installFn != nil { + return s.installFn + } + return s.installBackend +} + +func (s *backendSupervisor) upgrader() upgradeFunc { + if s.upgradeFn != nil { + return s.upgradeFn + } + return s.upgradeBackend +} + +// RegisterControlRoutes mounts every control verb on mux. +// +// The caller is responsible for putting mux behind authentication; see +// nodes.AuthenticatedRoutes, which is how the worker mounts this so the control +// plane shares one bearer check with the file routes rather than growing a +// second one. +func (s *backendSupervisor) RegisterControlRoutes(mux *http.ServeMux) { + // post registers one JSON-in, JSON-out verb. reply may be nil, in which + // case the verb answers 204: that is the shape the two former + // publish-no-reply subjects take. + post := func(path string, h func(ctx context.Context, body []byte) (any, error)) { + mux.HandleFunc(path, func(w http.ResponseWriter, r *http.Request) { + body, ok := readControlBody(w, r) + if !ok { + return + } + reply, err := h(r.Context(), body) + if err != nil { + // Reaching here means the request could not be READ or routed, + // which is this worker failing rather than answering. A verb's + // own failure never reaches here: it comes back as a reply with + // Error set, and a 200. + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + if reply == nil { + w.WriteHeader(http.StatusNoContent) + return + } + w.Header().Set("Content-Type", "application/json") + if encErr := json.NewEncoder(w).Encode(reply); encErr != nil { + xlog.Debug("worker control reply could not be written", "path", path, "error", encErr) + } + }) + } + + post(workerctl.PathModelsRunning, func(context.Context, []byte) (any, error) { + return messaging.ModelsRunningReply{Models: s.runningModels()}, nil + }) + + post(workerctl.PathModelStop, func(_ context.Context, body []byte) (any, error) { + var req messaging.ModelStopRequest + if err := json.Unmarshal(body, &req); err != nil { + return nil, fmt.Errorf("invalid model.stop request: %w", err) + } + return s.stopModelExact(req), nil + }) + + post(workerctl.PathBackendList, func(context.Context, []byte) (any, error) { + return s.backendList(), nil + }) + + post(workerctl.PathBackendDelete, func(_ context.Context, body []byte) (any, error) { + var req messaging.BackendDeleteRequest + if err := json.Unmarshal(body, &req); err != nil { + return nil, fmt.Errorf("invalid backend.delete request: %w", err) + } + return s.deleteBackend(req), nil + }) + + post(workerctl.PathModelUnload, func(ctx context.Context, body []byte) (any, error) { + var req messaging.ModelUnloadRequest + if err := json.Unmarshal(body, &req); err != nil { + return nil, fmt.Errorf("invalid model.unload request: %w", err) + } + return s.unloadModel(ctx, req), nil + }) + + post(workerctl.PathModelDelete, func(_ context.Context, body []byte) (any, error) { + var req messaging.ModelDeleteRequest + if err := json.Unmarshal(body, &req); err != nil { + return nil, fmt.Errorf("invalid model.delete request: %w", err) + } + return s.deleteModel(req), nil + }) + + post(workerctl.PathBackendStop, func(_ context.Context, body []byte) (any, error) { + req, stopAll, err := decodeBackendStopRequest(body) + if err != nil { + return nil, fmt.Errorf("invalid backend.stop request: %w", err) + } + s.stopBackends(req, stopAll) + return nil, nil + }) + + post(workerctl.PathNodeStop, func(context.Context, []byte) (any, error) { + // The signal is sent before the 204 is written, and that ordering is + // safe rather than lucky: sigCh is buffered and this send never blocks, + // and the shutdown it starts is a graceful one that waits for this + // request to finish before closing the listener. + s.signalNodeStop() + return nil, nil + }) + + // The two streaming verbs. They write NDJSON rather than one JSON object, + // so they do not go through post. + mux.HandleFunc(workerctl.PathBackendInstall, s.serveInstall) + mux.HandleFunc(workerctl.PathBackendUpgrade, s.serveUpgrade) + + mux.HandleFunc(workerctl.Prefix, func(w http.ResponseWriter, r *http.Request) { + // The catch-all. A path under the control prefix that no verb claims is + // a frontend newer than this worker, and the body says so, because a + // bare 404 through a tunnel is indistinguishable from a proxy fault. + http.Error(w, "unknown worker control path "+truncate(r.URL.Path, maxEchoedPathBytes), http.StatusNotFound) + }) +} + +// readControlBody enforces the two things every control verb requires of a +// request: that it is a POST, and that its body is bounded. +// +// A GET is refused rather than served because a control verb is a command, and +// a liveness probe, a link prefetch or a browser address bar must not be able +// to stop a node. +func readControlBody(w http.ResponseWriter, r *http.Request) ([]byte, bool) { + if r.Method != http.MethodPost { + http.Error(w, "control verbs are POST only", http.StatusMethodNotAllowed) + return nil, false + } + body, err := io.ReadAll(http.MaxBytesReader(w, r.Body, maxControlRequestBytes)) + if err != nil { + // A body this worker could not READ is not an answer about any + // backend, so it must not look like one: 400 is what the frontend maps + // onto "the request was rejected", never onto "that model is gone". + http.Error(w, "reading the control request body: "+err.Error(), http.StatusBadRequest) + return nil, false + } + return body, true +} + +// truncate bounds a caller-controlled string that is about to be echoed. +// It cuts on a rune boundary: a byte-wise cut can split a multi-byte rune, and +// the half-rune then travels as a replacement character through every log and +// UI that reads it. +func truncate(s string, max int) string { + if len(s) <= max { + return s + } + cut := max + for cut > 0 && !isRuneStart(s[cut]) { + cut-- + } + return s[:cut] + "…" +} + +// isRuneStart reports whether b begins a UTF-8 rune (i.e. is not a +// continuation byte). +func isRuneStart(b byte) bool { return b&0xC0 != 0x80 } + +// ndjsonStream writes the Envelope lines of one streaming control response. +// +// The mutex is not optional. A progress line can be emitted from the debounce +// timer's own goroutine, so without serialization a progress write can +// interleave with the terminal reply write and put a torn line on the wire, +// and the frontend's contract is that the reply line is the LAST thing on the +// body. done is what enforces that half: once the reply is written, a late +// progress line is dropped rather than appended after it. +type ndjsonStream struct { + mu sync.Mutex + w http.ResponseWriter + enc *json.Encoder + done bool +} + +func newNDJSONStream(w http.ResponseWriter) *ndjsonStream { + w.Header().Set("Content-Type", workerctl.ContentTypeStream) + // A streaming body must not be buffered into a guessed content type: the + // caller reads line by line and the first line may be minutes before the + // last. + w.Header().Set("X-Content-Type-Options", "nosniff") + w.WriteHeader(http.StatusOK) + return &ndjsonStream{w: w, enc: json.NewEncoder(w)} +} + +// progress writes one progress line. It is a no-op once the reply has been +// written. +func (n *ndjsonStream) progress(ev messaging.BackendInstallProgressEvent) { + raw, err := json.Marshal(ev) + if err != nil { + xlog.Debug("worker control progress event could not be marshalled", "error", err) + return + } + n.mu.Lock() + defer n.mu.Unlock() + if n.done { + return + } + n.write(workerctl.Envelope{Progress: raw}) +} + +// reply writes the single terminal reply line and closes the stream to any +// further progress. +func (n *ndjsonStream) reply(v any) { + raw, err := json.Marshal(v) + if err != nil { + // The caller is waiting for a terminal line and will otherwise read to + // EOF without one, which it cannot tell apart from a truncated + // response. Send a reply it can decode as a failure of THIS worker's + // own making rather than sending nothing. + xlog.Error("worker control reply could not be marshalled", "error", err) + raw = json.RawMessage(`{"success":false,"error":"the worker could not encode its own reply"}`) + } + n.mu.Lock() + defer n.mu.Unlock() + if n.done { + return + } + n.write(workerctl.Envelope{Reply: raw}) + n.done = true +} + +// write encodes one envelope and flushes it. Callers hold n.mu. +func (n *ndjsonStream) write(env workerctl.Envelope) { + if err := n.enc.Encode(env); err != nil { + xlog.Debug("worker control stream line could not be written", "error", err) + return + } + if f, ok := n.w.(http.Flusher); ok { + f.Flush() + } +} + +// serveInstall answers backend.install, streaming download progress ahead of +// the single terminal reply. +// +// The NATS handler this replaces ran its work on a fresh goroutine so a slow +// install could not head-of-line-block the one subscription every install +// arrived on. Over HTTP each request already has its own goroutine, so there is +// nothing left to block and no goroutine is started here. Per-backend +// serialization is unchanged and still comes from lockBackend, which is what +// actually prevented two requests racing the gallery directory. +func (s *backendSupervisor) serveInstall(w http.ResponseWriter, r *http.Request) { + body, ok := readControlBody(w, r) + if !ok { + return + } + var req messaging.BackendInstallRequest + if err := json.Unmarshal(body, &req); err != nil { + http.Error(w, fmt.Sprintf("invalid backend.install request: %v", err), http.StatusBadRequest) + return + } + xlog.Info("Serving backend.install", "backend", req.Backend, "model", req.ModelID) + + stream := newNDJSONStream(w) + + release := s.lockBackend(req.Backend) + defer release() + + // req.Force=true is the legacy path used by pre-2026-05-08 masters that + // don't know about backend.upgrade. Honor it so a rolling update with new + // worker + old master keeps working; new masters send to backend.upgrade + // instead. + addr, err := s.installer()(r.Context(), req, req.Force, stream.progress) + if err != nil { + xlog.Error("Failed to install backend", "error", err) + stream.reply(messaging.BackendInstallReply{Success: false, Error: err.Error()}) + return + } + + // The address goes back exactly as the process listens on it. It used to be + // rewritten onto this worker's advertise host, which made the reply the + // worker's third advertisement site; the frontend now reads only the port + // out of it and dials nothing. + stream.reply(messaging.BackendInstallReply{Success: true, WorkerLocalAddress: addr}) +} + +// serveUpgrade answers backend.upgrade, a force-reinstall, on the same +// streaming shape as install. +func (s *backendSupervisor) serveUpgrade(w http.ResponseWriter, r *http.Request) { + body, ok := readControlBody(w, r) + if !ok { + return + } + var req messaging.BackendUpgradeRequest + if err := json.Unmarshal(body, &req); err != nil { + http.Error(w, fmt.Sprintf("invalid backend.upgrade request: %v", err), http.StatusBadRequest) + return + } + xlog.Info("Serving backend.upgrade", "backend", req.Backend) + + stream := newNDJSONStream(w) + + release := s.lockBackend(req.Backend) + defer release() + + // stopped is meaningful even on the error paths: it lists processes already + // terminated (and ports already recycled) before the failure, so the + // controller must drop those rows regardless of the outcome. + stopped, err := s.upgrader()(r.Context(), req, stream.progress) + if err != nil { + xlog.Error("Failed to upgrade backend", "error", err) + stream.reply(messaging.BackendUpgradeReply{ + Success: false, + Error: err.Error(), + StoppedProcessKeys: stopped, + ReportsStoppedProcesses: true, + }) + return + } + stream.reply(messaging.BackendUpgradeReply{ + Success: true, + StoppedProcessKeys: stopped, + ReportsStoppedProcesses: true, + }) +} diff --git a/core/services/worker/control_routes_test.go b/core/services/worker/control_routes_test.go new file mode 100644 index 000000000..fb8576e71 --- /dev/null +++ b/core/services/worker/control_routes_test.go @@ -0,0 +1,524 @@ +package worker + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "io" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "strings" + "sync" + "syscall" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + + "github.com/mudler/LocalAI/core/services/messaging" + "github.com/mudler/LocalAI/core/services/nodes" + "github.com/mudler/LocalAI/core/services/workerctl" + "github.com/mudler/LocalAI/pkg/system" +) + +// lastReplyOf drains an NDJSON control body and decodes the single terminal +// reply line into out. It fails the spec when the body carries no reply line, +// which is the shape the frontend cannot recover from: it would sit reading a +// body that already ended. +func lastReplyOf(body io.Reader, out any) error { + GinkgoHelper() + dec := json.NewDecoder(body) + var reply json.RawMessage + for { + var env workerctl.Envelope + err := dec.Decode(&env) + if errors.Is(err, io.EOF) { + break + } + Expect(err).NotTo(HaveOccurred()) + if env.Reply != nil { + reply = env.Reply + } + } + Expect(reply).NotTo(BeNil(), "the control body carried no terminal reply line") + return json.Unmarshal(reply, out) +} + +// envelopeKindsOf reports the order of progress/reply lines in an NDJSON body. +func envelopeKindsOf(body io.Reader) []string { + GinkgoHelper() + var kinds []string + dec := json.NewDecoder(body) + for { + var env workerctl.Envelope + err := dec.Decode(&env) + if errors.Is(err, io.EOF) { + break + } + Expect(err).NotTo(HaveOccurred()) + if env.Reply != nil { + kinds = append(kinds, "reply") + continue + } + Expect(env.Progress).NotTo(BeNil(), "an envelope carried neither progress nor reply") + kinds = append(kinds, "progress") + } + return kinds +} + +var _ = Describe("worker control routes", func() { + var ( + sup *backendSupervisor + srv *httptest.Server + sigCh chan os.Signal + ) + + BeforeEach(func() { + sigCh = make(chan os.Signal, 1) + sup = &backendSupervisor{ + cfg: &Config{}, + nodeID: "node-under-test", + sigCh: sigCh, + processes: map[string]*backendProcess{}, + } + mux := http.NewServeMux() + sup.RegisterControlRoutes(mux) + srv = httptest.NewServer(mux) + DeferCleanup(srv.Close) + }) + + // post marshals, POSTs, and hands back the response. + post := func(path string, body any) *http.Response { + GinkgoHelper() + buf, err := json.Marshal(body) + Expect(err).NotTo(HaveOccurred()) + req, err := http.NewRequest(http.MethodPost, srv.URL+path, bytes.NewReader(buf)) + Expect(err).NotTo(HaveOccurred()) + req.Header.Set("Content-Type", "application/json") + resp, err := srv.Client().Do(req) + Expect(err).NotTo(HaveOccurred()) + DeferCleanup(func() { _ = resp.Body.Close() }) + return resp + } + + It("answers models.running with the worker's process table", func() { + resp := post(workerctl.PathModelsRunning, messaging.ModelsRunningRequest{}) + Expect(resp.StatusCode).To(Equal(http.StatusOK)) + var reply messaging.ModelsRunningReply + Expect(json.NewDecoder(resp.Body).Decode(&reply)).To(Succeed()) + Expect(reply.Models).To(BeEmpty()) + }) + + It("refuses a GET on a control route, so a probe cannot fire a command", func() { + resp, err := srv.Client().Get(srv.URL + workerctl.PathNodeStop) + Expect(err).NotTo(HaveOccurred()) + DeferCleanup(func() { _ = resp.Body.Close() }) + Expect(resp.StatusCode).To(Equal(http.StatusMethodNotAllowed)) + Expect(sigCh).NotTo(Receive(), "a GET must not have signalled shutdown") + }) + + It("refuses a GET on the streaming install route too", func() { + resp, err := srv.Client().Get(srv.URL + workerctl.PathBackendInstall) + Expect(err).NotTo(HaveOccurred()) + DeferCleanup(func() { _ = resp.Body.Close() }) + Expect(resp.StatusCode).To(Equal(http.StatusMethodNotAllowed)) + }) + + It("answers an unknown control path with 404 and a body that names the prefix", func() { + resp := post(workerctl.Prefix+"no-such-verb", struct{}{}) + Expect(resp.StatusCode).To(Equal(http.StatusNotFound)) + body, err := io.ReadAll(resp.Body) + Expect(err).NotTo(HaveOccurred()) + // A mixed-version deployment has to be diagnosable from one line in a + // log, not from a bare 404 that looks like a proxy. + Expect(string(body)).To(ContainSubstring("control")) + }) + + It("bounds the unknown path it echoes back, so a long URL cannot be reflected wholesale", func() { + long := strings.Repeat("a", 4096) + resp := post(workerctl.Prefix+long, struct{}{}) + Expect(resp.StatusCode).To(Equal(http.StatusNotFound)) + body, err := io.ReadAll(resp.Body) + Expect(err).NotTo(HaveOccurred()) + Expect(len(body)).To(BeNumerically("<", 512)) + }) + + It("reports a malformed request body as 400 and does not touch the process table", func() { + resp, err := srv.Client().Post(srv.URL+workerctl.PathModelStop, "application/json", + strings.NewReader("{not json")) + Expect(err).NotTo(HaveOccurred()) + DeferCleanup(func() { _ = resp.Body.Close() }) + Expect(resp.StatusCode).To(Equal(http.StatusBadRequest)) + }) + + It("refuses a request body larger than the control bound", func() { + // The body is DELIBERATELY well-formed JSON for the verb it is sent to. + // A body of garbage would be rejected by the decoder whether or not the + // bound exists, so a spec built on one is green with the bound removed + // and pins nothing. This one is refused only because of the bound. + oversized := []byte(`{"process_key":"` + strings.Repeat("a", maxControlRequestBytes) + `"}`) + Expect(len(oversized)).To(BeNumerically(">", maxControlRequestBytes)) + resp, err := srv.Client().Post(srv.URL+workerctl.PathModelStop, "application/json", + bytes.NewReader(oversized)) + Expect(err).NotTo(HaveOccurred()) + DeferCleanup(func() { _ = resp.Body.Close() }) + Expect(resp.StatusCode).To(Equal(http.StatusBadRequest)) + }) + + It("accepts a large but in-bound body, so the bound is a ceiling and not a shape", func() { + // BackendInstallRequest.BackendGalleries is a serialized gallery list of + // a few hundred kilobytes on a real cluster, so the bound has to sit + // well above that rather than at the size of a typical request. + sup.installFn = func(context.Context, messaging.BackendInstallRequest, bool, + func(messaging.BackendInstallProgressEvent)) (string, error) { + return "127.0.0.1:1", nil + } + big := []byte(`{"backend":"mock","backend_galleries":"` + strings.Repeat("a", 1<<20) + `"}`) + Expect(len(big)).To(BeNumerically("<", maxControlRequestBytes)) + resp, err := srv.Client().Post(srv.URL+workerctl.PathBackendInstall, "application/json", + bytes.NewReader(big)) + Expect(err).NotTo(HaveOccurred()) + DeferCleanup(func() { _ = resp.Body.Close() }) + Expect(resp.StatusCode).To(Equal(http.StatusOK)) + }) + + It("answers backend.stop with 204, the shape a fire-and-forget verb takes", func() { + resp := post(workerctl.PathBackendStop, messaging.BackendStopRequest{Backend: "no-such-backend"}) + Expect(resp.StatusCode).To(Equal(http.StatusNoContent)) + }) + + It("answers node.stop with 204 and signals shutdown", func() { + resp := post(workerctl.PathNodeStop, struct{}{}) + Expect(resp.StatusCode).To(Equal(http.StatusNoContent)) + Eventually(sigCh).Should(Receive(Equal(os.Signal(syscall.SIGTERM)))) + }) + + It("answers model.unload with the worker's own reply rather than a transport error", func() { + resp := post(workerctl.PathModelUnload, messaging.ModelUnloadRequest{ModelName: "m"}) + Expect(resp.StatusCode).To(Equal(http.StatusOK)) + var reply messaging.ModelUnloadReply + Expect(json.NewDecoder(resp.Body).Decode(&reply)).To(Succeed()) + Expect(reply.Success).To(BeTrue()) + }) + + It("answers model.stop for an unknown process with the worker's verdict, not a 5xx", func() { + // The whole invariant of the phase: "that backend is not there" is the + // WORKER answering, and it must never arrive as the status code a + // frontend reads as "I could not reach that worker". + resp := post(workerctl.PathModelStop, messaging.ModelStopRequest{ProcessKey: "ghost#0"}) + Expect(resp.StatusCode).To(Equal(http.StatusOK)) + var reply messaging.ModelStopReply + Expect(json.NewDecoder(resp.Body).Decode(&reply)).To(Succeed()) + Expect(reply.Matched).To(BeFalse()) + Expect(reply.ProcessKey).To(Equal("ghost#0")) + }) + + Context("streaming install", func() { + It("writes every progress line before the single terminal reply line", func() { + sup.installFn = func(_ context.Context, _ messaging.BackendInstallRequest, _ bool, + progress func(messaging.BackendInstallProgressEvent)) (string, error) { + progress(messaging.BackendInstallProgressEvent{Percentage: 50}) + progress(messaging.BackendInstallProgressEvent{Percentage: 100}) + return "127.0.0.1:50051", nil + } + resp := post(workerctl.PathBackendInstall, messaging.BackendInstallRequest{Backend: "mock", OpID: "op-1"}) + Expect(resp.StatusCode).To(Equal(http.StatusOK)) + Expect(resp.Header.Get("Content-Type")).To(Equal(workerctl.ContentTypeStream)) + Expect(envelopeKindsOf(resp.Body)).To(Equal([]string{"progress", "progress", "reply"})) + }) + + It("carries the address the backend actually listens on back in the reply", func() { + sup.installFn = func(_ context.Context, _ messaging.BackendInstallRequest, _ bool, + _ func(messaging.BackendInstallProgressEvent)) (string, error) { + return "127.0.0.1:50099", nil + } + resp := post(workerctl.PathBackendInstall, messaging.BackendInstallRequest{Backend: "mock"}) + var reply messaging.BackendInstallReply + Expect(lastReplyOf(resp.Body, &reply)).To(Succeed()) + Expect(reply.Success).To(BeTrue()) + Expect(reply.WorkerLocalAddress).To(Equal("127.0.0.1:50099")) + }) + + It("still writes a terminal reply line when the install fails", func() { + sup.installFn = func(_ context.Context, _ messaging.BackendInstallRequest, _ bool, + _ func(messaging.BackendInstallProgressEvent)) (string, error) { + return "", errors.New("boom") + } + resp := post(workerctl.PathBackendInstall, messaging.BackendInstallRequest{Backend: "mock", OpID: "op-2"}) + // 200 with a failed reply, not 500: the WORKER answered, and a 5xx + // is what the frontend reads as the worker not answering at all. + Expect(resp.StatusCode).To(Equal(http.StatusOK)) + var reply messaging.BackendInstallReply + Expect(lastReplyOf(resp.Body, &reply)).To(Succeed()) + Expect(reply.Success).To(BeFalse()) + Expect(reply.Error).To(ContainSubstring("boom")) + }) + + It("passes the request's Force flag through, which is the legacy upgrade path", func() { + forced := make(chan bool, 1) + sup.installFn = func(_ context.Context, _ messaging.BackendInstallRequest, force bool, + _ func(messaging.BackendInstallProgressEvent)) (string, error) { + forced <- force + return "127.0.0.1:1", nil + } + post(workerctl.PathBackendInstall, messaging.BackendInstallRequest{Backend: "mock", Force: true}) + Expect(forced).To(Receive(BeTrue())) + }) + + It("gives the install the caller's context, so a frontend that gave up stops the work", func() { + started := make(chan struct{}) + observed := make(chan error, 1) + // abandon releases the handler if the context never arrives, so a + // build where the caller's budget was dropped fails this spec + // instead of parking a goroutine until the suite times out. + abandon := make(chan struct{}) + DeferCleanup(func() { close(abandon) }) + sup.installFn = func(ctx context.Context, _ messaging.BackendInstallRequest, _ bool, + _ func(messaging.BackendInstallProgressEvent)) (string, error) { + close(started) + select { + case <-ctx.Done(): + case <-abandon: + } + observed <- ctx.Err() + return "", ctx.Err() + } + ctx, cancel := context.WithCancel(context.Background()) + req, err := http.NewRequestWithContext(ctx, http.MethodPost, + srv.URL+workerctl.PathBackendInstall, strings.NewReader(`{"backend":"mock"}`)) + Expect(err).NotTo(HaveOccurred()) + go func() { + defer GinkgoRecover() + resp, derr := srv.Client().Do(req) + if derr == nil { + _ = resp.Body.Close() + } + }() + Eventually(started).Should(BeClosed()) + cancel() + Eventually(observed).Should(Receive(MatchError(context.Canceled))) + }) + + It("reports a malformed install body as 400 rather than as a failed install", func() { + resp, err := srv.Client().Post(srv.URL+workerctl.PathBackendInstall, "application/json", + strings.NewReader("{not json")) + Expect(err).NotTo(HaveOccurred()) + DeferCleanup(func() { _ = resp.Body.Close() }) + Expect(resp.StatusCode).To(Equal(http.StatusBadRequest)) + }) + }) + + Context("streaming upgrade", func() { + It("writes progress before the terminal reply and reports what it stopped", func() { + sup.upgradeFn = func(_ context.Context, _ messaging.BackendUpgradeRequest, + progress func(messaging.BackendInstallProgressEvent)) ([]string, error) { + progress(messaging.BackendInstallProgressEvent{Percentage: 10}) + return []string{"m#0"}, nil + } + resp := post(workerctl.PathBackendUpgrade, messaging.BackendUpgradeRequest{Backend: "mock", OpID: "op-3"}) + Expect(resp.Header.Get("Content-Type")).To(Equal(workerctl.ContentTypeStream)) + body, err := io.ReadAll(resp.Body) + Expect(err).NotTo(HaveOccurred()) + Expect(envelopeKindsOf(bytes.NewReader(body))).To(Equal([]string{"progress", "reply"})) + var reply messaging.BackendUpgradeReply + Expect(lastReplyOf(bytes.NewReader(body), &reply)).To(Succeed()) + Expect(reply.Success).To(BeTrue()) + Expect(reply.StoppedProcessKeys).To(Equal([]string{"m#0"})) + Expect(reply.ReportsStoppedProcesses).To(BeTrue()) + }) + + It("reports the processes it stopped even when the upgrade then failed", func() { + // stopped is meaningful on the error path: those ports are already + // recycled, so the controller must drop their rows regardless. + sup.upgradeFn = func(_ context.Context, _ messaging.BackendUpgradeRequest, + _ func(messaging.BackendInstallProgressEvent)) ([]string, error) { + return []string{"m#0"}, errors.New("upgrade boom") + } + resp := post(workerctl.PathBackendUpgrade, messaging.BackendUpgradeRequest{Backend: "mock"}) + Expect(resp.StatusCode).To(Equal(http.StatusOK)) + var reply messaging.BackendUpgradeReply + Expect(lastReplyOf(resp.Body, &reply)).To(Succeed()) + Expect(reply.Success).To(BeFalse()) + Expect(reply.Error).To(ContainSubstring("upgrade boom")) + Expect(reply.StoppedProcessKeys).To(Equal([]string{"m#0"})) + Expect(reply.ReportsStoppedProcesses).To(BeTrue()) + }) + }) +}) + +// These specs pin the PRODUCTION wiring rather than the handler. Every defect +// this branch has shipped in the last two tasks was a call site no spec +// touched, and "the control plane is mounted, on the same listener, behind the +// same token" is exactly that kind of fact: the handlers can be perfect and a +// worker that never mounts them answers 404 to every command while looking +// healthy on /healthz. +var _ = Describe("the worker's HTTP server", func() { + const token = "worker-token" + + var ( + srv *http.Server + base string + sup *backendSupervisor + ) + + BeforeEach(func() { + dir := GinkgoT().TempDir() + st, err := system.GetSystemState( + system.WithModelPath(dir), + system.WithBackendPath(filepath.Join(dir, "backends")), + system.WithBackendSystemPath(filepath.Join(dir, "backends-system")), + ) + Expect(err).NotTo(HaveOccurred()) + sup = &backendSupervisor{ + cfg: &Config{ModelsPath: dir}, + systemState: st, + nodeID: "node-under-test", + sigCh: make(chan os.Signal, 1), + processes: map[string]*backendProcess{}, + // This Describe is about MOUNTING, so the two verbs that would + // otherwise reach a gallery are scripted. Everything else runs its + // real body against an empty worker. + installFn: func(context.Context, messaging.BackendInstallRequest, bool, + func(messaging.BackendInstallProgressEvent)) (string, error) { + return "127.0.0.1:50051", nil + }, + upgradeFn: func(context.Context, messaging.BackendUpgradeRequest, + func(messaging.BackendInstallProgressEvent)) ([]string, error) { + return nil, nil + }, + } + srv, err = startWorkerHTTPServer("127.0.0.1:0", filepath.Join(dir, "staging"), dir, + filepath.Join(dir, "data"), token, &nodes.WorkerReadiness{}, sup, nil) + Expect(err).NotTo(HaveOccurred()) + DeferCleanup(func() { nodes.ShutdownFileTransferServer(srv) }) + Expect(srv.Addr).NotTo(BeEmpty(), "the worker HTTP server must report the address it bound") + base = "http://" + srv.Addr + }) + + postCtl := func(path, bearer string) *http.Response { + GinkgoHelper() + req, err := http.NewRequest(http.MethodPost, base+path, strings.NewReader("{}")) + Expect(err).NotTo(HaveOccurred()) + if bearer != "" { + req.Header.Set("Authorization", "Bearer "+bearer) + } + resp, err := http.DefaultClient.Do(req) + Expect(err).NotTo(HaveOccurred()) + DeferCleanup(func() { _ = resp.Body.Close() }) + return resp + } + + It("serves the control plane on the same listener as the file routes", func() { + resp := postCtl(workerctl.PathModelsRunning, token) + Expect(resp.StatusCode).To(Equal(http.StatusOK)) + var reply messaging.ModelsRunningReply + Expect(json.NewDecoder(resp.Body).Decode(&reply)).To(Succeed()) + Expect(reply.Models).To(BeEmpty()) + }) + + It("puts the control plane behind the registration token", func() { + Expect(postCtl(workerctl.PathModelsRunning, "").StatusCode).To(Equal(http.StatusUnauthorized)) + Expect(postCtl(workerctl.PathModelsRunning, "wrong").StatusCode).To(Equal(http.StatusUnauthorized)) + }) + + It("mounts every control verb, not just the one this spec reads", func() { + for _, p := range workerctl.AllPaths() { + if p == workerctl.PathNodeStop { + // Firing it would tear down the worker under the other specs. + continue + } + resp := postCtl(p, token) + Expect(resp.StatusCode).NotTo(Equal(http.StatusNotFound), "control verb %q is not mounted", p) + Expect(resp.StatusCode).NotTo(Equal(http.StatusUnauthorized), "control verb %q rejected a valid token", p) + } + }) +}) + +// lockedRecorder is a ResponseWriter safe for concurrent writes. httptest's +// recorder is not, and the concurrency spec below is about what ndjsonStream +// serializes, not about what the recorder does. +type lockedRecorder struct { + mu sync.Mutex + buf bytes.Buffer + hdr http.Header +} + +func newLockedRecorder() *lockedRecorder { return &lockedRecorder{hdr: http.Header{}} } + +func (l *lockedRecorder) Header() http.Header { return l.hdr } + +func (l *lockedRecorder) Write(p []byte) (int, error) { + l.mu.Lock() + defer l.mu.Unlock() + return l.buf.Write(p) +} + +func (l *lockedRecorder) WriteHeader(int) {} + +func (l *lockedRecorder) body() []byte { + l.mu.Lock() + defer l.mu.Unlock() + return append([]byte(nil), l.buf.Bytes()...) +} + +var _ = Describe("the NDJSON control stream", func() { + It("drops a progress line that arrives after the reply, so the reply stays last", func() { + // The debounce timer emits from its own goroutine and can fire after + // the install has already returned. Appending that line would break the + // one thing the frontend relies on to stop reading. + rec := httptest.NewRecorder() + stream := newNDJSONStream(rec) + stream.progress(messaging.BackendInstallProgressEvent{Percentage: 10}) + stream.reply(messaging.BackendInstallReply{Success: true}) + stream.progress(messaging.BackendInstallProgressEvent{Percentage: 100}) + + Expect(envelopeKindsOf(bytes.NewReader(rec.Body.Bytes()))).To(Equal([]string{"progress", "reply"})) + }) + + It("writes at most one reply line even when reply is called twice", func() { + rec := httptest.NewRecorder() + stream := newNDJSONStream(rec) + stream.reply(messaging.BackendInstallReply{Success: true}) + stream.reply(messaging.BackendInstallReply{Success: false, Error: "second"}) + + Expect(envelopeKindsOf(bytes.NewReader(rec.Body.Bytes()))).To(Equal([]string{"reply"})) + Expect(rec.Body.String()).NotTo(ContainSubstring("second")) + }) + + It("names the streaming content type and refuses to let it be sniffed", func() { + rec := httptest.NewRecorder() + newNDJSONStream(rec) + Expect(rec.Header().Get("Content-Type")).To(Equal(workerctl.ContentTypeStream)) + Expect(rec.Header().Get("X-Content-Type-Options")).To(Equal("nosniff")) + }) + + It("serializes concurrent progress against the reply, so no line is torn", func() { + rec := newLockedRecorder() + stream := newNDJSONStream(rec) + + const writers = 16 + start := make(chan struct{}) + var wg sync.WaitGroup + wg.Add(writers) + for i := 0; i < writers; i++ { + go func(n int) { + defer GinkgoRecover() + defer wg.Done() + <-start + stream.progress(messaging.BackendInstallProgressEvent{Percentage: float64(n)}) + }(i) + } + close(start) + stream.reply(messaging.BackendInstallReply{Success: true}) + wg.Wait() + + kinds := envelopeKindsOf(bytes.NewReader(rec.body())) + Expect(kinds).NotTo(BeEmpty()) + Expect(kinds[len(kinds)-1]).To(Equal("reply"), "the reply line must be the last line on the body") + for _, k := range kinds[:len(kinds)-1] { + Expect(k).To(Equal("progress")) + } + }) +}) diff --git a/core/services/worker/install.go b/core/services/worker/install.go index 122b5d266..2390ca28d 100644 --- a/core/services/worker/install.go +++ b/core/services/worker/install.go @@ -56,11 +56,16 @@ func buildProcessKey(modelID, backend string, replicaIndex int) string { // // Returns the gRPC address of the backend process. // +// ctx is the CALLER's budget, carried in from the control request, so a +// frontend that stopped reading stops the download rather than leaving the +// worker pulling gigabytes for a response nobody will read. onProgress receives +// debounced download ticks and may be nil. +// // ProcessKey includes the replica index so a worker with MaxReplicasPerModel>1 // can host multiple processes for the same model on distinct ports. Old // controllers (no replica_index in the request) implicitly target replica 0, // which preserves single-replica behavior. -func (s *backendSupervisor) installBackend(req messaging.BackendInstallRequest, force bool) (string, error) { +func (s *backendSupervisor) installBackend(ctx context.Context, req messaging.BackendInstallRequest, force bool, onProgress func(messaging.BackendInstallProgressEvent)) (string, error) { processKey := buildProcessKey(req.ModelID, req.Backend, int(req.ReplicaIndex)) if !force { @@ -129,16 +134,16 @@ func (s *backendSupervisor) installBackend(req messaging.BackendInstallRequest, galleries = reqGalleries } - // When the master tagged this install with an OpID, stream the - // gallery download progress back to it on the per-op NATS subject. - // Old masters that omit OpID stay on the silent path so they keep - // working without changes. The publisher releases its mutex before - // every Publish so a slow link never stalls the download loop, and - // the deferred Flush guarantees a terminal-percentage event reaches - // the master even when the install errors out. + // When the caller tagged this install with an OpID and is listening for + // progress, stream the gallery download ticks back on the response the + // caller is already reading. Callers that omit OpID stay on the silent path. + // The publisher releases its mutex before every emit so a slow link never + // stalls the download loop, and the deferred Flush guarantees a + // terminal-percentage event reaches the caller even when the install errors + // out. var downloadCb func(file, current, total string, percentage float64) - if req.OpID != "" && s.nats != nil { - publisher := nodes.NewDebouncedInstallProgressPublisher(s.nats, s.nodeID, req.OpID, req.Backend, installProgressDebounce) + if req.OpID != "" && onProgress != nil { + publisher := nodes.NewDebouncedInstallProgressSink(onProgress, s.nodeID, req.OpID, req.Backend, installProgressDebounce) downloadCb = publisher.OnDownload defer publisher.Flush() } @@ -155,14 +160,14 @@ func (s *backendSupervisor) installBackend(req messaging.BackendInstallRequest, if req.URI != "" { xlog.Info("Installing backend from external URI", "backend", req.Backend, "uri", req.URI, "force", force) if err := galleryop.InstallExternalBackend( - context.Background(), galleries, s.systemState, s.ml, downloadCb, req.URI, req.Name, req.Alias, force, s.cfg.RequireBackendIntegrity, + ctx, galleries, s.systemState, s.ml, downloadCb, req.URI, req.Name, req.Alias, force, s.cfg.RequireBackendIntegrity, ); err != nil { return "", fmt.Errorf("installing backend from gallery: %w", err) } } else { xlog.Info("Installing backend from gallery", "backend", req.Backend, "force", force) if err := gallery.InstallBackendFromGallery( - context.Background(), galleries, s.systemState, s.ml, req.Backend, downloadCb, force, s.cfg.RequireBackendIntegrity, + ctx, galleries, s.systemState, s.ml, req.Backend, downloadCb, force, s.cfg.RequireBackendIntegrity, ); err != nil { return "", fmt.Errorf("installing backend from gallery: %w", err) } @@ -191,13 +196,15 @@ func (s *backendSupervisor) installBackend(req messaging.BackendInstallRequest, // It does NOT start any new gRPC process — the next routine model load via // backend.install will spawn a fresh process picking up the new binary. // -// The caller is responsible for holding s.lockBackend(req.Backend). +// The caller is responsible for holding s.lockBackend(req.Backend). ctx is the +// caller's budget and onProgress its progress sink, with the same meaning as on +// installBackend. // // It returns the process keys it terminated so the controller can drop the // NodeModel rows addressing them: an upgrade stops every process using the // binary and starts none back up, recycling their gRPC ports while the rows // still point at those addresses. -func (s *backendSupervisor) upgradeBackend(req messaging.BackendUpgradeRequest) ([]string, error) { +func (s *backendSupervisor) upgradeBackend(ctx context.Context, req messaging.BackendUpgradeRequest, onProgress func(messaging.BackendInstallProgressEvent)) ([]string, error) { // Stop every live process for this backend (peer replicas + the bare // processKey). Same logic as the force branch in installBackend. toStop := s.resolveProcessKeysForBackend(s.backendIdentity(req.Backend)) @@ -228,14 +235,14 @@ func (s *backendSupervisor) upgradeBackend(req messaging.BackendUpgradeRequest) galleries = reqGalleries } - // When the master tagged this upgrade with an OpID, stream gallery download - // progress back on the per-op subject (reused from install — an upgrade is a - // force-reinstall). Old masters omit OpID and stay on the silent path. The + // When the caller tagged this upgrade with an OpID, stream gallery download + // progress back on the same sink install uses — an upgrade IS a + // force-reinstall. Callers that omit OpID stay on the silent path. The // deferred Flush guarantees a terminal-percentage event even if the upgrade - // errors out, so the master's per-node bar never hangs mid-download. + // errors out, so the caller's per-node bar never hangs mid-download. var downloadCb func(file, current, total string, percentage float64) - if req.OpID != "" && s.nats != nil { - publisher := nodes.NewDebouncedInstallProgressPublisher(s.nats, s.nodeID, req.OpID, req.Backend, installProgressDebounce) + if req.OpID != "" && onProgress != nil { + publisher := nodes.NewDebouncedInstallProgressSink(onProgress, s.nodeID, req.OpID, req.Backend, installProgressDebounce) downloadCb = publisher.OnDownload defer publisher.Flush() } @@ -243,14 +250,14 @@ func (s *backendSupervisor) upgradeBackend(req messaging.BackendUpgradeRequest) if req.URI != "" { xlog.Info("Upgrading backend from external URI", "backend", req.Backend, "uri", req.URI) if err := galleryop.InstallExternalBackend( - context.Background(), galleries, s.systemState, s.ml, downloadCb, req.URI, req.Name, req.Alias, true, s.cfg.RequireBackendIntegrity, + ctx, galleries, s.systemState, s.ml, downloadCb, req.URI, req.Name, req.Alias, true, s.cfg.RequireBackendIntegrity, ); err != nil { return stopped, fmt.Errorf("upgrading backend from external URI: %w", err) } } else { xlog.Info("Upgrading backend from gallery", "backend", req.Backend) if err := gallery.InstallBackendFromGallery( - context.Background(), galleries, s.systemState, s.ml, req.Backend, downloadCb, true, /* force */ + ctx, galleries, s.systemState, s.ml, req.Backend, downloadCb, true, /* force */ s.cfg.RequireBackendIntegrity, ); err != nil { return stopped, fmt.Errorf("upgrading backend from gallery: %w", err) diff --git a/core/services/worker/lifecycle.go b/core/services/worker/lifecycle.go index 69a238b32..824b569e0 100644 --- a/core/services/worker/lifecycle.go +++ b/core/services/worker/lifecycle.go @@ -14,157 +14,70 @@ import ( "github.com/mudler/xlog" ) -// subscribeLifecycleEvents wires every NATS subject this worker accepts to its -// per-event handler method. Each handler lives on *backendSupervisor below; -// keeping the dispatcher to a single line per subject makes adding a new -// subject a 2-line patch (one line here, one new method) instead of grafting -// onto a monolith. -func (s *backendSupervisor) subscribeLifecycleEvents() error { - if _, err := s.nats.SubscribeReply(messaging.SubjectNodeBackendInstall(s.nodeID), s.handleBackendInstall); err != nil { - return fmt.Errorf("subscribing to backend install events: %w", err) - } - if _, err := s.nats.SubscribeReply(messaging.SubjectNodeBackendUpgrade(s.nodeID), s.handleBackendUpgrade); err != nil { - return fmt.Errorf("subscribing to backend upgrade events: %w", err) - } - if _, err := s.nats.Subscribe(messaging.SubjectNodeBackendStop(s.nodeID), s.handleBackendStop); err != nil { - return fmt.Errorf("subscribing to backend stop events: %w", err) - } - if _, err := s.nats.SubscribeReply(messaging.SubjectNodeBackendDelete(s.nodeID), s.handleBackendDelete); err != nil { - return fmt.Errorf("subscribing to backend delete events: %w", err) - } - if _, err := s.nats.SubscribeReply(messaging.SubjectNodeBackendList(s.nodeID), s.handleBackendList); err != nil { - return fmt.Errorf("subscribing to backend list events: %w", err) - } - if _, err := s.nats.SubscribeReply(messaging.SubjectNodeModelsRunning(s.nodeID), s.handleModelsRunning); err != nil { - return fmt.Errorf("subscribing to models running events: %w", err) - } - if _, err := s.nats.SubscribeReply(messaging.SubjectNodeModelUnload(s.nodeID), s.handleModelUnload); err != nil { - return fmt.Errorf("subscribing to model unload events: %w", err) - } - if _, err := s.nats.SubscribeReply(messaging.SubjectNodeModelStop(s.nodeID), s.handleModelStop); err != nil { - return fmt.Errorf("subscribing to model stop events: %w", err) - } - if _, err := s.nats.SubscribeReply(messaging.SubjectNodeModelDelete(s.nodeID), s.handleModelDelete); err != nil { - return fmt.Errorf("subscribing to model delete events: %w", err) - } - if _, err := s.nats.Subscribe(messaging.SubjectNodeStop(s.nodeID), s.handleNodeStop); err != nil { - return fmt.Errorf("subscribing to node stop events: %w", err) - } - return nil -} - -func (s *backendSupervisor) handleModelStop(data []byte, reply func([]byte)) { - var req messaging.ModelStopRequest - if err := json.Unmarshal(data, &req); err != nil { - replyJSON(reply, messaging.ModelStopReply{Error: fmt.Sprintf("invalid request: %v", err)}) - return - } - replyJSON(reply, s.stopModelExact(req)) -} - -// handleBackendInstall is the NATS callback for backend.install — install -// backend (idempotent: skips download if binary exists on disk) + start gRPC -// process (request-reply). +// The worker's lifecycle verbs. // -// Each request runs in its own goroutine so that a slow install on one -// backend does NOT head-of-line-block install requests for unrelated -// backends arriving on the same subscription. Per-backend serialization -// is provided by lockBackend so two requests targeting the same on-disk -// artifact don't race the gallery directory. -func (s *backendSupervisor) handleBackendInstall(data []byte, reply func([]byte)) { - go func() { - xlog.Info("Received NATS backend.install event") - var req messaging.BackendInstallRequest - if err := json.Unmarshal(data, &req); err != nil { - resp := messaging.BackendInstallReply{Success: false, Error: fmt.Sprintf("invalid request: %v", err)} - replyJSON(reply, resp) - return - } +// Each takes a decoded request and returns a reply value. They carry no +// carrier: the HTTP control plane in control_routes.go is what decodes the +// request, calls one of these, and encodes what comes back. Keeping the verb +// free of its transport is what let the ten NATS subscriptions these replaced +// be deleted without touching a line of what they actually do. - release := s.lockBackend(req.Backend) - defer release() - - // req.Force=true is the legacy path used by pre-2026-05-08 masters - // that don't know about backend.upgrade. Honor it so a rolling - // update with new worker + old master keeps working; new masters - // send to backend.upgrade instead. - addr, err := s.installBackend(req, req.Force) - if err != nil { - xlog.Error("Failed to install backend via NATS", "error", err) - resp := messaging.BackendInstallReply{Success: false, Error: err.Error()} - replyJSON(reply, resp) - return - } - - // The address goes back exactly as the process listens on it. It used - // to be rewritten onto this worker's advertise host, which made the - // reply the worker's third advertisement site; the frontend now reads - // only the port out of it and dials nothing. - // - // The rewrite was also wrong in a way nothing caught: the worker - // records the loopback address and stopModelExact refuses a stop whose - // ExpectedAddress does not match it, so on any worker whose advertise - // host was not 127.0.0.1 every acknowledged model stop failed with an - // address mismatch. - replyJSON(reply, messaging.BackendInstallReply{Success: true, WorkerLocalAddress: addr}) - }() -} - -// handleBackendUpgrade is the NATS callback for backend.upgrade — force-reinstall -// a backend (request-reply). Lives on its own subscription so a multi-minute -// download here does NOT block the install fast-path subscription on the same -// worker. -func (s *backendSupervisor) handleBackendUpgrade(data []byte, reply func([]byte)) { - go func() { - xlog.Info("Received NATS backend.upgrade event") - var req messaging.BackendUpgradeRequest - if err := json.Unmarshal(data, &req); err != nil { - resp := messaging.BackendUpgradeReply{Success: false, Error: fmt.Sprintf("invalid request: %v", err)} - replyJSON(reply, resp) - return - } - - release := s.lockBackend(req.Backend) - defer release() - - // stopped is meaningful even on the error paths: it lists processes - // already terminated (and ports already recycled) before the failure, so - // the controller must drop those rows regardless of the outcome. - stopped, err := s.upgradeBackend(req) - if err != nil { - xlog.Error("Failed to upgrade backend via NATS", "error", err) - replyJSON(reply, messaging.BackendUpgradeReply{ - Success: false, - Error: err.Error(), - StoppedProcessKeys: stopped, - ReportsStoppedProcesses: true, - }) - return - } - replyJSON(reply, messaging.BackendUpgradeReply{ - Success: true, - StoppedProcessKeys: stopped, - ReportsStoppedProcesses: true, - }) - }() -} - -// handleBackendStop is the NATS callback for backend.stop — stop a specific -// backend process (fire-and-forget, no reply expected). -func (s *backendSupervisor) handleBackendStop(data []byte) { - req, stopAll, err := decodeBackendStopRequest(data) +// backendList answers backend.list with the backends installed in this node's +// gallery. +func (s *backendSupervisor) backendList() messaging.BackendListReply { + xlog.Info("Serving backend.list") + backends, err := gallery.ListSystemBackends(s.systemState) if err != nil { - xlog.Error("Ignoring malformed NATS backend.stop event", "error", err) - return + return messaging.BackendListReply{Error: err.Error()} } + + var infos []messaging.NodeBackendInfo + for name, b := range backends { + // Drop synthetic alias rows: ListSystemBackends emits an entry + // keyed by the alias name that re-uses the chosen concrete's + // metadata. The frontend can't reconstruct that aliasing + // faithfully from a flat NodeBackendInfo, and for upgrade + // detection it would surface as a phantom `` install + // pointing at the dev concrete's URI/digest — tricking the + // upgrade check into flagging the non-dev gallery entry of the + // same alias. Concrete and meta entries always have + // `name == b.Metadata.Name`, so this drops aliases only. + if b.Metadata != nil && b.Metadata.Name != "" && name != b.Metadata.Name { + continue + } + info := messaging.NodeBackendInfo{ + Name: name, + IsSystem: b.IsSystem, + IsMeta: b.IsMeta, + } + if b.Metadata != nil { + info.InstalledAt = b.Metadata.InstalledAt + info.GalleryURL = b.Metadata.GalleryURL + info.Version = b.Metadata.Version + info.URI = b.Metadata.URI + info.Digest = b.Metadata.Digest + } + infos = append(infos, info) + } + + return messaging.BackendListReply{Backends: infos} +} + +// stopBackends serves backend.stop: it terminates the processes the request +// names, or every process when it names none. +// +// It reports nothing. The verb has always been fire-and-forget, and a stop that +// could report per-process failure would be a different contract than the one +// the frontend was written against. +func (s *backendSupervisor) stopBackends(req messaging.BackendStopRequest, stopAll bool) { if stopAll { - xlog.Info("Received NATS backend.stop event (all)", "force", req.Force) + xlog.Info("Serving backend.stop (all)", "force", req.Force) s.stopAllBackends(req.Force) return } - xlog.Info("Received NATS backend.stop event", "backend", req.Backend, "force", req.Force) + xlog.Info("Serving backend.stop", "backend", req.Backend, "force", req.Force) // The identifier may be a backend name, a model name, or an exact - // modelID#replica key depending on the publisher; resolveStopTargets + // modelID#replica key depending on the caller; resolveStopTargets // handles all three. stopBackend alone resolves only the model meanings. for _, key := range s.resolveStopTargets(req.Backend) { if err := s.stopBackendExact(key, req.Force); err != nil { @@ -173,6 +86,9 @@ func (s *backendSupervisor) handleBackendStop(data []byte) { } } +// decodeBackendStopRequest reads a backend.stop body. An EMPTY body means "stop +// everything", which is the shape the frontend uses to drain a node, so it is +// not a decode failure. func decodeBackendStopRequest(data []byte) (messaging.BackendStopRequest, bool, error) { if len(data) == 0 { return messaging.BackendStopRequest{}, true, nil @@ -184,16 +100,10 @@ func decodeBackendStopRequest(data []byte) (messaging.BackendStopRequest, bool, return req, req.Backend == "", nil } -// handleBackendDelete is the NATS callback for backend.delete — stop the -// backend process if running, then remove its files from disk (request-reply). -func (s *backendSupervisor) handleBackendDelete(data []byte, reply func([]byte)) { - var req messaging.BackendDeleteRequest - if err := json.Unmarshal(data, &req); err != nil { - resp := messaging.BackendDeleteReply{Success: false, Error: fmt.Sprintf("invalid request: %v", err)} - replyJSON(reply, resp) - return - } - xlog.Info("Received NATS backend.delete event", "backend", req.Backend) +// deleteBackend serves backend.delete: stop the backend's processes if running, +// then remove its files from disk. +func (s *backendSupervisor) deleteBackend(req messaging.BackendDeleteRequest) messaging.BackendDeleteReply { + xlog.Info("Serving backend.delete", "backend", req.Backend) // Resolve the backend's identity (concrete name + alias) BEFORE touching // the filesystem: DeleteBackendFromSystem removes the metadata.json that @@ -235,8 +145,7 @@ func (s *backendSupervisor) handleBackendDelete(data []byte, reply func([]byte)) // "backend deleted" while the process keeps serving requests. xlog.Error("Failed to stop backend process during delete; aborting delete", "backend", req.Backend, "processKey", key, "error", err) - replyJSON(reply, deleteReply(false, fmt.Sprintf("could not stop running process %s: %v", key, err))) - return + return deleteReply(false, fmt.Sprintf("could not stop running process %s: %v", key, err)) } stopped = append(stopped, key) } @@ -244,74 +153,22 @@ func (s *backendSupervisor) handleBackendDelete(data []byte, reply func([]byte)) // Delete the backend files if err := gallery.DeleteBackendFromSystem(s.systemState, req.Backend); err != nil { xlog.Warn("Failed to delete backend files", "backend", req.Backend, "error", err) - replyJSON(reply, deleteReply(false, err.Error())) - return + return deleteReply(false, err.Error()) } // Re-register backends after deletion if err := gallery.RegisterBackends(s.systemState, s.ml); err != nil { xlog.Error("Failed to refresh registered backends after deletion", "backend", req.Backend, "error", err) - replyJSON(reply, deleteReply(false, err.Error())) - return + return deleteReply(false, err.Error()) } - replyJSON(reply, deleteReply(true, "")) + return deleteReply(true, "") } -// handleBackendList is the NATS callback for backend.list — reply with the -// installed backends from this node's gallery (request-reply). -func (s *backendSupervisor) handleBackendList(data []byte, reply func([]byte)) { - xlog.Info("Received NATS backend.list event") - backends, err := gallery.ListSystemBackends(s.systemState) - if err != nil { - resp := messaging.BackendListReply{Error: err.Error()} - replyJSON(reply, resp) - return - } - - var infos []messaging.NodeBackendInfo - for name, b := range backends { - // Drop synthetic alias rows: ListSystemBackends emits an entry - // keyed by the alias name that re-uses the chosen concrete's - // metadata. The frontend can't reconstruct that aliasing - // faithfully from a flat NodeBackendInfo, and for upgrade - // detection it would surface as a phantom `` install - // pointing at the dev concrete's URI/digest — tricking the - // upgrade check into flagging the non-dev gallery entry of the - // same alias. Concrete and meta entries always have - // `name == b.Metadata.Name`, so this drops aliases only. - if b.Metadata != nil && b.Metadata.Name != "" && name != b.Metadata.Name { - continue - } - info := messaging.NodeBackendInfo{ - Name: name, - IsSystem: b.IsSystem, - IsMeta: b.IsMeta, - } - if b.Metadata != nil { - info.InstalledAt = b.Metadata.InstalledAt - info.GalleryURL = b.Metadata.GalleryURL - info.Version = b.Metadata.Version - info.URI = b.Metadata.URI - info.Digest = b.Metadata.Digest - } - infos = append(infos, info) - } - - resp := messaging.BackendListReply{Backends: infos} - replyJSON(reply, resp) -} - -// handleModelUnload is the NATS callback for model.unload — call gRPC Free() -// to release GPU memory without killing the backend process (request-reply). -func (s *backendSupervisor) handleModelUnload(data []byte, reply func([]byte)) { - xlog.Info("Received NATS model.unload event") - var req messaging.ModelUnloadRequest - if err := json.Unmarshal(data, &req); err != nil { - resp := messaging.ModelUnloadReply{Success: false, Error: fmt.Sprintf("invalid request: %v", err)} - replyJSON(reply, resp) - return - } +// unloadModel serves model.unload: a gRPC Free() that releases GPU memory +// without killing the backend process. +func (s *backendSupervisor) unloadModel(ctx context.Context, req messaging.ModelUnloadRequest) messaging.ModelUnloadReply { + xlog.Info("Serving model.unload") // Find the backend address for this model's backend type // The request includes an Address field if the router knows which process to target @@ -328,42 +185,37 @@ func (s *backendSupervisor) handleModelUnload(data []byte, reply func([]byte)) { if targetAddr != "" { // Best-effort bounded gRPC Free(). A model.unload request must not - // occupy the NATS reply handler forever when a backend is wedged. + // occupy the handler forever when a backend is wedged. The bound is + // derived from the caller's own budget where it has one, so a caller + // that allowed less than workerBackendFreeTimeout is not made to wait + // longer than it asked for. client := grpc.NewClientWithToken(targetAddr, false, nil, false, s.cfg.RegistrationToken) - freeCtx, cancel := context.WithTimeout(context.Background(), workerBackendFreeTimeout) + freeCtx, cancel := context.WithTimeout(ctx, workerBackendFreeTimeout) if err := client.Free(freeCtx); err != nil { xlog.Warn("Free() failed during model.unload", "error", err, "addr", targetAddr) } cancel() } - resp := messaging.ModelUnloadReply{Success: true} - replyJSON(reply, resp) + return messaging.ModelUnloadReply{Success: true} } -// handleModelDelete is the NATS callback for model.delete — remove model -// files from disk (request-reply). -func (s *backendSupervisor) handleModelDelete(data []byte, reply func([]byte)) { - xlog.Info("Received NATS model.delete event") - var req messaging.ModelDeleteRequest - if err := json.Unmarshal(data, &req); err != nil { - replyJSON(reply, messaging.ModelDeleteReply{Success: false, Error: "invalid request"}) - return - } +// deleteModel serves model.delete: remove a model's staged files from disk. +func (s *backendSupervisor) deleteModel(req messaging.ModelDeleteRequest) messaging.ModelDeleteReply { + xlog.Info("Serving model.delete", "model", req.ModelName) if err := gallery.DeleteStagedModelFiles(s.cfg.ModelsPath, req.ModelName); err != nil { xlog.Warn("Failed to delete model files", "model", req.ModelName, "error", err) - replyJSON(reply, messaging.ModelDeleteReply{Success: false, Error: err.Error()}) - return + return messaging.ModelDeleteReply{Success: false, Error: err.Error()} } - replyJSON(reply, messaging.ModelDeleteReply{Success: true}) + return messaging.ModelDeleteReply{Success: true} } -// handleNodeStop is the NATS callback for node.stop — trigger the normal -// shutdown path via sigCh so deferred cleanup runs (fire-and-forget). -func (s *backendSupervisor) handleNodeStop(data []byte) { - xlog.Info("Received NATS stop event — signaling shutdown") +// signalNodeStop serves node.stop: it triggers the normal shutdown path via +// sigCh so deferred cleanup runs, rather than exiting the process here. +func (s *backendSupervisor) signalNodeStop() { + xlog.Info("Serving node.stop — signaling shutdown") select { case s.sigCh <- syscall.SIGTERM: default: diff --git a/core/services/worker/model_stop_test.go b/core/services/worker/model_stop_test.go index 345b61a30..c54aec821 100644 --- a/core/services/worker/model_stop_test.go +++ b/core/services/worker/model_stop_test.go @@ -1,13 +1,17 @@ package worker import ( + "bytes" "context" "encoding/json" "errors" "net" + "net/http" + "net/http/httptest" "sync/atomic" "github.com/mudler/LocalAI/core/services/messaging" + "github.com/mudler/LocalAI/core/services/workerctl" process "github.com/mudler/go-processmanager" . "github.com/onsi/ginkgo/v2" . "github.com/onsi/gomega" @@ -42,13 +46,27 @@ func startModelStopProcess() *process.Process { return proc } +// requestModelStop drives the model.stop verb over the HTTP control plane the +// worker actually serves, rather than calling the method directly. These specs +// are the ones that pin the acknowledged-stop contract, so routing them through +// the carrier is what keeps a routing mistake from passing them. func requestModelStop(s *backendSupervisor, req messaging.ModelStopRequest) messaging.ModelStopReply { + GinkgoHelper() data, err := json.Marshal(req) Expect(err).NotTo(HaveOccurred()) - var response []byte - s.handleModelStop(data, func(data []byte) { response = append([]byte(nil), data...) }) + + mux := http.NewServeMux() + s.RegisterControlRoutes(mux) + srv := httptest.NewServer(mux) + defer srv.Close() + + resp, err := srv.Client().Post(srv.URL+workerctl.PathModelStop, "application/json", bytes.NewReader(data)) + Expect(err).NotTo(HaveOccurred()) + defer func() { _ = resp.Body.Close() }() + Expect(resp.StatusCode).To(Equal(http.StatusOK)) + var reply messaging.ModelStopReply - Expect(json.Unmarshal(response, &reply)).To(Succeed()) + Expect(json.NewDecoder(resp.Body).Decode(&reply)).To(Succeed()) return reply } diff --git a/core/services/worker/models_running.go b/core/services/worker/models_running.go index efc4700a0..4cdb08edb 100644 --- a/core/services/worker/models_running.go +++ b/core/services/worker/models_running.go @@ -55,11 +55,3 @@ func (s *backendSupervisor) runningModels() []messaging.RunningModelInfo { } return running } - -// handleModelsRunning answers a models.running request with this worker's live -// process set. -func (s *backendSupervisor) handleModelsRunning(_ []byte, reply func([]byte)) { - running := s.runningModels() - xlog.Debug("Answering models.running", "nodeID", s.nodeID, "count", len(running)) - replyJSON(reply, messaging.ModelsRunningReply{Models: running}) -} diff --git a/core/services/worker/supervisor.go b/core/services/worker/supervisor.go index 1d1c6ebd3..3f90830f3 100644 --- a/core/services/worker/supervisor.go +++ b/core/services/worker/supervisor.go @@ -119,9 +119,15 @@ type backendSupervisor struct { systemState *system.SystemState galleries []config.Gallery nodeID string - nats messaging.MessagingClient sigCh chan<- os.Signal // send shutdown signal instead of os.Exit + // installFn and upgradeFn override the two long-running verbs. Non-nil + // only in specs: they exist so the control plane's ROUTING can be exercised + // without a gallery, a registry or a real download, which is the same + // argument tunnelServices was extracted under. See installer/upgrader. + installFn installFunc + upgradeFn upgradeFunc + mu sync.Mutex processes map[string]*backendProcess // key: backend name nextPort int // next unhanded-out port; grows within [minPort, maxPort] diff --git a/core/services/worker/worker.go b/core/services/worker/worker.go index a208eeb43..22f116654 100644 --- a/core/services/worker/worker.go +++ b/core/services/worker/worker.go @@ -5,6 +5,7 @@ import ( "context" "encoding/json" "fmt" + "net/http" "os" "os/signal" "path/filepath" @@ -18,6 +19,7 @@ import ( "github.com/mudler/LocalAI/core/services/messaging" "github.com/mudler/LocalAI/core/services/nodes" + "github.com/mudler/LocalAI/core/services/workerctl" grpc "github.com/mudler/LocalAI/pkg/grpc" "github.com/mudler/LocalAI/pkg/model" "github.com/mudler/LocalAI/pkg/sanitize" @@ -25,8 +27,8 @@ import ( "github.com/mudler/xlog" ) -// Run starts the distributed agent worker: registers with the frontend, -// subscribes to NATS lifecycle subjects, and blocks on signals. +// Run starts the distributed agent worker: registers with the frontend, opens +// its tunnel, serves its control plane on that tunnel, and blocks on signals. func Run(ctx *cliContext.Context, cfg *Config) error { xlog.Info("Starting worker", "basePort", cfg.effectiveBasePort()) @@ -189,7 +191,42 @@ func Run(ctx *cliContext.Context, cfg *Config) error { // means the worker has already registered with the frontend, so it is // mid-startup rather than broken. readiness := &nodes.WorkerReadiness{} - httpServer, err := nodes.StartFileTransferServer(httpAddr, stagingDir, cfg.ModelsPath, dataDir, cfg.RegistrationToken, config.DefaultMaxUploadSize, readiness, ml.BackendLogs()) + + // The supervisor is built BEFORE the HTTP server, not after, because the + // server is what serves its control plane. Ten NATS subscriptions used to + // be attached to it later, and could be, because the bus buffered nothing + // the worker had not subscribed to; a control route that is not mounted + // when the tunnel comes up is instead a 404 the frontend reads as a worker + // that does not implement the verb. + basePort := cfg.effectiveBasePort() + // Buffered so the node.stop verb can signal without blocking its response. + sigCh := make(chan os.Signal, 1) + signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM) + + // Set the registration token once before any backends are started. + if cfg.RegistrationToken != "" { + if err := os.Setenv(grpc.AuthTokenEnvVar, cfg.RegistrationToken); err != nil { + return fmt.Errorf("setting backend authentication token: %w", err) + } + } + + // Process supervisor — manages multiple backend gRPC processes on different ports + supervisor := &backendSupervisor{ + cfg: cfg, + ml: ml, + systemState: systemState, + galleries: galleries, + nodeID: nodeID, + sigCh: sigCh, + processes: make(map[string]*backendProcess), + portAffinity: make(map[string]portOwnership), + nextPort: basePort, + minPort: basePort, + maxPort: cfg.effectiveMaxPort(basePort), + } + + httpServer, err := startWorkerHTTPServer(httpAddr, stagingDir, cfg.ModelsPath, dataDir, + cfg.RegistrationToken, readiness, supervisor, ml.BackendLogs()) if err != nil { return fmt.Errorf("starting HTTP file transfer server: %w", err) } @@ -268,39 +305,6 @@ func Run(ctx *cliContext.Context, cfg *Config) error { } }() - // Process supervisor — manages multiple backend gRPC processes on different ports - basePort := cfg.effectiveBasePort() - // Buffered so NATS stop handler can send without blocking - sigCh := make(chan os.Signal, 1) - signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM) - - // Set the registration token once before any backends are started - if cfg.RegistrationToken != "" { - if err := os.Setenv(grpc.AuthTokenEnvVar, cfg.RegistrationToken); err != nil { - nodes.ShutdownFileTransferServer(httpServer) - return fmt.Errorf("setting backend authentication token: %w", err) - } - } - - supervisor := &backendSupervisor{ - cfg: cfg, - ml: ml, - systemState: systemState, - galleries: galleries, - nodeID: nodeID, - nats: natsClient, - sigCh: sigCh, - processes: make(map[string]*backendProcess), - portAffinity: make(map[string]portOwnership), - nextPort: basePort, - minPort: basePort, - maxPort: cfg.effectiveMaxPort(basePort), - } - if err := supervisor.subscribeLifecycleEvents(); err != nil { - nodes.ShutdownFileTransferServer(httpServer) - return fmt.Errorf("subscribing to worker lifecycle events: %w", err) - } - // Subscribe to file staging NATS subjects if S3 is configured if cfg.StorageURL != "" { if err := cfg.subscribeFileStaging(natsClient, nodeID); err != nil { @@ -309,7 +313,7 @@ func Run(ctx *cliContext.Context, cfg *Config) error { } } - xlog.Info("Worker ready, waiting for backend.install events") + xlog.Info("Worker ready, serving its control plane over the tunnel") // Exit on an OS signal or on an internal fatal condition (e.g. NATS // credentials became unrenewable), so the worker restarts and re-acquires // rather than lingering unable to serve. @@ -328,3 +332,21 @@ func Run(ctx *cliContext.Context, cfg *Config) error { nodes.ShutdownFileTransferServer(httpServer) return runErr } + +// startWorkerHTTPServer starts the worker's loopback HTTP server with sup's +// control plane mounted on it. +// +// It takes the supervisor rather than an optional route set on purpose: the +// control plane and the file routes are served by one listener behind one +// bearer check, and there is no worker that wants the second without the first. +// Making the supervisor a parameter is what stops a future edit from starting +// the server without the control plane and producing a worker that looks +// healthy while answering 404 to every command. +func startWorkerHTTPServer(addr, stagingDir, modelsDir, dataDir, token string, + readiness *nodes.WorkerReadiness, sup *backendSupervisor, logStore *model.BackendLogStore) (*http.Server, error) { + return nodes.StartFileTransferServer(addr, stagingDir, modelsDir, dataDir, token, + config.DefaultMaxUploadSize, readiness, &nodes.AuthenticatedRoutes{ + Prefix: workerctl.Prefix, + Register: sup.RegisterControlRoutes, + }, logStore) +} diff --git a/core/services/workerctl/paths.go b/core/services/workerctl/paths.go new file mode 100644 index 000000000..e76b5566d --- /dev/null +++ b/core/services/workerctl/paths.go @@ -0,0 +1,71 @@ +// Package workerctl names the HTTP control plane a worker serves to the +// frontends that manage it. +// +// It is a leaf on the standard library alone, and deliberately so: the worker +// registers these paths and the frontend calls them, so both sides must agree +// on the literals without either importing the other's package. +package workerctl + +import "encoding/json" + +// Prefix is the one path prefix the worker mounts its whole control plane +// under. Everything the frontend may command a worker to do lives below it, +// which is what lets the worker put the control plane behind a single +// authentication check instead of one per verb. +const Prefix = "/v1/control/" + +// The control verbs. Each replaces one NATS subject; the request and reply +// bodies are the messaging DTOs those subjects already carried, unchanged, so +// a worker still reachable over NATS and one reachable over the tunnel answer +// with the same bytes. +const ( + PathBackendInstall = "/v1/control/backend/install" + PathBackendUpgrade = "/v1/control/backend/upgrade" + PathBackendList = "/v1/control/backend/list" + PathBackendStop = "/v1/control/backend/stop" + PathBackendDelete = "/v1/control/backend/delete" + PathModelStop = "/v1/control/model/stop" + PathModelUnload = "/v1/control/model/unload" + PathModelDelete = "/v1/control/model/delete" + PathModelsRunning = "/v1/control/models/running" + PathNodeStop = "/v1/control/node/stop" +) + +// AllPaths returns every control verb's path. +// +// It exists so a spec can assert a property of the whole set rather than of a +// list it re-types, which would go stale the moment a verb is added. +func AllPaths() []string { + return []string{ + PathBackendInstall, + PathBackendUpgrade, + PathBackendList, + PathBackendStop, + PathBackendDelete, + PathModelStop, + PathModelUnload, + PathModelDelete, + PathModelsRunning, + PathNodeStop, + } +} + +// Envelope is one line of a streaming control response. +// +// Exactly one of the two is set. Zero or more Progress lines are followed by +// exactly ONE Reply line, and the Reply line is the last thing on the body. +// That ordering is the contract: it is what lets the frontend stop reading, and +// it is what replaces the subscribe-before-request dance the NATS carrier +// needed, since progress and reply now share one response and nothing can +// arrive before the caller is listening. +// +// Progress carrying the reply's own bytes is also why the 8000-byte +// notification cap that bounded the NATS progress subject has no analogue here: +// a line is written into the response body the caller is already reading. +type Envelope struct { + Progress json.RawMessage `json:"progress,omitempty"` + Reply json.RawMessage `json:"reply,omitempty"` +} + +// ContentTypeStream is the media type of a streaming control response. +const ContentTypeStream = "application/x-ndjson" diff --git a/core/services/workerctl/paths_test.go b/core/services/workerctl/paths_test.go new file mode 100644 index 000000000..c5685a99d --- /dev/null +++ b/core/services/workerctl/paths_test.go @@ -0,0 +1,72 @@ +package workerctl_test + +import ( + "encoding/json" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + + "github.com/mudler/LocalAI/core/services/workerctl" +) + +// The literals below are written out BY HAND and are deliberately not derived +// from the constants under test. A spec that says PathBackendInstall equals +// PathBackendInstall pins nothing: a rename moves both sides at once and stays +// green. What these paths actually are is a cross-version contract, because a +// frontend and a worker built from different commits reach each other over +// them, and a renamed path is a 404 that looks exactly like a broken tunnel. +var _ = Describe("control plane paths on the wire", func() { + DescribeTable("is the exact path both sides agree on", + func(got, want string) { Expect(got).To(Equal(want)) }, + Entry("install", workerctl.PathBackendInstall, "/v1/control/backend/install"), + Entry("upgrade", workerctl.PathBackendUpgrade, "/v1/control/backend/upgrade"), + Entry("list", workerctl.PathBackendList, "/v1/control/backend/list"), + Entry("backend stop", workerctl.PathBackendStop, "/v1/control/backend/stop"), + Entry("backend delete", workerctl.PathBackendDelete, "/v1/control/backend/delete"), + Entry("model stop", workerctl.PathModelStop, "/v1/control/model/stop"), + Entry("model unload", workerctl.PathModelUnload, "/v1/control/model/unload"), + Entry("model delete", workerctl.PathModelDelete, "/v1/control/model/delete"), + Entry("models running", workerctl.PathModelsRunning, "/v1/control/models/running"), + Entry("node stop", workerctl.PathNodeStop, "/v1/control/node/stop"), + ) + + It("names the prefix exactly, since the worker mounts its whole control plane behind it", func() { + Expect(workerctl.Prefix).To(Equal("/v1/control/")) + }) + + It("puts every path under the one prefix", func() { + for _, p := range workerctl.AllPaths() { + Expect(p).To(HavePrefix(workerctl.Prefix)) + } + }) + + It("enumerates every verb, so a new one cannot be added without the prefix check seeing it", func() { + Expect(workerctl.AllPaths()).To(HaveLen(10)) + Expect(workerctl.AllPaths()).To(ContainElement(workerctl.PathBackendInstall)) + Expect(workerctl.AllPaths()).To(ContainElement(workerctl.PathNodeStop)) + }) + + It("gives each verb a distinct path", func() { + seen := map[string]bool{} + for _, p := range workerctl.AllPaths() { + Expect(seen[p]).To(BeFalse(), "duplicate control path %q", p) + seen[p] = true + } + }) + + It("marshals an envelope with exactly one populated field", func() { + b, err := json.Marshal(workerctl.Envelope{Reply: json.RawMessage(`{"success":true}`)}) + Expect(err).NotTo(HaveOccurred()) + Expect(string(b)).To(Equal(`{"reply":{"success":true}}`)) + }) + + It("marshals a progress envelope without a reply key, which is what ends the body", func() { + b, err := json.Marshal(workerctl.Envelope{Progress: json.RawMessage(`{"percentage":50}`)}) + Expect(err).NotTo(HaveOccurred()) + Expect(string(b)).To(Equal(`{"progress":{"percentage":50}}`)) + }) + + It("names the streaming media type", func() { + Expect(workerctl.ContentTypeStream).To(Equal("application/x-ndjson")) + }) +}) diff --git a/core/services/workerctl/workerctl_suite_test.go b/core/services/workerctl/workerctl_suite_test.go new file mode 100644 index 000000000..a47f1c8ff --- /dev/null +++ b/core/services/workerctl/workerctl_suite_test.go @@ -0,0 +1,13 @@ +package workerctl_test + +import ( + "testing" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +func TestWorkerctl(t *testing.T) { + RegisterFailHandler(Fail) + RunSpecs(t, "Workerctl test suite") +} diff --git a/docs/content/features/distributed-mode.md b/docs/content/features/distributed-mode.md index 4e5351ebb..a057dbed1 100644 --- a/docs/content/features/distributed-mode.md +++ b/docs/content/features/distributed-mode.md @@ -30,7 +30,7 @@ The SmartRouter uses **idle-first** scheduling with **preemptive eviction**: 4. Fall back to idle nodes (zero models), then least-loaded nodes 5. If no node has capacity → **evict the least-recently-used model with zero in-flight requests** to free a node 6. If all models are busy → wait (with timeout) for a model to become idle, then evict -7. Send `backend.install` NATS event with backend name + model ID → worker starts a new gRPC process on a dynamic port +7. `POST /v1/control/backend/install` through the worker's tunnel with backend name + model ID → worker starts a new gRPC process on a dynamic port 8. SmartRouter calls gRPC `LoadModel` on the model-specific port, records in DB Each model gets its own gRPC backend process, so a single worker can serve multiple models simultaneously (e.g., a chat model and an embedding model). @@ -170,6 +170,34 @@ Every connection the frontend makes to a worker now goes through that worker's t | Inference, model load, health checks | gRPC to a backend process | `grpc` | | Model file staging, backend-log listing | HTTP to the worker's own server | `http` | | Live backend-log streaming | WebSocket to the same server | `http` | +| Backend install/upgrade/list/delete, model stop/unload/delete, node stop | HTTP to the worker's own server | `http` | + +#### The worker control plane + +A serve-backend worker serves the commands the frontend gives it as ordinary +HTTP routes under `/v1/control/`, on the same loopback server that already +carries file staging and backend logs, behind the same `LOCALAI_REGISTRATION_TOKEN` +bearer check. They replace the ten `nodes..*` NATS subjects a worker used to +subscribe to; the request and reply bodies are unchanged, so nothing an operator +inspects on the wire has a new shape. Agent workers still take `nodes..backend.stop` +over NATS. + +Two of the routes stream. `POST /v1/control/backend/install` and +`/v1/control/backend/upgrade` answer with `application/x-ndjson`: zero or more +`{"progress":{...}}` lines carrying the same download-progress payload the +per-op NATS subject carried, followed by exactly one `{"reply":{...}}` line, +which is always the last line on the body. An install that FAILS is still a +`200` with a reply whose `success` is `false`. That is deliberate, and it is the +same distinction the refusal table above draws: a non-2xx means the frontend +could not get the request to the worker, which nothing may act on, while the +worker's own verdict, including "there is no such backend", is evidence a reap +guard may act on. A worker that answered `500` for a failed install would put +its own verdict in the bucket reserved for a broken link. + +A control request carries the caller's deadline and nothing else: the worker +does not impose a timeout of its own on an install, and a caller that gives up +cancels the download rather than leaving the worker pulling gigabytes for a +response nobody will read. The address the frontend holds for a backend (the per-replica port a worker reports after an install) is still what identifies it, and it is still what appears in logs and errors. What it no longer is, is somewhere the frontend connects to: it travels inside the tunnel as the stream's target, and the worker decides what to do with it. @@ -341,7 +369,7 @@ A frontend replica that dies mid-load does not wedge the model: the job row carr ### NATS JWT authentication (recommended for production) -By default, NATS connections are anonymous: any client that can reach port `4222` may publish control-plane subjects such as `nodes..backend.install`. Enable JWT auth to scope workers to their own node subjects and give the frontend a dedicated service credential. +By default, NATS connections are anonymous: any client that can reach port `4222` may publish the subjects still carried on it. Serve-backend workers no longer subscribe to `nodes..backend.install` and its nine siblings - those are HTTP routes on the worker's tunnel now, see [The worker control plane](#the-worker-control-plane) - but agent workers, file staging and the frontend's own service credential still use NATS. Enable JWT auth to scope workers to their own node subjects and give the frontend a dedicated service credential. | Flag | Env Var | Description | |------|---------|-------------| @@ -425,7 +453,7 @@ during installation as well as the committed snapshot. {{% /notice %}} {{% notice warning %}} -The worker HTTP file transfer server is authenticated by `LOCALAI_REGISTRATION_TOKEN`. If the token is **empty**, the server **fails open** - anyone who can reach the port gets read/write access to the worker's models/staging/data directories (a remote model-poisoning / exfiltration vector). The worker logs a loud warning at startup in this case. Always set `LOCALAI_REGISTRATION_TOKEN` in distributed mode, and set `LOCALAI_DISTRIBUTED_REQUIRE_AUTH=true` (frontend **and** workers) to make a missing token *or* missing NATS credentials a hard startup error rather than a silent fail-open. +The worker HTTP file transfer server is authenticated by `LOCALAI_REGISTRATION_TOKEN`. If the token is **empty**, the server **fails open** - anyone who can reach the port gets read/write access to the worker's models/staging/data directories (a remote model-poisoning / exfiltration vector), **and to the `/v1/control/` routes that install, upgrade and delete backends and stop the node**. The worker logs a loud warning at startup in this case. Always set `LOCALAI_REGISTRATION_TOKEN` in distributed mode, and set `LOCALAI_DISTRIBUTED_REQUIRE_AUTH=true` (frontend **and** workers) to make a missing token *or* missing NATS credentials a hard startup error rather than a silent fail-open. By default the server binds loopback, so "anyone who can reach the port" means a process on the worker host, and no firewall rule is required. Setting `LOCALAI_HTTP_ADDR` to a routable address opts back out of that and puts the fail-open case back on the network - if you do it, firewall the port. {{% /notice %}}