From dd9aff58ffbe38e24d7863aac404068770b4bb83 Mon Sep 17 00:00:00 2001 From: Ettore Di Giacinto Date: Thu, 3 Sep 2026 01:14:29 +0000 Subject: [PATCH] feat(distributed): take the backend worker off NATS entirely A local-ai worker no longer opens a bus connection. connectNATS and its spec are gone; Run registers once, starts its tunnel, arms /readyz on that tunnel, and heartbeats. The worker's bus credential flags (--nats-jwt, --nats-user-seed, --nats-require-auth, the three TLS flags) and Config.NatsAuthRequired go with it. --nats-url stays, accepted and ignored, so an existing worker command line still parses. /readyz was the thing most likely to wedge a tunnel-only worker: it required a live NATS link, so a worker with no bus would have reported itself unready forever. nodes.NATSReadiness becomes nodes.TunnelReadiness over a local interface{ Connected() bool }, and worker.Tunnel gains Connected(), backed by a mutex-guarded session field the loop publishes and clears. A closed-but-not-yet-cleared session reads as disconnected: the loop waits for every in-flight stream before it clears the field, and the probe must answer not-ready through that wait. The heartbeat gate is DELETED rather than re-pointed at the tunnel. The heartbeat is the worker's own answer that its process is alive; whether the frontend can reach it is a separate fact the frontend already holds and ages against LOCALAI_WORKER_RECONNECT_GRACE. Withholding the heartbeat would report an unreachable worker as an absent one on the one path with no grace, where the health monitor marks it offline and its pending backend ops are deleted behind it. heartbeatLoop is given no view of the tunnel, so a gate cannot be added back without changing its signature. Removing the NATS credential manager from this path also removes a defect it carried: its refresh loop re-registered on a timer to renew a JWT, and Register CLEARS a node's NodeModel rows. Any backend worker running on frontend-minted credentials had its replica rows deleted roughly every 18 hours. Of core/cli/workerregistry, everything survives. The manager is still used in full by core/cli/agent_worker.go, which still needs NATS: Acquire, Provider, RefreshLoop, HasCredentials and TunnelToken are all untouched. The backend worker simply calls RegisterFullWithRetry directly now. WorkerPermissions is documented as serving agent nodes, and its non-agent branch narrowed to _INBOX.> on both sides. It is NOT deleted: NATS reads an empty allow list as no restriction, so returning nil would upgrade every JWT the frontend still mints for a backend node from its own inbox to the whole account. Agent workers keep the bus everywhere: their CLI flags, their subscriptions, the agent branch of WorkerPermissions, and the compose service with its LOCALAI_NATS_URL and depends_on: nats. Also corrected two flags the Nodes page advertised that do not exist (--distributed-nats, --distributed-db), and a log line plus several comments that still named a bus the code no longer touches. Assisted-by: Claude Opus 5 [claude-code] Signed-off-by: Ettore Di Giacinto --- core/http/react-ui/e2e/nodes-roster.spec.js | 45 ++++ core/http/react-ui/src/pages/Nodes.jsx | 15 +- core/services/nodes/worker_readiness.go | 53 ++-- core/services/nodes/worker_readiness_test.go | 38 +-- core/services/worker/addr_test.go | 19 +- core/services/worker/auth_required_test.go | 14 +- core/services/worker/config.go | 60 +++-- core/services/worker/heartbeat_test.go | 88 +++++++ core/services/worker/nats_connect.go | 33 --- core/services/worker/nats_connect_test.go | 29 --- core/services/worker/prefetch.go | 14 +- core/services/worker/supervisor.go | 6 +- core/services/worker/tunnel.go | 41 ++++ core/services/worker/tunnel_test.go | 136 +++++++++++ core/services/worker/worker.go | 241 ++++++++----------- docker-compose.distributed.yaml | 14 +- docs/content/features/distributed-mode.md | 53 ++-- docs/content/reference/cli-reference.md | 4 +- pkg/natsauth/mint_test.go | 7 +- pkg/natsauth/permissions.go | 36 ++- pkg/natsauth/permissions_coverage_test.go | 32 ++- tests/e2e/distributed/cluster/cluster.go | 6 +- tests/e2e/distributed/nats_jwt_test.go | 23 +- 23 files changed, 634 insertions(+), 373 deletions(-) create mode 100644 core/services/worker/heartbeat_test.go delete mode 100644 core/services/worker/nats_connect.go delete mode 100644 core/services/worker/nats_connect_test.go diff --git a/core/http/react-ui/e2e/nodes-roster.spec.js b/core/http/react-ui/e2e/nodes-roster.spec.js index c6396d5b3..0f229d433 100644 --- a/core/http/react-ui/e2e/nodes-roster.spec.js +++ b/core/http/react-ui/e2e/nodes-roster.spec.js @@ -62,3 +62,48 @@ test.describe('Nodes roster panels', () => { await expect(page.getByText('alpha')).toHaveCount(0) }) }) + +test.describe('Nodes join command', () => { + // The panel emits BOTH the backend and the agent join command from one + // component, so the bus flag has to differ per tab rather than be deleted. + // Backend workers connect to no NATS server; agent workers still do. + test('omits the NATS flag for a backend worker and keeps it for an agent worker', async ({ page }) => { + await mockCluster(page, []) + await page.goto('/app/nodes') + + await page.getByRole('radio', { name: /^Backend$/ }).click() + const backendCli = page.locator('.p2p-cmd pre').first() + await expect(backendCli).toContainText('local-ai worker', { timeout: 15_000 }) + await expect(backendCli).not.toContainText('--nats-url') + const backendDocker = page.locator('.p2p-cmd pre').nth(1) + await expect(backendDocker).not.toContainText('LOCALAI_NATS_URL') + + await page.getByRole('radio', { name: /^Agent$/ }).click() + const agentCli = page.locator('.p2p-cmd pre').first() + await expect(agentCli).toContainText('local-ai agent-worker', { timeout: 15_000 }) + await expect(agentCli).toContainText('--nats-url') + const agentDocker = page.locator('.p2p-cmd pre').nth(1) + await expect(agentDocker).toContainText('LOCALAI_NATS_URL') + }) + + test('does not advertise flags the CLI does not have', async ({ page }) => { + // The "How to Enable Distributed Mode" card renders ONLY on the disabled + // state, which the page enters when /api/nodes answers 503. Mocking a + // healthy cluster here would assert absence against a card that was never + // on the page. + await page.route('**/api/nodes', r => r.fulfill({ status: 503, contentType: 'application/json', body: '{}' })) + await page.route('**/api/nodes/models', r => r.fulfill({ status: 503, contentType: 'application/json', body: '{}' })) + await page.route('**/api/nodes/scheduling', r => r.fulfill({ status: 503, contentType: 'application/json', body: '{}' })) + await page.goto('/app/nodes') + + const card = page.locator('.p2p-enable') + await expect(card).toBeVisible({ timeout: 15_000 }) + // --distributed-nats and --distributed-db were never real flags; a copied + // command carrying them fails at kong before LocalAI does anything. + await expect(card).not.toContainText('--distributed-nats') + await expect(card).not.toContainText('--distributed-db') + // And the worker step no longer tells an operator to point a backend + // worker at a bus it does not dial. + await expect(card.locator('.p2p-cmd pre').nth(1)).not.toContainText('--nats-url') + }) +}) diff --git a/core/http/react-ui/src/pages/Nodes.jsx b/core/http/react-ui/src/pages/Nodes.jsx index 769371854..8725c8477 100644 --- a/core/http/react-ui/src/pages/Nodes.jsx +++ b/core/http/react-ui/src/pages/Nodes.jsx @@ -41,6 +41,13 @@ function WorkerHintCard({ addToast, activeTab, hasWorkers }) { const { selected, setSelected, option, dev, setDev } = useImageSelector('cpu') const isAgent = activeTab === 'agent' const workerCmd = isAgent ? 'agent-worker' : 'worker' + // Only the agent worker still uses the bus. A backend worker reaches this + // frontend over one outbound tunnel and connects to no NATS server, so + // emitting --nats-url on its join command would tell an operator to stand up + // infrastructure the command does not use. Both commands come from this one + // panel, which is why the flag is conditional rather than deleted. + const natsFlag = isAgent ? ' --nats-url "nats://nats:4222" \\\n' : '' + const natsEnv = isAgent ? ' -e LOCALAI_NATS_URL="nats://nats:4222" \\\n' : '' const flags = dockerFlags(option) const flagsStr = flags ? `${flags} \\\n ` : '' @@ -67,14 +74,14 @@ function WorkerHintCard({ addToast, activeTab, hasWorkers }) {

CLI

Docker

