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