@@ -240,7 +247,7 @@ export default function Nodes() {

Start LocalAI with distributed mode

@@ -250,7 +257,7 @@ export default function Nodes() {

Register backend nodes

diff --git a/core/services/nodes/worker_readiness.go b/core/services/nodes/worker_readiness.go index 89530abc8..9cfddd709 100644 --- a/core/services/nodes/worker_readiness.go +++ b/core/services/nodes/worker_readiness.go @@ -8,7 +8,7 @@ import ( // WorkerReadiness is the gate behind a worker's /readyz probe. // // It exists because the worker's HTTP file-transfer server is started before -// the worker has connected to NATS, and must keep serving after NATS drops. +// the worker's tunnel is up, and must keep serving after that tunnel drops. // The probe is therefore installed after the fact rather than passed as a // value, and must be safe to read from HTTP handler goroutines while the // startup goroutine is still installing it. @@ -39,36 +39,43 @@ func (r *WorkerReadiness) Check() error { return (*fn)() } -// natsConn is the slice of *messaging.Client the readiness probe needs. Kept -// as a local interface so this package does not import messaging (which would -// be an import cycle) and so tests can supply a fake. -type natsConn interface { - IsConnected() bool +// tunnelConn is the slice of *worker.Tunnel the readiness probe needs. Kept as +// a local interface so this package does not import the worker package (which +// would be an import cycle) and so tests can supply a fake. +type tunnelConn interface { + Connected() bool } -// ErrNATSDisconnected is reported by NATSReadiness when the worker has lost its -// NATS connection. -var ErrNATSDisconnected = errors.New("NATS connection is down: worker cannot receive work") +// ErrTunnelDisconnected is reported by TunnelReadiness when the worker holds no +// tunnel session. +var ErrTunnelDisconnected = errors.New("worker tunnel is down: the frontend cannot reach this worker") -// NATSReadiness builds the worker's readiness probe. +// TunnelReadiness builds the worker's readiness probe. // // A worker's real health is not "a port is open" — that is precisely the // failure mode of issue #10987, where a process that serves nothing still -// answered 200. All of a worker's actual work (backend install/start/stop -// events, inference dispatch, file-staging notifications) arrives over NATS, so -// a worker with a dead NATS link is up and useless. Registration is already -// implied by the probe being reachable at all: the file-transfer server is only -// started after the worker has successfully registered with the frontend. +// answered 200. All of a worker's actual work (backend install/start/stop, +// model lifecycle, file staging, and every inference stream) arrives over its +// tunnel to the frontend, so a worker with no tunnel session is up and +// unreachable. It binds only loopback and advertises no address, so there is no +// second way in. Registration is already implied by the probe being reachable +// at all: the file-transfer server is only started after the worker has +// successfully registered with the frontend. // -// This is deliberately something the controller cannot already see. The node -// registry's status/last_heartbeat is fed by an HTTP heartbeat to the frontend, -// a completely different network path — a worker can keep heartbeating happily -// while its NATS connection is dead, and look healthy in the registry. The -// local probe closes that gap. -func NATSReadiness(conn natsConn) func() error { +// This is deliberately something the LOCAL supervisor cannot already see. The +// node registry's status/last_heartbeat is fed by an HTTP heartbeat to the +// frontend, a completely different network path, so a worker can keep +// heartbeating happily while its tunnel is dead and look healthy in the +// registry. The local probe closes that gap. +// +// It is a readiness answer and nothing more. The frontend decides whether a +// worker is GONE from the tunnel session it holds, aged against +// LOCALAI_WORKER_RECONNECT_GRACE; a 503 here is one container's own report that +// it cannot serve right now. +func TunnelReadiness(conn tunnelConn) func() error { return func() error { - if conn == nil || !conn.IsConnected() { - return ErrNATSDisconnected + if conn == nil || !conn.Connected() { + return ErrTunnelDisconnected } return nil } diff --git a/core/services/nodes/worker_readiness_test.go b/core/services/nodes/worker_readiness_test.go index 6771291ce..7b6f6fb8b 100644 --- a/core/services/nodes/worker_readiness_test.go +++ b/core/services/nodes/worker_readiness_test.go @@ -10,11 +10,11 @@ import ( . "github.com/onsi/gomega" ) -// fakeConn stands in for *messaging.Client, which cannot be constructed without -// a live NATS server. Only IsConnected() is consulted by the readiness probe. -type fakeConn struct{ connected bool } +// fakeTunnel stands in for *worker.Tunnel, which this package cannot import +// (worker imports nodes). Only Connected() is consulted by the readiness probe. +type fakeTunnel struct{ connected bool } -func (f *fakeConn) IsConnected() bool { return f.connected } +func (f fakeTunnel) Connected() bool { return f.connected } var _ = Describe("WorkerReadiness", func() { Describe("the gate itself", func() { @@ -39,16 +39,24 @@ var _ = Describe("WorkerReadiness", func() { }) }) - Describe("NATSReadiness", func() { - It("reports ready while the NATS connection is up", func() { - Expect(NATSReadiness(&fakeConn{connected: true})()).To(Succeed()) + Describe("TunnelReadiness", func() { + It("reports ready once the tunnel holds a session", func() { + Expect(TunnelReadiness(fakeTunnel{connected: true})()).To(Succeed()) }) - It("reports not-ready once the NATS connection drops", func() { + It("reports not ready while the tunnel holds no session", func() { // This is the failure mode issue #10987 is about: the process is - // up and the port is bound, but the worker can receive no work. - err := NATSReadiness(&fakeConn{connected: false})() - Expect(err).To(MatchError(ContainSubstring("NATS"))) + // up and the port is bound, but nothing can reach this worker, + // because every request the frontend makes of it arrives over the + // tunnel. + Expect(TunnelReadiness(fakeTunnel{connected: false})()).To(MatchError(ErrTunnelDisconnected)) + }) + + It("reports not ready for a nil tunnel rather than panicking", func() { + // Run installs the probe after StartTunnel, so a nil here means a + // wiring mistake. Reporting it beats taking the HTTP handler + // goroutine down with it. + Expect(TunnelReadiness(nil)()).To(MatchError(ErrTunnelDisconnected)) }) }) @@ -86,15 +94,15 @@ var _ = Describe("WorkerReadiness", func() { }) It("serves /readyz 503 once the probe reports not-ready", func() { - ready.Set(func() error { return errors.New("NATS disconnected") }) + ready.Set(func() error { return errors.New("tunnel disconnected") }) Expect(get("/readyz")).To(Equal(http.StatusServiceUnavailable)) }) It("keeps /healthz at 200 even when readiness fails", func() { // Liveness is deliberately independent of readiness: a worker whose - // NATS link is briefly down must not be killed and restarted, or a - // NATS outage turns into a restart storm across every worker. - ready.Set(func() error { return errors.New("NATS disconnected") }) + // tunnel is briefly down must not be killed and restarted, or one + // frontend restart turns into a restart storm across every worker. + ready.Set(func() error { return errors.New("tunnel disconnected") }) Expect(get("/healthz")).To(Equal(http.StatusOK)) }) }) diff --git a/core/services/worker/addr_test.go b/core/services/worker/addr_test.go index 447880653..10bdfc35b 100644 --- a/core/services/worker/addr_test.go +++ b/core/services/worker/addr_test.go @@ -93,15 +93,30 @@ var _ = Describe("Worker address resolution", func() { var _ = Describe("Worker startup validation", func() { // A Config as kong would hand it over with nothing unusual set: the tunnel - // on by its default, no auth enforcement. + // on by its default, the required frontend URL present, no auth + // enforcement. Every case below starts here and changes ONE thing, so a + // refusal it asserts is the clause it names and not an earlier one. newConfig := func() *Config { - return &Config{WorkerTunnel: true} + return &Config{WorkerTunnel: true, RegisterTo: "http://frontend:8080"} } It("accepts the default configuration", func() { Expect(newConfig().validateStartup()).To(Succeed()) }) + It("starts a backend worker with no NATS URL", func() { + // The point of this phase: a backend worker's work arrives over its + // tunnel, so a bus address is no longer part of its startup contract. + // newConfig sets none, and there is no longer a field to set. + Expect(newConfig().validateStartup()).To(Succeed()) + }) + + It("still refuses a worker with no frontend URL", func() { + cfg := newConfig() + cfg.RegisterTo = "" + Expect(cfg.validateStartup()).To(MatchError(ContainSubstring("LOCALAI_REGISTER_TO"))) + }) + It("refuses to start with the tunnel turned off", func() { // Not a warning and not a degraded mode. A worker without its tunnel // advertises nothing, binds only loopback, and has no frontend path diff --git a/core/services/worker/auth_required_test.go b/core/services/worker/auth_required_test.go index ff1deba5c..937a52928 100644 --- a/core/services/worker/auth_required_test.go +++ b/core/services/worker/auth_required_test.go @@ -5,18 +5,10 @@ import ( . "github.com/onsi/gomega" ) +// The umbrella switch used to imply NATS auth as well. A backend worker no +// longer authenticates to a bus, so registration auth is all it still implies, +// and this is the only helper left to keep honest. var _ = Describe("Worker auth-required helpers", func() { - DescribeTable("NatsAuthRequired", - func(nats, umbrella, want bool) { - cfg := &Config{NatsRequireAuth: nats, DistributedRequireAuth: umbrella} - Expect(cfg.NatsAuthRequired()).To(Equal(want)) - }, - Entry("neither", false, false, false), - Entry("granular only", true, false, true), - Entry("umbrella only", false, true, true), - Entry("both", true, true, true), - ) - DescribeTable("RegistrationAuthRequired", func(reg, umbrella, want bool) { cfg := &Config{RegistrationRequireAuth: reg, DistributedRequireAuth: umbrella} diff --git a/core/services/worker/config.go b/core/services/worker/config.go index a3bb09fa0..84d06c566 100644 --- a/core/services/worker/config.go +++ b/core/services/worker/config.go @@ -8,15 +8,11 @@ import "fmt" // which embeds Config; this package does NOT import kong and the tags are inert // here. // -// Workers are backend-agnostic — they wait for backend.install NATS events -// from the SmartRouter to install and start the required backend. -// -// NATS is required. The worker acts as a process supervisor: -// - Receives backend.install → installs backend from gallery, starts gRPC process, replies success -// - Receives backend.stop → stops the gRPC process -// - Receives stop → full shutdown (deregister + exit) -// -// Model loading (LoadModel) is always via direct gRPC — no NATS needed for that. +// Workers are backend-agnostic: they install and start whichever backend the +// frontend asks for. The worker acts as a process supervisor, and every verb +// the frontend gives it is an HTTP route on its own loopback server, served to +// the frontend through this worker's outbound tunnel (see control_routes.go and +// core/services/workerctl). A backend worker connects to no message bus. type Config struct { // Addr and ServeAddr are read for their PORT only. A worker binds nothing // on a routable interface: backend processes and the file-transfer server @@ -49,8 +45,7 @@ type Config struct { // cluster-internal path is slow (slirp/circuit-relay, CGNAT) but outbound NAT // works fine. Resolution reuses the same gallery installer the master uses, so // the on-disk /models layout is identical. Errors are non-fatal — if the gallery - // is unreachable on boot, the worker logs a warning and starts the NATS loop - // anyway; the master can still push the file on demand (existing behaviour). + // is unreachable on boot, the worker logs a warning and starts anyway; the master can still push the file on demand (existing behaviour). PrefetchModels []string `env:"LOCALAI_PREFETCH_MODELS,PREFETCH_MODELS" help:"Comma-separated gallery model IDs to download from LOCALAI_GALLERIES at worker boot (e.g. 'llama-3.2-1b-instruct,phi-3-mini-4k'). Skipped if already on disk and SHA matches." group:"server"` // HTTPAddr binds the HTTP file-transfer server. Default is loopback on @@ -62,7 +57,7 @@ type Config struct { NodeName string `env:"LOCALAI_NODE_NAME" help:"Node name for registration (defaults to hostname)" group:"registration"` RegistrationToken string `env:"LOCALAI_REGISTRATION_TOKEN" help:"Token for authenticating with the frontend" group:"registration"` RegistrationRequireAuth bool `env:"LOCALAI_REGISTRATION_REQUIRE_AUTH" default:"false" help:"Refuse to start the HTTP file-transfer server when no registration token is set (otherwise it fails open and serves read/write to models/staging/data unauthenticated)" group:"registration"` - DistributedRequireAuth bool `env:"LOCALAI_DISTRIBUTED_REQUIRE_AUTH" default:"false" help:"Umbrella switch implying both --nats-require-auth and --registration-require-auth" group:"distributed"` + DistributedRequireAuth bool `env:"LOCALAI_DISTRIBUTED_REQUIRE_AUTH" default:"false" help:"Umbrella switch implying --registration-require-auth" group:"distributed"` HeartbeatInterval string `env:"LOCALAI_HEARTBEAT_INTERVAL" default:"10s" help:"Interval between heartbeats" group:"registration"` // WorkerTunnel holds one outbound multiplexed connection to the frontend // and serves the frontend's requests over it, so the worker needs no @@ -97,14 +92,14 @@ type Config struct { // enforces it against the raw VRAM this worker reports. Empty = no cap. VRAMBudget string `env:"LOCALAI_VRAM_BUDGET" help:"Cap VRAM used for model allocation on this worker node, as a percentage (e.g. 80%) or absolute amount (e.g. 12GB)." group:"registration"` - // NATS (required) - NatsURL string `env:"LOCALAI_NATS_URL" required:"" help:"NATS server URL" group:"distributed"` - NatsJWT string `env:"LOCALAI_NATS_JWT" help:"NATS user JWT override (normally from registration nats_jwt)" group:"distributed"` - NatsUserSeed string `env:"LOCALAI_NATS_USER_SEED" help:"NATS user signing seed override (normally from registration nats_user_seed)" group:"distributed"` - NatsRequireAuth bool `env:"LOCALAI_NATS_REQUIRE_AUTH" default:"false" help:"Require NATS JWT+seed from registration or env" group:"distributed"` - NatsTLSCA string `env:"LOCALAI_NATS_TLS_CA" type:"existingfile" help:"PEM file for NATS server CA (private PKI)" group:"distributed"` - NatsTLSCert string `env:"LOCALAI_NATS_TLS_CERT" type:"existingfile" help:"Client certificate for NATS mTLS" group:"distributed"` - NatsTLSKey string `env:"LOCALAI_NATS_TLS_KEY" type:"existingfile" help:"Client private key for NATS mTLS" group:"distributed"` + // NatsURL is accepted and ignored. A backend worker no longer connects to + // NATS at all, and the credential and TLS flags that went with it are gone + // from this command. This one is kept so that an operator whose worker + // command line or unit file still carries --nats-url gets a worker that + // starts, rather than a kong parse error on an upgrade whose whole point is + // that the bus is no longer needed here. The frontend and agent workers + // still take it and still mean it. + NatsURL string `env:"LOCALAI_NATS_URL" help:"Ignored. A backend worker connects to no message bus; the frontend reaches it over its outbound tunnel. Accepted so an existing worker command line still starts." group:"distributed" hidden:""` // S3 storage for distributed file transfer StorageURL string `env:"LOCALAI_STORAGE_URL" help:"S3 endpoint URL" group:"distributed"` @@ -114,12 +109,6 @@ type Config struct { StorageSecretKey string `env:"LOCALAI_STORAGE_SECRET_KEY" help:"S3 secret key" group:"distributed"` } -// NatsAuthRequired reports whether NATS JWT credentials must be present — the -// granular flag or the umbrella (LOCALAI_DISTRIBUTED_REQUIRE_AUTH). -func (c Config) NatsAuthRequired() bool { - return c.NatsRequireAuth || c.DistributedRequireAuth -} - // RegistrationAuthRequired reports whether a registration token must be set // before the file-transfer server may start — the granular flag or the umbrella. func (c Config) RegistrationAuthRequired() bool { @@ -129,13 +118,20 @@ func (c Config) RegistrationAuthRequired() bool { // validateStartup reports a configuration this worker must refuse to boot on, // as opposed to one it can degrade under. // -// It runs before prefetch, registration and NATS, so a refusal happens while -// the worker is still invisible to the cluster. That ordering is the point of -// checking here at all: both conditions below produce a worker that would -// register, heartbeat and be scheduled onto, so discovering them later means -// discovering them as failed inferences on a node the frontend believes is -// healthy. +// It runs before prefetch and registration, so a refusal happens while the +// worker is still invisible to the cluster. That ordering is the point of +// checking here at all: these conditions produce a worker that would register, +// heartbeat and be scheduled onto, so discovering them later means discovering +// them as failed inferences on a node the frontend believes is healthy. func (c Config) validateStartup() error { + // kong marks --register-to required, so a worker started from the CLI + // cannot miss it. Checked again here because this is the fail-fast site the + // other startup refusals live at, and because RegisterTo is now load + // bearing twice over: it is where the worker registers AND the endpoint its + // tunnel dials, which is the only way anything reaches it. + if c.RegisterTo == "" { + return fmt.Errorf("no frontend URL: set LOCALAI_REGISTER_TO (or --register-to). It is where this worker registers and the endpoint its tunnel dials, and nothing can reach a worker without it") + } // The file-transfer server fails open on an empty token (see // nodes.checkBearerToken), so enforcement plus no token is a request to // serve the models directory unauthenticated. diff --git a/core/services/worker/heartbeat_test.go b/core/services/worker/heartbeat_test.go new file mode 100644 index 000000000..5153183ad --- /dev/null +++ b/core/services/worker/heartbeat_test.go @@ -0,0 +1,88 @@ +package worker + +import ( + "context" + "errors" + "time" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +var _ = Describe("Worker heartbeat loop", func() { + var ( + ctx context.Context + cancel context.CancelFunc + tick chan time.Time + sent chan struct{} + done chan struct{} + ) + + BeforeEach(func() { + ctx, cancel = context.WithCancel(context.Background()) + tick = make(chan time.Time) + sent = make(chan struct{}, 8) + done = make(chan struct{}) + }) + + AfterEach(func() { + cancel() + Eventually(done, "5s").Should(BeClosed()) + }) + + // tick delivers one tick, and fails the spec rather than parking forever if + // the loop has stopped reading. A loop that returned early would otherwise + // hang the suite instead of reporting a failure. + fire := func() { + select { + case tick <- time.Now(): + case <-time.After(5 * time.Second): + Fail("the heartbeat loop stopped reading its ticker") + } + } + + run := func(send func(context.Context) error) { + go func() { + defer close(done) + heartbeatLoop(ctx, tick, send) + }() + } + + It("posts a heartbeat on every tick", func() { + run(func(context.Context) error { + sent <- struct{}{} + return nil + }) + for i := 0; i < 3; i++ { + fire() + Eventually(sent, "5s").Should(Receive(), "tick %d produced no heartbeat", i+1) + } + }) + + // The heartbeat is the worker's own answer that its process is alive, and + // the frontend reads absence from the tunnel it holds, aged against the + // reconnect grace. A loop that gave up on a failing post would let one + // frontend restart silence a worker for the rest of its life, and the + // health monitor marks a silent node offline with no grace at all. + It("keeps posting after a heartbeat fails", func() { + run(func(context.Context) error { + sent <- struct{}{} + return errors.New("frontend unreachable") + }) + for i := 0; i < 3; i++ { + fire() + Eventually(sent, "5s").Should(Receive(), "tick %d produced no heartbeat after a failure", i+1) + } + }) + + It("stops once the shutdown context is cancelled", func() { + run(func(context.Context) error { + sent <- struct{}{} + return nil + }) + fire() + Eventually(sent, "5s").Should(Receive()) + cancel() + Eventually(done, "5s").Should(BeClosed()) + }) +}) diff --git a/core/services/worker/nats_connect.go b/core/services/worker/nats_connect.go deleted file mode 100644 index 25485701d..000000000 --- a/core/services/worker/nats_connect.go +++ /dev/null @@ -1,33 +0,0 @@ -package worker - -import ( - "fmt" - - "github.com/mudler/LocalAI/core/services/messaging" -) - -// connectNATS opens a NATS client using JWT+seed from env or registration (env wins). -func connectNATS(url, envJWT, envSeed, registerJWT, registerSeed string, requireAuth bool, tls messaging.TLSFiles) (*messaging.Client, error) { - // Env credentials take precedence, but only fall back to registration when - // the env supplied neither half — otherwise a JWT set without its seed (or - // vice-versa) would be silently completed from a different source. - jwt, seed := envJWT, envSeed - if jwt == "" && seed == "" { - jwt, seed = registerJWT, registerSeed - } - // A JWT without its paired seed (or vice-versa) is a misconfiguration: refuse - // rather than silently connecting anonymously, which would look authenticated. - if (jwt == "") != (seed == "") { - return nil, fmt.Errorf("NATS JWT and seed must be provided together (got JWT set=%t, seed set=%t)", jwt != "", seed != "") - } - var opts []messaging.Option - if jwt != "" && seed != "" { - opts = append(opts, messaging.WithUserJWT(jwt, seed)) - } else if requireAuth { - return nil, fmt.Errorf("NATS JWT+seed required: set LOCALAI_NATS_JWT/LOCALAI_NATS_USER_SEED or enable frontend minting") - } - if tls.Enabled() { - opts = append(opts, messaging.WithTLS(tls)) - } - return messaging.New(url, opts...) -} diff --git a/core/services/worker/nats_connect_test.go b/core/services/worker/nats_connect_test.go deleted file mode 100644 index 8f554de4e..000000000 --- a/core/services/worker/nats_connect_test.go +++ /dev/null @@ -1,29 +0,0 @@ -package worker - -import ( - "github.com/mudler/LocalAI/core/services/messaging" - - . "github.com/onsi/ginkgo/v2" - . "github.com/onsi/gomega" -) - -var _ = Describe("connectNATS", func() { - It("requires JWT when requireAuth is set and no credentials are provided", func() { - _, err := connectNATS("nats://127.0.0.1:4222", "", "", "", "", true, messaging.TLSFiles{}) - Expect(err).To(HaveOccurred()) - Expect(err.Error()).To(ContainSubstring("NATS JWT+seed required")) - }) - - // A JWT supplied without its paired seed (or vice-versa) is an operator - // misconfiguration. Today connectNATS silently drops the unpaired credential - // and connects anonymously, so the operator believes the link is - // authenticated when it is not. It should refuse instead. - It("rejects a JWT supplied without a seed instead of connecting anonymously", func() { - client, err := connectNATS("nats://127.0.0.1:4222", "jwt-without-seed", "", "", "", false, messaging.TLSFiles{}) - if client != nil { - client.Close() - } - Expect(err).To(HaveOccurred(), - "connectNATS should reject an unpaired JWT rather than silently connecting anonymously") - }) -}) diff --git a/core/services/worker/prefetch.go b/core/services/worker/prefetch.go index 4aec36a99..b62effd74 100644 --- a/core/services/worker/prefetch.go +++ b/core/services/worker/prefetch.go @@ -38,8 +38,8 @@ var realModelInstaller modelInstaller = func( ) error { // enforceScan=false: workers fetch from the same gallery the master already // trusts, and the master would have scanned at install time anyway. - // autoloadBackendGalleries=false: the worker installs backends on demand via - // backend.install NATS events; prefetching the backend here would race the + // autoloadBackendGalleries=false: the worker installs backends on demand when + // the frontend calls its install control route; prefetching one here would race the // supervisor's own install path and double-trigger gallery work. // requireBackendIntegrity=false: same reason — we're not installing a backend. return gallery.InstallModelFromGallery( @@ -57,13 +57,15 @@ var realModelInstaller modelInstaller = func( // prefetchModels resolves each configured gallery ID against the model gallery // and downloads the artifact into the worker's /models. It is called once at -// worker startup, BEFORE the NATS lifecycle subscription, so that the steady -// state has the file already on disk and the master never needs to stream it. +// worker startup, BEFORE the worker registers or opens its tunnel, so that the +// steady state has the file already on disk and the master never needs to +// stream it. // // Errors are intentionally non-fatal: on a fresh worker with no outbound // connectivity (or a misconfigured gallery JSON), we want the worker to still // register and serve traffic — the master will fall back to pushing files -// on-demand over NATS/HTTP, which is the pre-existing behavior. Per-model +// on-demand over the worker's file-transfer routes, which is the pre-existing +// behavior. Per-model // failures are logged at warn level and the loop continues with the next ID. // // Idempotency comes for free from pkg/downloader.URI.DownloadFileWithContext: @@ -98,7 +100,7 @@ func prefetchModels( installer = realModelInstaller } - xlog.Info("Prefetching models from gallery before entering NATS loop", "count", len(models), "models", models) + xlog.Info("Prefetching models from gallery before registering", "count", len(models), "models", models) for _, name := range models { xlog.Info("Prefetching model", "model", name) if err := installer(ctx, modelGalleries, backendGalleries, systemState, ml, name); err != nil { diff --git a/core/services/worker/supervisor.go b/core/services/worker/supervisor.go index d06b9d69f..3ccaf4033 100644 --- a/core/services/worker/supervisor.go +++ b/core/services/worker/supervisor.go @@ -167,8 +167,8 @@ type backendSupervisor struct { // reply and drops the NodeModel rows naming that address, a row still resolves // to a live listener. probeHealth verifies liveness, not identity, so a port // re-bound inside that window is dispatched to as if it were the original -// backend. The window is a NATS round-trip plus a row delete, so seconds of -// slack are ample. +// backend. The window is one control-plane round-trip plus a row delete, so +// seconds of slack are ample. // // Deliberately NOT derived from the controller's HealthCheckInterval or from // the per-model miss threshold. Tying a worker-local constant to a @@ -627,7 +627,7 @@ func (s *backendSupervisor) releaseBackendStart(key string, bp *backendProcess) // resolveProcessKeys turns a caller-supplied identifier into the set of // process map keys it refers to. PR #9583 changed s.processes to be keyed by -// `modelID#replicaIndex`, but external NATS handlers still pass the bare +// `modelID#replicaIndex`, but external callers still pass the bare // model ID — without this resolver, those lookups silently no-op'd, so // admin "Unload model" / "Delete backend" left the worker process alive. // diff --git a/core/services/worker/tunnel.go b/core/services/worker/tunnel.go index dab799cb0..fb22b3368 100644 --- a/core/services/worker/tunnel.go +++ b/core/services/worker/tunnel.go @@ -142,6 +142,39 @@ type Tunnel struct { cancel context.CancelFunc done chan struct{} closeOnce sync.Once + + // mu guards session, which is the tunnel's CURRENT session or nil between + // them. It is written by the one loop goroutine and read by whatever asks + // Connected, which on a running worker is an HTTP handler goroutine + // serving /readyz. + mu sync.Mutex + session *yamux.Session +} + +// Connected reports whether the tunnel currently holds a live session. +// +// It is false between sessions and while the first dial is still in flight. +// That is a statement about REACHABILITY and nothing else: a worker whose +// tunnel is re-homing after a frontend restart reports false here and is still +// a registered, running node. Nothing may read it as the worker being gone. +func (t *Tunnel) Connected() bool { + if t == nil { + return false + } + t.mu.Lock() + sess := t.session + t.mu.Unlock() + // A session that has been closed is still the field's value until the loop + // clears it, and the gap between those two is exactly the window a probe + // must not answer 200 in. + return sess != nil && !sess.IsClosed() +} + +// setSession publishes (or clears) the session Connected reports on. +func (t *Tunnel) setSession(sess *yamux.Session) { + t.mu.Lock() + t.session = sess + t.mu.Unlock() } // StartTunnel dials the frontend and holds the tunnel until ctx is cancelled or @@ -257,6 +290,14 @@ func (t *Tunnel) connectAndServe(ctx context.Context) error { } xlog.Info("Worker tunnel established", "node", t.nodeID, "frontend", t.endpoint) + // Published before the accept loop starts. What makes the answer correct + // once the session dies is Connected's own IsClosed check, not this clear: + // the clear runs only after every in-flight stream has finished, which is a + // wait the probe must already be answering "not ready" through. The clear + // is here so a dead session is not held for the life of the reconnect. + t.setSession(sess) + defer t.setSession(nil) + // Streams are served under a context of the SESSION's, not the loop's. A // stream goroutine parked in a local dial would otherwise outlive the // session it belongs to and hold the reconnect below behind it. diff --git a/core/services/worker/tunnel_test.go b/core/services/worker/tunnel_test.go index 6afafaa20..0930b6f72 100644 --- a/core/services/worker/tunnel_test.go +++ b/core/services/worker/tunnel_test.go @@ -21,6 +21,7 @@ import ( . "github.com/onsi/gomega" "github.com/mudler/LocalAI/core/services/cluster" + "github.com/mudler/LocalAI/core/services/nodes" ) // awaitErr runs fn on its own goroutine and reports its result on a channel. @@ -726,6 +727,141 @@ var _ = Describe("Worker tunnel client", func() { Expect(second.nodeID).To(Equal("node-1")) }) }) + + // Connected is what the worker's /readyz reports, so these specs run + // against the real WebSocket and the real yamux handshake rather than a + // flag someone sets. A double that never touches the transport cannot go + // false the way a dropped session does. + Describe("reporting whether it holds a session", func() { + It("reports connected once the frontend has accepted its dial", func() { + frontend = newFakeFrontend(false) + start(nil) + sess := session() + Expect(sess).ToNot(BeNil()) + Eventually(tunnel.Connected, "10s").Should(BeTrue()) + }) + + It("reports disconnected once the session is gone", func() { + frontend = newFakeFrontend(false) + start(func(c *TunnelConfig) { + // Park the reconnect so the spec observes the gap between + // sessions rather than racing the next dial. + c.sleep = func(ctx context.Context, _ time.Duration) error { + <-ctx.Done() + return ctx.Err() + } + }) + sess := session() + Eventually(tunnel.Connected, "10s").Should(BeTrue()) + + Expect(sess.Close()).To(Succeed()) + Eventually(tunnel.Connected, "10s").Should(BeFalse()) + }) + + It("reports disconnected before the first dial has landed", func() { + // The frontend is never started, so nothing can accept. A worker + // that answered ready here would announce itself the moment its + // process came up, which is exactly the 200-on-a-useless-port that + // the readiness probe exists to stop. + frontend = newFakeFrontend(false) + frontend.srv.Close() + start(func(c *TunnelConfig) { + c.sleep = func(ctx context.Context, _ time.Duration) error { + <-ctx.Done() + return ctx.Err() + } + }) + Consistently(tunnel.Connected, "500ms", "50ms").Should(BeFalse()) + }) + + It("reports disconnected while it still holds a session that has been closed", func() { + // The window this closes is real and is not the same as the one + // the loop's own clear closes. When a session dies, the loop waits + // for every stream already in flight before it returns and clears + // the field, and for the whole of that wait the tunnel still HOLDS + // a session that can carry nothing new. Reading only "the field is + // set" would answer ready for the length of that wait. + c1, c2 := net.Pipe() + DeferCleanup(func() { _ = c1.Close(); _ = c2.Close() }) + sess, err := yamux.Client(c1, nil, nil) + Expect(err).ToNot(HaveOccurred()) + + held := &Tunnel{} + held.setSession(sess) + Expect(held.Connected()).To(BeTrue()) + + Expect(sess.Close()).To(Succeed()) + Expect(held.Connected()).To(BeFalse()) + }) + + It("reports disconnected on a nil tunnel rather than panicking", func() { + var absent *Tunnel + Expect(absent.Connected()).To(BeFalse()) + }) + }) + + // The worker's /readyz is armed on the tunnel, and the arming is the kind + // of line whose loss has no symptom: WorkerReadiness fails open, so a + // worker that never armed it answers 200 forever with no session. These + // specs are what makes that line's absence visible. + Describe("arming the readiness gate", func() { + It("answers not ready until the frontend has accepted the dial", func() { + frontend = newFakeFrontend(false) + frontend.srv.Close() + readiness := &nodes.WorkerReadiness{} + var err error + tunnel, err = startTunnelAndArmReadiness(ctx, readiness, TunnelConfig{ + FrontendURL: frontend.srv.URL, + NodeID: "node-1", + Token: func() string { return "tunnel-secret" }, + sleep: func(ctx context.Context, _ time.Duration) error { + <-ctx.Done() + return ctx.Err() + }, + }) + Expect(err).ToNot(HaveOccurred()) + Consistently(readiness.Check, "500ms", "50ms").Should(MatchError(nodes.ErrTunnelDisconnected)) + }) + + It("answers ready once the tunnel holds a session, and not ready again once it goes", func() { + // Both halves, in one spec, on purpose. WorkerReadiness fails open + // when no probe is installed, so "ready once connected" passes just + // as well against a gate that was never armed at all. Only the + // return to ErrTunnelDisconnected tells those two apart. + frontend = newFakeFrontend(false) + readiness := &nodes.WorkerReadiness{} + var err error + tunnel, err = startTunnelAndArmReadiness(ctx, readiness, TunnelConfig{ + FrontendURL: frontend.srv.URL, + NodeID: "node-1", + Token: func() string { return "tunnel-secret" }, + sleep: func(ctx context.Context, _ time.Duration) error { + <-ctx.Done() + return ctx.Err() + }, + }) + Expect(err).ToNot(HaveOccurred()) + sess := session() + Eventually(readiness.Check, "10s").Should(Succeed()) + + Expect(sess.Close()).To(Succeed()) + Eventually(readiness.Check, "10s").Should(MatchError(nodes.ErrTunnelDisconnected)) + }) + + It("leaves the gate alone when the tunnel cannot start at all", func() { + // A configuration refusal is not a readiness answer: Run turns it + // into a fatal error, and a gate armed on a tunnel that does not + // exist would report on nothing. + readiness := &nodes.WorkerReadiness{} + t, err := startTunnelAndArmReadiness(ctx, readiness, TunnelConfig{ + FrontendURL: "http://frontend:8080", + Token: func() string { return "tunnel-secret" }, + }) + Expect(err).To(HaveOccurred()) + Expect(t).To(BeNil()) + Expect(readiness.Check()).To(Succeed()) + }) + }) }) // echoListenerOn is echoListener bound to a specific address, so a spec can put diff --git a/core/services/worker/worker.go b/core/services/worker/worker.go index a5f220eb1..374a9cbd6 100644 --- a/core/services/worker/worker.go +++ b/core/services/worker/worker.go @@ -17,13 +17,11 @@ import ( "github.com/mudler/LocalAI/core/config" "github.com/mudler/LocalAI/core/gallery" - "github.com/mudler/LocalAI/core/services/messaging" "github.com/mudler/LocalAI/core/services/nodes" "github.com/mudler/LocalAI/core/services/storage" "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" "github.com/mudler/LocalAI/pkg/system" "github.com/mudler/xlog" ) @@ -33,7 +31,7 @@ import ( func Run(ctx *cliContext.Context, cfg *Config) error { xlog.Info("Starting worker", "basePort", cfg.effectiveBasePort()) - // Fail fast, before prefetch, registration and NATS, on any configuration + // Fail fast, before prefetch and registration, on any configuration // that would produce a worker the cluster believes in and cannot use. See // validateStartup for what those are and why each is fatal rather than // degraded. @@ -65,7 +63,7 @@ func Run(ctx *cliContext.Context, cfg *Config) error { } // Prefetch gallery models over the worker's outbound internet before we - // start accepting backend.install events. Non-fatal on every failure path: + // serve backend installs. Non-fatal on every failure path: // if the gallery is unreachable, an ID is unknown, or LOCALAI_GALLERIES is // malformed, the worker still starts and the master can push files on // demand (existing fallback behaviour). Placed BEFORE registration so a @@ -87,93 +85,46 @@ func Run(ctx *cliContext.Context, cfg *Config) error { defer shutdownCancel() registrationBody := cfg.registrationBody() - natsTLS := messaging.TLSFiles{CA: cfg.NatsTLSCA, Cert: cfg.NatsTLSCert, Key: cfg.NatsTLSKey} - // Resolve how to connect to NATS. Static env credentials cannot be re-minted, - // so register once and use them directly. Otherwise the credential manager - // (re)registers to obtain credentials — waiting through admin approval — and - // refreshes them before the minted JWT expires, so the connection survives - // expiry via a transparent reconnect. - var ( - nodeID string - connectNats func() (*messaging.Client, error) - // tunnelToken reads the node's CURRENT tunnel credential. It is a - // function because the frontend rotates the credential on every - // registration, so the value a reconnect must present is not - // necessarily the one this worker started with. - tunnelToken func() string - ) - if cfg.NatsJWT != "" || cfg.NatsUserSeed != "" { - res, regErr := regClient.RegisterFullWithRetry(shutdownCtx, registrationBody, 10) - if regErr != nil { - return fmt.Errorf("failed to register with frontend: %w", regErr) - } - nodeID = res.ID - // This path registers exactly once and never again, so the credential - // it holds cannot go stale by rotation from its own side. - // - // It CAN be superseded from outside: Register upserts by NAME, so a - // second worker registering under this node's name rotates the row's - // credential, and this worker then fails every tunnel dial with 401 for - // the life of the process. It logs that once per backoff and never - // recovers on its own; a restart fixes it only until the other worker - // registers again. - // - // Still deliberately not auto-re-registered, and now for a concrete - // reason rather than a deferral. Register CLEARS this node's NodeModel - // rows, on the assumption that a re-registering worker restarted with - // nothing loaded, so re-registering on a 401 would delete a live - // worker's replica rows on every retry, and under the name collision - // that produces the 401 the two workers would take turns doing it - // forever. That is a credential failure causing model reclamation, - // which is the one outcome this whole design exists to prevent. - // - // The fix belongs to whichever comes first: a re-auth path that mints a - // tunnel credential WITHOUT the rest of registration's side effects, or - // a worker identity that is not the operator-chosen name, which is what - // would make a collision detectable instead of silent. Until then the - // 401 is loud, names both causes, and the operator acts on it. - staticTunnelToken := res.TunnelToken - tunnelToken = func() string { return staticTunnelToken } - connectNats = func() (*messaging.Client, error) { - return connectNATS(cfg.NatsURL, cfg.NatsJWT, cfg.NatsUserSeed, "", "", cfg.NatsAuthRequired(), natsTLS) - } - } else { - credMgr := workerregistry.NewNATSCredentialManager( - func(ctx context.Context) (*workerregistry.RegisterResponse, error) { - return regClient.RegisterFull(ctx, registrationBody) - }, - cfg.NatsAuthRequired(), - ) - res, regErr := credMgr.Acquire(shutdownCtx) - if regErr != nil { - return fmt.Errorf("failed to register with frontend: %w", regErr) - } - nodeID = res.ID - // The manager re-registers to refresh NATS credentials, and every - // registration rotates the tunnel credential too, so this reads the - // manager rather than capturing a value. - tunnelToken = credMgr.TunnelToken - connectNats = func() (*messaging.Client, error) { - var opts []messaging.Option - if credMgr.HasCredentials() { - opts = append(opts, messaging.WithUserJWTProvider(credMgr.Provider())) - } - if natsTLS.Enabled() { - opts = append(opts, messaging.WithTLS(natsTLS)) - } - client, cerr := messaging.New(cfg.NatsURL, opts...) - if cerr == nil && credMgr.HasCredentials() { - go func() { - if err := credMgr.RefreshLoop(shutdownCtx); err != nil { - xlog.Error("NATS credential refresh permanently failed; shutting down worker", "error", err) - shutdownCancel() - } - }() - } - return client, cerr - } + // One registration, and the tunnel credential it returns is the only + // credential a backend worker holds. There is no second acquisition path: + // the bus this worker used to also authenticate against is gone from its + // startup entirely. + // + // This path registers exactly once and never again, so the credential it + // holds cannot go stale by rotation from its own side. + // + // It CAN be superseded from outside: Register upserts by NAME, so a second + // worker registering under this node's name rotates the row's credential, + // and this worker then fails every tunnel dial with 401 for the life of the + // process. It logs that once per backoff and never recovers on its own; a + // restart fixes it only until the other worker registers again. + // + // Still deliberately not auto-re-registered, and now for a concrete reason + // rather than a deferral. Register CLEARS this node's NodeModel rows, on + // the assumption that a re-registering worker restarted with nothing + // loaded, so re-registering on a 401 would delete a live worker's replica + // rows on every retry, and under the name collision that produces the 401 + // the two workers would take turns doing it forever. That is a credential + // failure causing model reclamation, which is the one outcome this whole + // design exists to prevent. It is also why the NATS credential manager, + // whose refresh loop re-registered on a timer, is no longer on this path: + // it was wiping a live worker's rows every time it renewed a JWT. + // + // The fix belongs to whichever comes first: a re-auth path that mints a + // tunnel credential WITHOUT the rest of registration's side effects, or a + // worker identity that is not the operator-chosen name, which is what would + // make a collision detectable instead of silent. Until then the 401 is + // loud, names both causes, and the operator acts on it. + res, err := regClient.RegisterFullWithRetry(shutdownCtx, registrationBody, 10) + if err != nil { + return fmt.Errorf("failed to register with frontend: %w", err) } + nodeID := res.ID + // Read through a function because StartTunnel presents the credential at + // DIAL time, not at start time; there is one value behind it today and the + // indirection is what keeps a future rotation from needing a new dial path. + tunnelToken := func() string { return res.TunnelToken } xlog.Info("Registered with frontend", "nodeID", nodeID, "frontend", cfg.RegisterTo) heartbeatInterval, err := time.ParseDuration(cfg.HeartbeatInterval) @@ -191,9 +142,9 @@ func Run(ctx *cliContext.Context, cfg *Config) error { // today would each stay self consistent if one moved, and the symptom would // be a verb that lists files the file server does not serve. dataDir := cfg.stagingDataDir() - // The readiness gate is created here but only armed once NATS is up, below. - // Until then /readyz reports ready, which is correct: reaching this line - // means the worker has already registered with the frontend, so it is + // The readiness gate is created here but only armed once the tunnel exists, + // below. Until then /readyz reports ready, which is correct: reaching this + // line means the worker has already registered with the frontend, so it is // mid-startup rather than broken. readiness := &nodes.WorkerReadiness{} @@ -274,7 +225,7 @@ func Run(ctx *cliContext.Context, cfg *Config) error { // before this point, so there is no configuration that reaches here without // one. A guard here would be a branch nothing can take, which reads as a // supported no-tunnel mode that does not exist. - tunnel, terr := StartTunnel(shutdownCtx, TunnelConfig{ + tunnel, terr := startTunnelAndArmReadiness(shutdownCtx, readiness, TunnelConfig{ FrontendURL: cfg.RegisterTo, NodeID: nodeID, Token: tunnelToken, @@ -293,59 +244,73 @@ func Run(ctx *cliContext.Context, cfg *Config) error { } }() - // Connect to NATS - xlog.Info("Connecting to NATS", "url", sanitize.URL(cfg.NatsURL)) - natsClient, err := connectNats() - if err != nil { - nodes.ShutdownFileTransferServer(httpServer) - return fmt.Errorf("connecting to NATS: %w", err) - } - defer natsClient.Close() - - // Arm the readiness gate now that the worker can actually receive work. - // From here /readyz tracks the live NATS link, so a worker that is up but - // cut off from the bus reports 503 instead of a meaningless 200 (#10987). - readiness.Set(nodes.NATSReadiness(natsClient)) - - // Start heartbeat goroutine (after NATS is connected so IsConnected check works) - go func() { - ticker := time.NewTicker(heartbeatInterval) - defer ticker.Stop() - for { - select { - case <-shutdownCtx.Done(): - return - case <-ticker.C: - if !natsClient.IsConnected() { - xlog.Warn("Skipping heartbeat: NATS disconnected") - continue - } - body := cfg.heartbeatBody() - if err := regClient.Heartbeat(shutdownCtx, nodeID, body); err != nil { - xlog.Warn("Heartbeat failed", "error", err) - } - } - } - }() + ticker := time.NewTicker(heartbeatInterval) + defer ticker.Stop() + go heartbeatLoop(shutdownCtx, ticker.C, func(ctx context.Context) error { + return regClient.Heartbeat(ctx, nodeID, cfg.heartbeatBody()) + }) 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. - var runErr error - select { - case <-sigCh: - case <-shutdownCtx.Done(): - runErr = fmt.Errorf("worker shutting down: NATS credentials unavailable") - xlog.Error("Internal shutdown requested", "error", runErr) - } + <-sigCh xlog.Info("Shutting down worker") shutdownCancel() // stop heartbeat loop immediately regClient.GracefulDeregister(nodeID) supervisor.stopAllBackends(false) nodes.ShutdownFileTransferServer(httpServer) - return runErr + return nil +} + +// heartbeatLoop posts this worker's heartbeat on every tick until ctx ends. +// +// It is given no view of the tunnel, and that absence is the point rather than +// an omission. The heartbeat is the WORKER'S OWN ANSWER that its process is +// alive; whether the frontend can REACH this worker is a separate fact, which +// the frontend reads from the tunnel session it holds and ages against +// LOCALAI_WORKER_RECONNECT_GRACE. Withholding the heartbeat while the tunnel +// re-homes would report an unreachable worker as an absent one, on the one path +// that has no grace at all: the health monitor marks a silent node offline or +// unhealthy and its pending backend ops are deleted behind it. +// +// A failed post is likewise not a reason to stop. The frontend being briefly +// unreachable is the exact moment a worker must keep trying, and a loop that +// returned here would silence a healthy worker for the rest of its life after +// one frontend restart. +func heartbeatLoop(ctx context.Context, tick <-chan time.Time, send func(context.Context) error) { + for { + select { + case <-ctx.Done(): + return + case <-tick: + if err := send(ctx); err != nil { + xlog.Warn("Heartbeat failed", "error", err) + } + } + } +} + +// startTunnelAndArmReadiness starts the worker's tunnel and points the +// readiness gate at it. +// +// One call rather than two lines at the call site, because the gate and the +// tunnel are one fact. /readyz means "the frontend can reach me", and a live +// tunnel session is the only thing that makes that true: the worker binds +// loopback, advertises no address, and every request the frontend makes of it +// arrives as a stream inside that session. Armed as a separate statement, the +// arming is a line whose loss has no symptom - the gate fails open, so the +// worker answers 200 forever with no session, which is issue #10987 back +// again and nothing else in the process would say a word. +// +// A tunnel that fails to START leaves the gate as it found it. There is no +// worker to report on: Run turns that into a fatal error before anything else +// happens. +func startTunnelAndArmReadiness(ctx context.Context, readiness *nodes.WorkerReadiness, cfg TunnelConfig) (*Tunnel, error) { + t, err := StartTunnel(ctx, cfg) + if err != nil { + return nil, err + } + readiness.Set(nodes.TunnelReadiness(t)) + return t, nil } // startWorkerHTTPServer starts the worker's loopback HTTP server with sup's diff --git a/docker-compose.distributed.yaml b/docker-compose.distributed.yaml index ffee64bdb..df01b452c 100644 --- a/docker-compose.distributed.yaml +++ b/docker-compose.distributed.yaml @@ -111,9 +111,14 @@ services: # No HEALTHCHECK_ENDPOINT override is needed either: the image's healthcheck # detects worker mode and derives the port from LOCALAI_SERVE_ADDR below # (gRPC base port - 1 = 50050). It runs inside the container, so a loopback - # bind is enough for it. The worker's /readyz reports 503 while its NATS - # connection is down, so `unhealthy` here means the worker genuinely cannot - # receive work. + # bind is enough for it. The worker's /readyz reports 503 while it holds no + # tunnel session, so `unhealthy` here means the frontend genuinely cannot + # reach this worker. + # + # No LOCALAI_NATS_URL and no dependency on the nats service: a backend + # worker connects to no bus. Everything the frontend asks of it travels the + # tunnel this container dials out to localai:8080. The frontend and the + # agent worker below still need NATS. environment: LOCALAI_SERVE_ADDR: "0.0.0.0:50051" DEBUG: "true" @@ -121,7 +126,6 @@ services: LOCALAI_NODE_NAME: "worker-1" LOCALAI_REGISTRATION_TOKEN: "changeme" # Must match frontend token LOCALAI_HEARTBEAT_INTERVAL: "10s" - LOCALAI_NATS_URL: "nats://nats:4222" GODEBUG: "netdns=go" # See note in localai service MODELS_PATH: /models volumes: @@ -129,8 +133,6 @@ services: depends_on: localai: condition: service_started - nats: - condition: service_started # --- GPU Support (NVIDIA) --- # Uncomment the following and change the image to a CUDA variant diff --git a/docs/content/features/distributed-mode.md b/docs/content/features/distributed-mode.md index e0019b0b7..856c622c8 100644 --- a/docs/content/features/distributed-mode.md +++ b/docs/content/features/distributed-mode.md @@ -38,7 +38,7 @@ Each model gets its own gRPC backend process, so a single worker can serve multi ## Prerequisites - **PostgreSQL** (with pgvector extension recommended for RAG) - used for node registry, job store, auth, and shared state -- **NATS** server - used for agent-worker coordination and the frontend's own cross-replica events. Serve-backend workers no longer take any command over it: the backend and model lifecycle verbs and file staging are HTTP routes on the worker's tunnel. +- **NATS** server - used for agent-worker coordination and the frontend's own cross-replica events. **Serve-backend workers do not connect to it at all**: every verb they take, and file staging with it, is an HTTP route on the worker's tunnel. Set no `LOCALAI_NATS_URL` on a `local-ai worker`. The frontend and any `local-ai agent-worker` still need one. - All services must be on the same network (or reachable via configured URLs) ## Quick Start with Docker Compose @@ -306,7 +306,7 @@ Registering against an upgraded frontend **clears** a node's `address` and `http A worker on this release opens **no inbound listener on a routable interface**. Its backend gRPC processes and its HTTP file-transfer server all bind loopback, and the frontend reaches both through the tunnel. Concretely: -- **No inbound firewall rule, published port, Service or Ingress is needed for a worker.** A worker needs outbound access to the frontend URL (`LOCALAI_REGISTER_TO`) and to NATS (`LOCALAI_NATS_URL`), and nothing else. +- **No inbound firewall rule, published port, Service or Ingress is needed for a worker.** A serve-backend worker needs outbound access to the frontend URL (`LOCALAI_REGISTER_TO`), and nothing else - not even to NATS. An agent worker also needs outbound access to `LOCALAI_NATS_URL`. - **`LOCALAI_ADVERTISE_ADDR` and `LOCALAI_ADVERTISE_HTTP_ADDR` are gone.** There is nothing to advertise. Both are ignored if still set; remove them. - **`LOCALAI_ADDR` and `LOCALAI_SERVE_ADDR` are read for their port only.** The port is the base of the backend port range, and `port-1` is the HTTP file-transfer port. The host half names an interface nothing binds. - The node's `address` and `http_address` fields in `GET /api/nodes` are empty, and are cleared for nodes that reported them before the upgrade. @@ -422,7 +422,9 @@ 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 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 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. +**This section is about agent workers and the frontend.** A serve-backend worker opens no NATS connection, so none of it applies to one; its own credential is the tunnel token it gets at registration, and its control plane is authenticated by `LOCALAI_REGISTRATION_TOKEN`. + +By default, NATS connections are anonymous: any client that can reach port `4222` may publish the subjects still carried on it. Those are the agent-worker job subjects, MCP, and the frontend's own cross-replica events. `nodes..backend.install` and its nine siblings are **not** among them - they are HTTP routes on the worker's tunnel, see [The worker control plane](#the-worker-control-plane). Enable JWT auth to scope agent workers to their own subjects and give the frontend a dedicated service credential. | Flag | Env Var | Description | |------|---------|-------------| @@ -454,9 +456,9 @@ The same env vars apply to backend workers and `local-ai agent-worker`. If the s } ``` -Workers connect with that JWT and seed automatically (shown once; store securely). Override with `LOCALAI_NATS_JWT` / `LOCALAI_NATS_USER_SEED` if needed. Set `LOCALAI_NATS_REQUIRE_AUTH=true` on workers when the bus requires credentials. +Agent workers connect with that JWT and seed automatically (shown once; store securely). Override with `LOCALAI_NATS_JWT` / `LOCALAI_NATS_USER_SEED` if needed. Set `LOCALAI_NATS_REQUIRE_AUTH=true` on an agent worker when the bus requires credentials; `local-ai worker` has no such flag, because it opens no connection to require credentials for. A JWT is still minted for a serve-backend node at registration and is simply unused; it grants nothing but that connection's own reply inbox. -When `LOCALAI_NATS_REQUIRE_AUTH=true` and no static credentials are provided, a worker that registers while still **pending admin approval** keeps re-registering (with backoff) until an admin approves it and the frontend mints its JWT - it does not start unauthenticated. This retry is **bounded**: if the node is never approved (or no credentials are minted) after a large number of attempts, the worker exits non-zero so the failure is visible (a crash-looping or failed worker) rather than hanging silently. Minted worker JWTs are also **refreshed automatically** before they expire (the worker re-registers at ~75% of the JWT lifetime), so long-running workers survive past `LOCALAI_NATS_WORKER_JWT_TTL`; the NATS connection picks up the new JWT on its next reconnect. If refresh fails persistently, the worker exits (to restart and re-acquire) rather than drifting toward an expired, unrenewable JWT. Statically configured (`LOCALAI_NATS_JWT`) and service (`LOCALAI_NATS_SERVICE_JWT`) credentials are used as-is and not refreshed. +When `LOCALAI_NATS_REQUIRE_AUTH=true` and no static credentials are provided, an agent worker that registers while still **pending admin approval** keeps re-registering (with backoff) until an admin approves it and the frontend mints its JWT - it does not start unauthenticated. This retry is **bounded**: if the node is never approved (or no credentials are minted) after a large number of attempts, the worker exits non-zero so the failure is visible (a crash-looping or failed worker) rather than hanging silently. Minted worker JWTs are also **refreshed automatically** before they expire (the worker re-registers at ~75% of the JWT lifetime), so long-running workers survive past `LOCALAI_NATS_WORKER_JWT_TTL`; the NATS connection picks up the new JWT on its next reconnect. If refresh fails persistently, the worker exits (to restart and re-acquire) rather than drifting toward an expired, unrenewable JWT. Statically configured (`LOCALAI_NATS_JWT`) and service (`LOCALAI_NATS_SERVICE_JWT`) credentials are used as-is and not refreshed. Generate operator/account material with [`scripts/nats-auth-setup.sh`](https://github.com/mudler/LocalAI/blob/master/scripts/nats-auth-setup.sh) (requires [nsc](https://docs.nats.io/running-a-nats-service/configuration/securing_nats/auth_intro/nsc)). Configure the NATS server with account resolver JWTs before enabling `LOCALAI_NATS_REQUIRE_AUTH`. @@ -511,7 +513,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), **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. +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 a hard startup error rather than a silent fail-open. On the frontend and on agent workers it also makes missing NATS credentials fatal; on a serve-backend worker it means the registration token alone, since that worker uses no bus credential. 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 %}} @@ -557,10 +559,11 @@ Workers are started with the `worker` subcommand. Each worker is generic - it do ```bash local-ai worker \ --register-to http://frontend:8080 \ - --registration-token changeme \ - --nats-url nats://nats:4222 + --registration-token changeme ``` +There is no `--nats-url` here. A serve-backend worker connects to no message bus: it dials one outbound tunnel to `--register-to` and serves every request the frontend makes of it over that. The flag is still accepted and ignored, so an existing command line keeps working. + | Flag | Env Var | Default | Description | |------|---------|---------|-------------| | `--addr` | `LOCALAI_ADDR` | *(unset)* | Base port for backend gRPC processes. Only the port is used; nothing binds the host | @@ -571,16 +574,10 @@ local-ai worker \ | `--node-name` | `LOCALAI_NODE_NAME` | hostname | Human-readable node name | | `--registration-token` | `LOCALAI_REGISTRATION_TOKEN` | *(empty)* | Token to authenticate with the frontend | | `--registration-require-auth` | `LOCALAI_REGISTRATION_REQUIRE_AUTH` | `false` | Refuse to start the HTTP file-transfer server when no registration token is set (it would otherwise fail open) | -| `--distributed-require-auth` | `LOCALAI_DISTRIBUTED_REQUIRE_AUTH` | `false` | Umbrella switch implying both `--registration-require-auth` and `--nats-require-auth` | +| `--distributed-require-auth` | `LOCALAI_DISTRIBUTED_REQUIRE_AUTH` | `false` | Umbrella switch implying `--registration-require-auth` | | `--heartbeat-interval` | `LOCALAI_HEARTBEAT_INTERVAL` | `10s` | Interval between heartbeat pings | | `--worker-tunnel` | `LOCALAI_WORKER_TUNNEL` | `true` | Hold one outbound multiplexed tunnel to the frontend and serve its requests over it, so this worker needs no inbound port (see [Worker tunnels](#worker-tunnels)). Setting it to `false` is a **fatal startup error**, not a degraded mode: the frontend has no path that dials a worker's advertised address, so a worker without its tunnel is a worker nothing can reach. To run without tunnels, run the pre-tunnel release on both the worker and the frontend. | -| `--nats-url` | `LOCALAI_NATS_URL` | *(required)* | NATS URL. A serve-backend worker takes no command over it - backend installation and file staging are HTTP routes on its tunnel - but it must still connect for the frontend to consider it healthy. | -| `--nats-jwt` | `LOCALAI_NATS_JWT` | *(empty)* | Optional override for the `nats_jwt` returned at registration | -| `--nats-user-seed` | `LOCALAI_NATS_USER_SEED` | *(empty)* | Optional override for `nats_user_seed` from registration | -| `--nats-require-auth` | `LOCALAI_NATS_REQUIRE_AUTH` | `false` | Require NATS JWT+seed (from registration or env) | -| `--nats-tls-ca` | `LOCALAI_NATS_TLS_CA` | *(empty)* | PEM file for NATS server CA | -| `--nats-tls-cert` | `LOCALAI_NATS_TLS_CERT` | *(empty)* | Client certificate for NATS mTLS | -| `--nats-tls-key` | `LOCALAI_NATS_TLS_KEY` | *(empty)* | Client private key for NATS mTLS | +| `--nats-url` | `LOCALAI_NATS_URL` | *(ignored)* | **Accepted and ignored.** A serve-backend worker opens no NATS connection. Kept so an existing worker command line still starts. | | `--backends-path` | `LOCALAI_BACKENDS_PATH` | `./backends` | Path to backend binaries | | `--models-path` | `LOCALAI_MODELS_PATH` | `./models` | Path to model files | | `--vram-budget` | `LOCALAI_VRAM_BUDGET` | *(empty)* | Cap the VRAM this node advertises for model placement, as a percentage (e.g. `80%`) or an absolute amount (e.g. `12GB`). Empty uses all detected VRAM. See [Per-node VRAM budget](#per-node-vram-budget). | @@ -597,10 +594,12 @@ The worker's HTTP server (loopback, base port - 1, default 50050) exposes two un | Endpoint | Meaning | |----------|---------| -| `/healthz` | **Liveness.** 200 whenever the process is up and serving. Deliberately independent of readiness, so a brief NATS outage does not trigger a restart storm across every worker. | -| `/readyz` | **Readiness.** 200 only when the worker is registered *and* its NATS connection is live; 503 otherwise. | +| `/healthz` | **Liveness.** 200 whenever the process is up and serving. Deliberately independent of readiness, so a frontend restart that drops every tunnel does not trigger a restart storm across every worker. | +| `/readyz` | **Readiness.** 200 only when the worker is registered *and* it currently holds a tunnel session; 503 otherwise. | -`/readyz` reports something the frontend cannot see on its own. The node registry's `status` and `last_heartbeat` are driven by an HTTP heartbeat to the frontend, which is a different network path from NATS — a worker can keep heartbeating while its NATS link is dead, and so appear `healthy` in the registry while being unable to receive any work. The local probe closes that gap. +`/readyz` tracks the **tunnel**, because that is the only way anything reaches this worker: it binds loopback, advertises no address, and every request the frontend makes of it arrives as a stream inside that tunnel. It reports something the local supervisor cannot see on its own. The node registry's `status` and `last_heartbeat` are driven by an HTTP heartbeat to the frontend, a different network path - a worker can keep heartbeating while its tunnel is dead, and so appear `healthy` in the registry while being unreachable. The local probe closes that gap. + +A 503 here is **this container's own report that it cannot serve right now**, and nothing else. It is not a claim that the worker is gone; the frontend decides that from the tunnel session it holds, aged against `LOCALAI_WORKER_RECONNECT_GRACE`. The worker keeps heartbeating throughout a tunnel outage for exactly that reason: withholding the heartbeat would report an unreachable worker as an absent one, on the one path that has no grace. The container image's `HEALTHCHECK` detects worker mode and probes this endpoint automatically, deriving the port from `LOCALAI_HTTP_ADDR`, else `LOCALAI_ADDR`, else `LOCALAI_SERVE_ADDR`, minus one - the same order the worker itself uses. No `HEALTHCHECK_ENDPOINT` override is needed. Set `HEALTHCHECK_ENDPOINT` only when the bind address is passed as a CLI flag rather than an environment variable, or to pin an explicit URL. @@ -610,7 +609,6 @@ A worker needs no address configuration at all. It binds only loopback and reach ```yaml environment: - LOCALAI_NATS_URL: "nats://frontend:4222" LOCALAI_REGISTER_TO: "http://frontend:8080" LOCALAI_REGISTRATION_TOKEN: "my-secret" ``` @@ -806,7 +804,7 @@ The edit response includes these fields: - `config_revision` identifies the saved semantic configuration. - `pending_cleanup` counts old replicas that still need cleanup when the response returns. -LocalAI sends an acknowledged stop request for each exact backend process. If a worker or NATS is unreachable, LocalAI keeps the replica in the `unloading` state and retries with durable backoff. The saved edit remains successful while cleanup is pending. +LocalAI sends an acknowledged stop request for each exact backend process, over that worker's tunnel. If the worker is unreachable, LocalAI keeps the replica in the `unloading` state and retries with durable backoff. The saved edit remains successful while cleanup is pending. Workers must support the exact model-stop protocol. Upgrade all workers before you rely on revision cleanup. An older worker cannot acknowledge the request, so its stale replica remains `unloading` until cleanup succeeds or the worker re-registers. @@ -1114,13 +1112,11 @@ ds4 layer-split inference is **manual setup** in this release (Phase 1): you pla local-ai worker \ --register-to http://frontend:8080 \ --node-name worker-2 \ - --nats-url nats://nats:4222 \ --registration-token changeme local-ai worker \ --register-to http://frontend:8080 \ --node-name worker-3 \ - --nats-url nats://nats:4222 \ --registration-token changeme ``` @@ -1327,12 +1323,12 @@ Notes: |---|---|---| | **Discovery** | Automatic via libp2p token | Self-registration to frontend URL | | **State storage** | In-memory / ledger | PostgreSQL | -| **Coordination** | Gossip protocol | NATS messaging | +| **Coordination** | Gossip protocol | The worker's own tunnel for serve-backend work; NATS for agent workers and cross-replica frontend events | | **Node management** | Automatic | REST API + WebUI | | **Health monitoring** | Peer heartbeats | Centralized HealthMonitor | | **Backend management** | Manual per node | Dynamic via the worker's `backend.install` control route | | **Best for** | Ad-hoc clusters, community sharing | Production, Kubernetes, managed infrastructure | -| **Setup complexity** | Minimal (share a token) | Requires PostgreSQL + NATS | +| **Setup complexity** | Minimal (share a token) | Requires PostgreSQL on the frontend, plus NATS if you run agent workers. Serve-backend workers need neither: only an outbound route to the frontend URL. | ## Troubleshooting @@ -1342,8 +1338,9 @@ Notes: - Ensure auth is enabled on the frontend (`LOCALAI_AUTH=true`) **NATS connection errors:** +- These concern the **frontend** and **agent workers** only. A `local-ai worker` opens no NATS connection; if one is failing to join, look at its tunnel and its `--register-to` instead. - Confirm NATS is running and reachable (`nats-server --signal ldm` or check port 4222) -- Check that `--nats-url` uses the correct hostname/IP from the worker's network perspective +- Check that `--nats-url` uses the correct hostname/IP from that component's network perspective **PostgreSQL connection errors:** - Verify the connection URL format: `postgresql://user:password@host:5432/dbname?sslmode=disable` @@ -1363,7 +1360,7 @@ Notes: - Confirm that every routable replica has `state: loaded` and the same current `config_revision`. - Treat a different `effective_options_hash` as diagnostic information. Node-specific defaults can cause valid differences. - Check `cleanup_error` and `cleanup_next_retry_at` on replicas in the `unloading` state. -- Check connectivity to the worker and NATS when cleanup reports a timeout or no responder. +- Check that the worker's tunnel is up when cleanup reports a timeout or no route. - Upgrade the worker when it does not support the exact model-stop request. - Stop and restart the stale backend only as an operational recovery action. LocalAI keeps it non-routable while durable cleanup is pending. @@ -1398,7 +1395,7 @@ Notes: - The HTTP file transfer server runs on the base port - 1 (default: 50050) - All of those bind loopback, so a firewall cannot be the cause. What can is another service on the same host already holding a port in the range: move the worker's range with `LOCALAI_ADDR` (see [Worker Port Configuration](#worker-port-configuration)) or bound it with `LOCALAI_GRPC_MAX_PORT` - Verify the backend gallery configuration is correct -- The worker needs OUTBOUND network access to the gallery, to `LOCALAI_REGISTER_TO` and to `LOCALAI_NATS_URL`. It needs no inbound access at all +- The worker needs OUTBOUND network access to the gallery and to `LOCALAI_REGISTER_TO`. It needs no inbound access at all, and no access to NATS ## Roadmap: Routing and Caching Enhancements diff --git a/docs/content/reference/cli-reference.md b/docs/content/reference/cli-reference.md index fb0872545..4e16c7c96 100644 --- a/docs/content/reference/cli-reference.md +++ b/docs/content/reference/cli-reference.md @@ -209,9 +209,9 @@ LocalAI supports several subcommands beyond `run`: - `local-ai transcript` - Convert audio to text - `local-ai agent` - Run agents standalone without the full LocalAI server - `local-ai mcp-server` - Run the LocalAI admin tool surface as a stdio MCP server (controls a remote LocalAI instance over HTTP) -- `local-ai worker` - Start a worker for distributed mode (generic, backend-agnostic) +- `local-ai worker` - Start a worker for distributed mode (generic, backend-agnostic; needs only an outbound route to the frontend, no message bus) - `local-ai p2p-worker` - Run workers to distribute workload via p2p (llama.cpp-only) -- `local-ai agent-worker` - Start an agent worker for distributed mode (executes agent chats via NATS) +- `local-ai agent-worker` - Start an agent worker for distributed mode (executes agent chats via NATS, which this command still needs) - `local-ai util` - Utility commands - `local-ai explorer` - Run P2P explorer - `local-ai federated` - Run LocalAI in federated mode diff --git a/pkg/natsauth/mint_test.go b/pkg/natsauth/mint_test.go index 195f05b2c..7c849cf16 100644 --- a/pkg/natsauth/mint_test.go +++ b/pkg/natsauth/mint_test.go @@ -36,7 +36,12 @@ var _ = Describe("MintWorkerJWT", func() { uc, err := jwt.DecodeUserClaims(token) Expect(err).NotTo(HaveOccurred()) - Expect(uc.Permissions.Sub.Allow).To(ContainElement("nodes.550e8400-e29b-41d4-a716-446655440000.>")) + // A backend worker opens no bus connection at all, so its node subtree + // went with it. The JWT is still minted at registration and simply + // unused; asserting BOTH lists are exactly the inbox is what keeps it + // from silently becoming an unrestricted credential, since NATS reads + // an empty allow list as no restriction. + Expect(uc.Permissions.Sub.Allow).To(ConsistOf("_INBOX.>")) // The install-progress subject is gone with the carrier: progress is a // line in the install response now, so a minted worker JWT must not // still be granted a publish right for it. File staging went the same diff --git a/pkg/natsauth/permissions.go b/pkg/natsauth/permissions.go index 8e0fa985e..4c081c672 100644 --- a/pkg/natsauth/permissions.go +++ b/pkg/natsauth/permissions.go @@ -9,6 +9,18 @@ func workerSubjectToken(nodeID string) string { } // WorkerPermissions returns NATS pub/sub allow lists for a registered node. +// +// It serves AGENT nodes. They are the only workers left that connect to the +// bus: an agent worker subscribes to the queue subjects listed below, while a +// backend worker connects to no bus at all, because every verb a frontend gives +// it is an HTTP route on its own server reached through its outbound tunnel +// (core/services/workerctl). +// +// The non-agent branch is therefore a grant of nothing, and it has to be +// spelled that way rather than deleted. NATS reads an EMPTY allow list as no +// restriction, so a function that returned nil here would upgrade every JWT the +// frontend still mints for a backend node from "its own inbox" to "the entire +// account". The inbox is self-scoped and reaches no cluster subject. func WorkerPermissions(nodeID, nodeType string) (pubAllow, subAllow []string) { tok := workerSubjectToken(nodeID) prefix := "nodes." + tok @@ -38,25 +50,11 @@ func WorkerPermissions(nodeID, nodeType string) (pubAllow, subAllow []string) { "_INBOX.>", } default: - // Backend worker. Every verb a frontend gives it left the bus: the - // backend and model lifecycle verbs and now file staging too are HTTP - // routes under workerctl.Prefix, served on the worker's own server and - // reached through its tunnel, so no subject is minted for them and none - // is allowed here. - // - // The subscribe wildcard stays for now. A worker subscribes to nothing - // under it on this build, but narrowing it is a change a worker - // mid-upgrade would feel, and the connection itself is what the next - // step of this removal deletes. - subAllow = []string{ - prefix + ".>", - "_INBOX.>", - } - // Nothing left to publish. backend.install.*.progress went with the - // install subject, and the file-staging replies went with theirs. - pubAllow = []string{ - "_INBOX.>", - } + // Backend worker: nothing, held open at its own inbox for the reason in + // the doc comment. The node subtree it used to subscribe on went with + // the connection itself, which this worker no longer opens. + subAllow = []string{"_INBOX.>"} + pubAllow = []string{"_INBOX.>"} } return pubAllow, subAllow } diff --git a/pkg/natsauth/permissions_coverage_test.go b/pkg/natsauth/permissions_coverage_test.go index bc5cfc693..ab1b993c0 100644 --- a/pkg/natsauth/permissions_coverage_test.go +++ b/pkg/natsauth/permissions_coverage_test.go @@ -59,19 +59,25 @@ var _ = Describe("WorkerPermissions subject coverage", func() { Context("backend worker", func() { pub, sub := natsauth.WorkerPermissions(nodeID, "backend") - // A backend worker subscribes to no subject of its own on this build. - // Every verb a frontend gives it — the backend and model lifecycle ten, - // and now the four file-staging verbs — is an HTTP route on its - // tunnelled control plane, so there is no subject left to cover. See + // A backend worker opens no connection at all on this build. Every verb + // a frontend gives it, the backend and model lifecycle ten plus the + // four file-staging verbs, is an HTTP route on its tunnelled control + // plane, so there is no subject left to cover. See // core/services/workerctl. - // - // The subscribe wildcard is asserted rather than removed because the - // grant is still minted and a worker mid-upgrade still uses it. - It("still grants a backend worker its own node subtree to subscribe on", func() { - Expect(sub).To(ConsistOf( - "nodes."+workerSubjectTokenForTest(nodeID)+".>", - "_INBOX.>", - )) + It("no longer grants a backend worker its own node subtree to subscribe on", func() { + Expect(sub).ToNot(ContainElement("nodes."+workerSubjectTokenForTest(nodeID)+".>"), + "the node subtree went with the connection the worker no longer opens") + }) + + // The grant must be a grant of NOTHING and not an ABSENT grant: NATS + // treats an empty allow list as no restriction, so a branch that + // returned nil would silently widen every backend JWT the frontend + // still mints to the whole account. ConsistOf, not BeEmpty, is what + // tells those two apart. + It("grants a backend worker its own inbox and nothing else", func() { + Expect(sub).To(ConsistOf("_INBOX.>")) + Expect(sub).ToNot(BeEmpty(), + "an empty allow list is unrestricted in NATS, not restrictive") }) // The negative half, and it is the one that would catch a verb quietly @@ -80,6 +86,8 @@ var _ = Describe("WorkerPermissions subject coverage", func() { // exception and is not any more. It("grants a backend worker no publish rights outside its inbox", func() { Expect(pub).To(ConsistOf("_INBOX.>")) + Expect(pub).ToNot(BeEmpty(), + "an empty allow list is unrestricted in NATS, not restrictive") }) It("no longer grants a backend worker the file-staging publish subtree", func() { diff --git a/tests/e2e/distributed/cluster/cluster.go b/tests/e2e/distributed/cluster/cluster.go index 38012204f..c7d9ce363 100644 --- a/tests/e2e/distributed/cluster/cluster.go +++ b/tests/e2e/distributed/cluster/cluster.go @@ -332,10 +332,14 @@ func (c *Cluster) startWorker(i int) (*Process, error) { // rest of its life: the loop posts to the URL it was given at boot and // never re-resolves it (core/cli/workerregistry/client.go), so killing a // worker's registrar orphans that worker rather than failing it over. + // + // No LOCALAI_NATS_URL: a backend worker connects to no bus, and passing + // one would make every spec here prove the tunnel-only path works while + // quietly handing the worker the thing it is supposed to do without. + // The frontends above still get it. "LOCALAI_REGISTER_TO="+c.workerFrontendURL(i), "LOCALAI_NODE_NAME="+name, "LOCALAI_REGISTRATION_TOKEN="+c.opts.RegistrationToken, - "LOCALAI_NATS_URL="+c.opts.NatsURL, "DEBUG=true", ) diff --git a/tests/e2e/distributed/nats_jwt_test.go b/tests/e2e/distributed/nats_jwt_test.go index 40008c824..b6b234385 100644 --- a/tests/e2e/distributed/nats_jwt_test.go +++ b/tests/e2e/distributed/nats_jwt_test.go @@ -18,10 +18,10 @@ var _ = Describe("NATS JWT Auth", Label("Distributed", "NatsJWT"), func() { }) It("connects with a minted backend worker JWT and publishes on its one remaining allowed subject", func() { - // A backend worker's publish grant is `_INBOX.>` and nothing else now. + // A backend worker's whole grant is `_INBOX.>` now, on both sides. // Every verb a frontend gives it, file staging included, is an HTTP - // route on its tunnel, so the `nodes..files.>` grant went with the - // subjects. See pkg/natsauth.WorkerPermissions. + // route on its tunnel, and it no longer opens a bus connection at all; + // the JWT is minted and unused. See pkg/natsauth.WorkerPermissions. Expect(infra.NC.Publish("_INBOX.probe", map[string]string{"path": "/tmp/model"})).To(Succeed()) Expect(infra.NC.Conn().FlushTimeout(2 * time.Second)).To(Succeed()) Expect(infra.NC.Conn().LastError()).ToNot(HaveOccurred()) @@ -43,13 +43,20 @@ var _ = Describe("NATS JWT Auth", Label("Distributed", "NatsJWT"), func() { }, "3s", "50ms").Should(HaveOccurred()) }) - It("allows backend subscribe on the node prefix", func() { + It("denies backend subscribe on the node prefix it no longer listens to", func() { + // The node subtree was granted while a backend worker still held a + // connection with nothing under it subscribed. It does not hold one at + // all now, so the grant went too; asserting the denial is what would + // catch a subject quietly coming back to the bus. wild := nodeSubjectPrefix(infra.NodeID) + ".>" sub, err := infra.NC.Subscribe(wild, func(_ []byte) {}) - Expect(err).ToNot(HaveOccurred()) - defer func() { _ = sub.Unsubscribe() }() - Expect(infra.NC.Conn().FlushTimeout(2 * time.Second)).To(Succeed()) - Expect(infra.NC.Conn().IsConnected()).To(BeTrue()) + if err == nil { + defer func() { _ = sub.Unsubscribe() }() + Eventually(func() error { + _ = infra.NC.Conn().FlushTimeout(500 * time.Millisecond) + return infra.NC.Conn().LastError() + }, "3s", "50ms").Should(HaveOccurred()) + } }) It("rejects anonymous publish on the JWT-enabled server", func() {