diff --git a/.github/workflows/tests-e2e-distributed.yml b/.github/workflows/tests-e2e-distributed.yml index 3f952ee42..f1a1a1a60 100644 --- a/.github/workflows/tests-e2e-distributed.yml +++ b/.github/workflows/tests-e2e-distributed.yml @@ -56,11 +56,12 @@ jobs: - name: Pre-pull test images # Pulling here rather than inside the suite keeps container-start timing # out of the spec timeouts and makes a registry outage read as a - # setup failure instead of a test failure. These two are the only images - # the suite needs once the testcontainers reaper is disabled below. + # setup failure instead of a test failure. This is the only image the + # suite needs once the testcontainers reaper is disabled below: the + # suite stands up no message broker, because nothing under test dials + # one. run: | docker pull postgres:16-alpine - docker pull nats:2-alpine - name: Distributed E2E # TESTCONTAINERS_RYUK_DISABLED keeps the pre-pull above meaningful. The # reaper exists to clean up leaked containers on a long-lived host, but @@ -91,9 +92,9 @@ jobs: # # Separate job from tests-e2e-distributed so the fast in-process suite is # not held behind a Go build of local-ai. Serial on purpose: each Ginkgo - # process would get its own PostgreSQL and NATS container and each spec - # spawns two or three local-ai children, so --procs on an unmeasured runner - # is a change to make with numbers, not by default. + # process would get its own PostgreSQL container and each spec spawns two or + # three local-ai children, so --procs on an unmeasured runner is a change to + # make with numbers, not by default. # # The two timeouts bound different things and are not alternatives. Ginkgo's # --timeout=20m bounds the SUITE only; this job timeout must additionally @@ -160,7 +161,6 @@ jobs: # setup failure rather than a test failure. run: | docker pull postgres:16-alpine - docker pull nats:2-alpine - name: Cluster E2E env: # No LOCALAI_E2E_BINARY and no separate build step: make test-e2e-cluster diff --git a/README.md b/README.md index 60c788c50..890cb7f36 100644 --- a/README.md +++ b/README.md @@ -209,7 +209,7 @@ For older news and full release notes, see [GitHub Releases](https://github.com/ - [Object Detection](https://localai.io/features/object-detection/) - [Reranker API](https://localai.io/features/reranker/) - [P2P Inferencing](https://localai.io/features/distribute/) -- [Distributed Mode](https://localai.io/features/distributed-mode/) — Horizontal scaling with PostgreSQL + NATS +- [Distributed Mode](https://localai.io/features/distributed-mode/): horizontal scaling on PostgreSQL, with no message broker to run - [Model Context Protocol (MCP)](https://localai.io/docs/features/mcp/) - [Built-in Agents](https://localai.io/features/agents/) — Autonomous AI agents with tool use, RAG, skills, SSE streaming, and [Agent Hub](https://agenthub.localai.io) - [Backend Gallery](https://localai.io/backends/) — Install/remove backends on the fly via OCI images diff --git a/core/http/react-ui/e2e/nodes-roster.spec.js b/core/http/react-ui/e2e/nodes-roster.spec.js index 0f229d433..8ccd3dd02 100644 --- a/core/http/react-ui/e2e/nodes-roster.spec.js +++ b/core/http/react-ui/e2e/nodes-roster.spec.js @@ -65,9 +65,17 @@ test.describe('Nodes roster panels', () => { 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 }) => { + // component. Neither worker kind dials a message bus any more: each holds one + // outward tunnel to --register-to and takes every verb on it. A join command + // carrying --nats-url would tell an operator to stand up, secure and pay for a + // broker that nothing in the deployment connects to, which is the one way this + // migration can still cost money after the code stopped using it. + // + // Asserted on the RENDERED command text rather than on the component's + // variables, because the variables are what the fix deletes: a spec reading + // them would stop compiling instead of failing, and a compile error is not + // evidence about what an operator is shown. + test('emits no bus flag for either worker kind', async ({ page }) => { await mockCluster(page, []) await page.goto('/app/nodes') @@ -76,14 +84,21 @@ test.describe('Nodes join command', () => { 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).toContainText('LOCALAI_REGISTER_TO') await expect(backendDocker).not.toContainText('LOCALAI_NATS_URL') + // The agent tab is the one that regressed: it was the last surface still + // emitting the flag, and it kept emitting it for two tasks after the agent + // worker stopped dialling. 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') + await expect(agentCli).toContainText('--register-to', + ) + await expect(agentCli).not.toContainText('--nats-url') const agentDocker = page.locator('.p2p-cmd pre').nth(1) - await expect(agentDocker).toContainText('LOCALAI_NATS_URL') + await expect(agentDocker).toContainText('LOCALAI_REGISTER_TO') + await expect(agentDocker).not.toContainText('LOCALAI_NATS_URL') }) test('does not advertise flags the CLI does not have', async ({ page }) => { @@ -102,8 +117,12 @@ test.describe('Nodes join command', () => { // 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. + // Neither step tells an operator to point anything at a bus. The FRONTEND + // command is asserted first and by itself: it is the one that used to carry + // --nats-url as a required flag, so an operator following this card would + // have stood a broker up before starting LocalAI at all. + await expect(card.locator('.p2p-cmd pre').nth(0)).toContainText('--auth-database-url') + await expect(card.locator('.p2p-cmd pre').nth(0)).not.toContainText('--nats-url') 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 8725c8477..2e64b1736 100644 --- a/core/http/react-ui/src/pages/Nodes.jsx +++ b/core/http/react-ui/src/pages/Nodes.jsx @@ -41,13 +41,6 @@ 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 ` : '' @@ -74,14 +67,14 @@ function WorkerHintCard({ addToast, activeTab, hasWorkers }) {

CLI

Docker

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

Start LocalAI with distributed mode

diff --git a/core/http/react-ui/src/utils/format.js b/core/http/react-ui/src/utils/format.js index 67e5a374f..9756d6663 100644 --- a/core/http/react-ui/src/utils/format.js +++ b/core/http/react-ui/src/utils/format.js @@ -16,7 +16,8 @@ export function percentColor(pct) { // milliseconds, regardless of its encoding. The agent SSE bridge emits the // json_message timestamp in three different shapes depending on deploy mode: // an RFC3339 string (standalone agent pool), Unix milliseconds (local -// dispatcher), or Unix nanoseconds (older NATS path). A numeric value is +// dispatcher), or Unix nanoseconds (releases before the tunnel migration, +// which a rolling upgrade still has in flight). A numeric value is // classified by magnitude (s / ms / us / ns) so any of them yields a sane // epoch. Falls back to Date.now() for null/empty/unparseable input. export function normalizeTimestampMs(ts) { diff --git a/docker-compose.distributed.yaml b/docker-compose.distributed.yaml index df01b452c..39cc963ba 100644 --- a/docker-compose.distributed.yaml +++ b/docker-compose.distributed.yaml @@ -1,7 +1,11 @@ # Docker Compose for LocalAI Distributed Mode # -# Starts a full distributed stack: PostgreSQL, NATS, a LocalAI frontend, -# and one llama-cpp backend node. +# Starts a full distributed stack: PostgreSQL, a LocalAI frontend, one +# llama-cpp backend node and one agent worker. +# +# There is no message broker in this file and none is needed. PostgreSQL carries +# every cross-replica broadcast, and each worker dials one outbound tunnel to the +# frontend and takes every verb on it. # # Model files are transferred from the frontend to backend nodes via HTTP # — no shared volumes needed between frontend and backends. @@ -28,13 +32,6 @@ services: timeout: 3s retries: 10 - nats: - image: nats:2-alpine - ports: - - "4222:4222" # Client connections - - "8222:8222" # HTTP monitoring (optional, useful for debugging) - command: ["--js", "-m", "8222"] # Enable JetStream + monitoring - # --- LocalAI Frontend --- # Stateless API server that routes requests to backend nodes. # Add more replicas behind a load balancer for HA. @@ -52,7 +49,6 @@ services: environment: # Distributed mode LOCALAI_DISTRIBUTED: "true" - LOCALAI_NATS_URL: "nats://nats:4222" LOCALAI_AGENT_POOL_EMBEDDING_MODEL: "granite-embedding-107m-multilingual" LOCALAI_AGENT_POOL_VECTOR_ENGINE: "postgres" LOCALAI_AGENT_POOL_DATABASE_URL: "postgresql://localai:localai@postgres:5432/localai?sslmode=disable" @@ -68,7 +64,7 @@ services: # Force pure-Go DNS resolver. The default cgo resolver follows the # container's nsswitch.conf and ends up forwarding to host # systemd-resolved (127.0.0.53), which isn't reachable from inside - # the container — failing every postgres/nats hostname lookup at + # the container, failing every postgres hostname lookup at # boot. The pure-Go path reads /etc/resolv.conf directly and uses # Docker's embedded DNS at 127.0.0.11. GODEBUG: "netdns=go" @@ -83,13 +79,12 @@ services: depends_on: postgres: condition: service_healthy - nats: - condition: service_started # --- Worker Node --- # A generic worker that self-registers with the frontend. # The same LocalAI image is used — no separate image needed. - # The SmartRouter dynamically tells workers which backend to install via NATS. + # The SmartRouter tells a worker which backend to install over that worker's + # own tunnel. # # Model files are transferred from the frontend via HTTP file staging. # The worker has its own independent models volume. @@ -115,10 +110,9 @@ services: # 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. + # This worker connects to nothing but the frontend. Everything the frontend + # asks of it travels the tunnel this container dials out to localai:8080, so + # the only service it depends on is localai itself. environment: LOCALAI_SERVE_ADDR: "0.0.0.0:50051" DEBUG: "true" @@ -186,15 +180,15 @@ services: # LOCALAI_NODE_NAME really must differ: the registry upserts by name, so two # workers sharing one steal each other's row and each other's tunnel credential. # - # Workers are generic — no backend type needed. The SmartRouter - # will dynamically install the required backend via NATS when - # a model request arrives. + # Workers are generic: no backend type needed. The SmartRouter installs the + # required backend over the worker's tunnel when a model request arrives. # --- Agent Worker --- # Dedicated process for agent chat execution. - # Receives chat jobs from NATS, runs cogito LLM calls via the LocalAI API, - # and publishes results back via NATS for SSE delivery. - # No database access needed — config and skills are sent in the NATS payload. + # The frontend claims a queued run from PostgreSQL and drives it as a + # streaming control RPC over this container's own outbound tunnel; progress, + # agent events and the terminal result all come back on that same response. + # No database access needed: config and skills are sent in the request. agent-worker-1: # image: localai/localai:latest-cpu @@ -212,11 +206,10 @@ services: - | apt-get update -qq && apt-get install -y -qq docker.io >/dev/null 2>&1 exec /entrypoint.sh agent-worker - # The agent worker is NATS-only — no HTTP server to probe. The image's - # healthcheck detects that mode and reports healthy rather than probing a - # port that will never bind, so no override is needed here. + # The agent worker binds its control server on loopback only and publishes + # no port. The image's healthcheck detects that mode and reports healthy + # rather than probing a port that will never bind, so no override is needed. environment: - LOCALAI_NATS_URL: "nats://nats:4222" LOCALAI_REGISTER_TO: "http://localai:8080" LOCALAI_NODE_NAME: "agent-worker-1" LOCALAI_REGISTRATION_TOKEN: "changeme" # Must match frontend token @@ -226,8 +219,6 @@ services: depends_on: localai: condition: service_started - nats: - condition: service_started volumes: postgres_data: diff --git a/docs/content/features/distributed-mode.md b/docs/content/features/distributed-mode.md index e4368bf06..9721d3419 100644 --- a/docs/content/features/distributed-mode.md +++ b/docs/content/features/distributed-mode.md @@ -13,7 +13,7 @@ Distributed mode requires authentication enabled with a **PostgreSQL** database ## Architecture Overview -![Distributed mode architecture: a load balancer fronts stateless SmartRouter frontends backed by a shared NATS/PostgreSQL/S3 plane, with generic workers running per-model gRPC backends](/images/diagrams/distributed-mode-arch.png) +![Distributed mode architecture: a load balancer fronts stateless SmartRouter frontends backed by a shared PostgreSQL/S3 plane, with generic workers reached over the tunnel each one dials out and running per-model gRPC backends](/images/diagrams/distributed-mode-arch.png) **Frontends** are stateless LocalAI instances that receive API requests and route them to worker nodes via the **SmartRouter**. All frontends share state through PostgreSQL, which also carries every cross-replica event they broadcast. @@ -38,10 +38,10 @@ 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 - - Each frontend replica holds **one extra PostgreSQL session** beyond its connection pool, pinned for the life of the process, and creates a `bus_messages` table. Both belong to the broadcast carrier that is replacing NATS for cross-replica fan-out; it already carries the four `state.*.delta` families (see [Cross-replica in-memory state](#cross-replica-in-memory-state)). Size `max_connections` for one additional session per frontend replica. + - Each frontend replica holds **one extra PostgreSQL session** beyond its connection pool, pinned for the life of the process, and creates a `bus_messages` table. Both belong to the broadcast carrier that carries every cross-replica fan-out, including the four `state.*.delta` families (see [Cross-replica in-memory state](#cross-replica-in-memory-state)). Size `max_connections` for one additional session per frontend replica. - That session reports an `application_name` of `localai_pgbus_`, so `SELECT count(*) FROM pg_stat_activity WHERE application_name LIKE 'localai_pgbus_%'` counts the replicas currently listening. If the carrier loses its session it redials and re-registers on its own; a broadcast published while it was down is not replayed, which is why nothing that must survive a gap is carried by a broadcast alone. - `bus_messages` holds only broadcasts too large for a PostgreSQL notification, and every replica retires rows older than ten minutes. The table is a spill buffer, not a log: it is not a place to read past events from. -- **No message bus.** Nothing in a distributed deployment connects to NATS any more. Everything a frontend broadcasts travels on the PostgreSQL the deployment already runs; every verb a frontend addresses to a worker, including an agent cancel, is an HTTP route on that worker's own tunnel. `LOCALAI_NATS_URL` is accepted and ignored everywhere - on the frontend, on `local-ai worker` and on `local-ai agent-worker` - so an existing command line still starts. +- **No message broker. Do not deploy one.** Everything a frontend broadcasts travels on the PostgreSQL the deployment already runs; every verb a frontend addresses to a worker, including an agent cancel, is an HTTP route on that worker's own tunnel. If you are upgrading from a release that ran one, see [Migrating off the message broker](#migrating-off-the-message-broker): your existing command lines keep working and the broker can be shut down. - All services must be on the same network (or reachable via configured URLs) ## Quick Start with Docker Compose @@ -52,7 +52,7 @@ The easiest way to try distributed mode locally is with the provided Docker Comp docker compose -f docker-compose.distributed.yaml up ``` -This starts PostgreSQL, a LocalAI frontend, and one worker node. The compose file still stands a NATS container up; nothing connects to it and you may delete that service. When you send an inference request, the SmartRouter automatically installs the needed backend on the worker and loads the model. See the file for details on adding GPU support, shared volumes, and additional workers. +This starts PostgreSQL, a LocalAI frontend, one worker node and one agent worker. Those four services are the whole deployment: there is no broker in the file and none to add. When you send an inference request, the SmartRouter automatically installs the needed backend on the worker and loads the model. See the file for details on adding GPU support, shared volumes, and additional workers. {{% notice tip %}} Use `docker-compose.distributed.yaml` for quick local testing. For production, deploy PostgreSQL as a managed service and run frontends/workers on separate hosts. There is no message bus to deploy. @@ -66,11 +66,10 @@ The frontend is a standard LocalAI instance with distributed mode enabled. These |------|---------|---------|-------------| | `--distributed` | `LOCALAI_DISTRIBUTED` | `false` | Enable distributed mode | | `--instance-id` | `LOCALAI_INSTANCE_ID` | auto UUID | Unique instance ID for this frontend | -| `--nats-url` | `LOCALAI_NATS_URL` | *(ignored)* | **Accepted and ignored.** A frontend opens no message-bus connection. Kept so an existing command line still starts. | | `--distributed-advertise-addr` | `LOCALAI_DISTRIBUTED_ADVERTISE_ADDR` | *(derived)* | `host:port` the **other frontend replicas** dial to reach this one. See [Replica peer links](#replica-peer-links). | | `--registration-token` | `LOCALAI_REGISTRATION_TOKEN` | *(empty)* | Token that workers must provide to register | | `--registration-require-auth` | `LOCALAI_REGISTRATION_REQUIRE_AUTH` | `false` | Fail startup when distributed mode is enabled but the registration token is empty (node endpoints and worker file-transfer would otherwise be unauthenticated) | -| `--distributed-require-auth` | `LOCALAI_DISTRIBUTED_REQUIRE_AUTH` | `false` | **Umbrella switch.** Implies `--registration-require-auth`, which is what guards registration, the worker control planes and file transfer. It also implies the inert `--nats-require-auth`. Set this in production instead of the granular flags. | +| `--distributed-require-auth` | `LOCALAI_DISTRIBUTED_REQUIRE_AUTH` | `false` | **Umbrella switch.** Implies `--registration-require-auth`, which is what guards registration, the worker control planes and file transfer. Set this in production instead of the granular flags. | | `--auto-approve-nodes` | `LOCALAI_AUTO_APPROVE_NODES` | `false` | Auto-approve new worker nodes (skip admin approval) | | `--distributed-shared-models` | `LOCALAI_DISTRIBUTED_SHARED_MODELS` | `false` | Assert that every node mounts the **same** models directory at the **same** path (a shared volume). When `true`, the router skips file staging entirely and workers load models directly from the shared path instead of re-downloading them. See [Shared models directory](#shared-models-directory). | | `--distributed-disk-headroom-check` | `LOCALAI_DISTRIBUTED_DISK_HEADROOM_CHECK` | `true` | Reject worker nodes that lack free space to store the model, at scheduling time rather than partway through staging. When `false`, node selection ignores free disk; the check still runs and warns when it would have rejected every node. Also toggleable at runtime via the `distributed_disk_headroom_check` setting. See [Disk headroom](#disk-headroom). | @@ -83,6 +82,8 @@ The frontend is a standard LocalAI instance with distributed mode enabled. These | `--worker-reconnect-grace` | `LOCALAI_WORKER_RECONNECT_GRACE` | `90s` | How long a worker whose tunnel was lost is treated as **reconnecting** rather than **gone**. Only after this window may the scheduler stop placing work on that worker, clean up its rows and release its models. Set it below the worker's own reconnect worst case and you will condemn workers that are re-homing exactly as designed. Measured on the database clock, so every replica agrees. See [A lost tunnel is a departure, not an absence](#a-lost-tunnel-is-a-departure-not-an-absence). | | `--expose-node-header` | `LOCALAI_EXPOSE_NODE_HEADER` | `false` | When enabled, inference responses carry an `X-LocalAI-Node` header with the ID of the worker node that served the request. Coverage spans the OpenAI-compatible endpoints (chat completions, completions, embeddings, audio transcriptions, audio speech / TTS, image generations, image inpainting), the Jina rerank endpoint (`/v1/rerank`), the VAD endpoints (`/v1/vad`, `/vad`), and the Anthropic Messages (`/v1/messages`) and Ollama (`/api/chat`, `/api/generate`, `/api/embed`) shims. Useful for debugging, observability and load-balancer attribution. Off by default: the node ID reveals internal cluster topology and should not be exposed on a public endpoint. Best-effort: under heavy concurrency for the same model across multiple replicas, the header may reflect a recent routing decision rather than this exact request's. Acceptable for observability and debugging. | +The three `LOCALAI_NATS_*_TIMEOUT` names above are **control-RPC budgets, not broker settings**, and are still read and enforced. They carry that prefix only because they were introduced alongside the message bus that distributed mode used to require; renaming them would break every existing deployment for cosmetics. Do not delete them when you [shut the broker down](#migrating-off-the-message-broker). + ### Replica peer links Frontend replicas record themselves in an `instances` table and open direct links to each other, so that a request arriving at one replica can be served by state another replica holds. Each replica publishes one address for this, and every other replica dials it: it is the address **peers** use, which is not necessarily the address the process binds. A replica behind a Kubernetes Service, a load balancer or a NAT binds one and is reached at another. @@ -125,7 +126,7 @@ The peer link is served at `/api/cluster/peer` and authenticates with `LOCALAI_R Several features keep state in a frontend's process memory and surface it over the API: fine-tune jobs, quantization jobs, agent tasks and Open Responses metadata. A round-robin load balancer sends a follow-up request to any replica, so each of those maps is kept current on every replica by a broadcast. -**Those four families travel on PostgreSQL, not on NATS.** Each mutation is a `NOTIFY` on the database the deployment already runs, and each replica holds one `LISTEN` session for it. There is nothing to configure: the carrier uses the same database URL as `--auth-database-url` / `LOCALAI_AUTH_DATABASE_URL`. +**Those four families travel on PostgreSQL.** Each mutation is a `NOTIFY` on the database the deployment already runs, and each replica holds one `LISTEN` session for it. There is nothing to configure: the carrier uses the same database URL as `--auth-database-url` / `LOCALAI_AUTH_DATABASE_URL`. | Map | Subject | |-----|---------| @@ -141,7 +142,7 @@ for the life of a request: which gallery operations are in flight and how far along they are, which admin operations have been admitted, which model files are being staged onto a worker, and which replica already holds the KV/prefix cache for a prompt. Each of those is kept current on every replica by a broadcast, and -**every one of them is on PostgreSQL**. Nothing in this table uses NATS. +**every one of them is on PostgreSQL**. | Family | Subject | What a peer does with it | |--------|---------|--------------------------| @@ -478,7 +479,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 serve-backend worker needs outbound access to the frontend URL (`LOCALAI_REGISTER_TO`), and nothing else - not even to NATS. An agent worker binds only loopback too, and needs outbound access to `LOCALAI_REGISTER_TO` and nothing else either: registration, heartbeats, its tunnel and every verb the frontend addresses to it all go there. +- **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. An agent worker binds only loopback too, and needs outbound access to `LOCALAI_REGISTER_TO` and nothing else either: registration, heartbeats, its tunnel and every verb the frontend addresses to it all go there. - **`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. @@ -596,22 +597,26 @@ The chat UI renders this state inline and retries automatically once the model r A frontend replica that dies mid-load does not wedge the model: the job row carries a heartbeat and another replica reclaims a job whose heartbeat has stopped. The heartbeat is time-based, not byte-based, because a checkpoint load legitimately transfers zero bytes for many minutes. {{% /notice %}} -### NATS credentials (inert) +### Migrating off the message broker -**No LocalAI component connects to NATS.** The frontend's cross-replica fan-out is on PostgreSQL, a serve-backend worker takes every verb on its own tunnel, and an agent worker now does too, including the cancel that was the last family on a bus. +Earlier releases of distributed mode required a NATS cluster alongside PostgreSQL. **They no longer do. Shut the broker down.** Nothing in LocalAI opens a connection to one: the frontend's cross-replica fan-out is on PostgreSQL, queued work is a claim on a PostgreSQL table, a serve-backend worker takes every verb on its own tunnel, and an agent worker does too, including the cancel that was the last family on a bus. -Every `LOCALAI_NATS_*` setting is therefore accepted and inert, so an existing command line, unit file or Helm values file starts unchanged: +There is no migration step and no cutover window. Stop the broker, delete its service from your compose file, chart or unit files, and delete the credentials you generated for it. A deployment that keeps running one is paying for infrastructure that carries nothing. + +**Your existing command lines still start.** Every `LOCALAI_NATS_*` setting below is parsed and then ignored, so an unedited command line, unit file or Helm values file needs no change on the day you upgrade. Remove them at your convenience. | Flag | Env Var | Status | |------|---------|--------| -| `--nats-url` | `LOCALAI_NATS_URL` | Accepted and ignored on the frontend, `local-ai worker` and `local-ai agent-worker`. | +| `--nats-url` | `LOCALAI_NATS_URL` | Accepted and ignored on the frontend, `local-ai worker` and `local-ai agent-worker`. The value is never dialled, so it may point at a broker that is already gone. | | `--nats-account-seed` | `LOCALAI_NATS_ACCOUNT_SEED` | The frontend still mints a per-node user JWT at registration (`nats_jwt` in the register response). Nothing consumes it. | | `--nats-service-jwt` / `--nats-service-seed` | `LOCALAI_NATS_SERVICE_JWT` / `LOCALAI_NATS_SERVICE_SEED` | Accepted, unused: the frontend opens no bus connection to present them on. | | `--nats-worker-jwt-ttl` | `LOCALAI_NATS_WORKER_JWT_TTL` | Lifetime of the minted-but-unused worker JWTs. | -| `--nats-require-auth` | `LOCALAI_NATS_REQUIRE_AUTH` | On an agent worker this still makes registration **wait through admin approval** rather than starting against a pending node. It no longer gates any bus connection. | +| `--nats-require-auth` | `LOCALAI_NATS_REQUIRE_AUTH` | On an agent worker this still makes registration **wait through admin approval** rather than starting against a pending node. It gates no connection. | | `--nats-tls-ca` / `--nats-tls-cert` / `--nats-tls-key` | `LOCALAI_NATS_TLS_*` | Accepted, unused. | -You may stop running a NATS server, and remove these settings at your convenience. +{{% notice warning %}} +`LOCALAI_NATS_BACKEND_INSTALL_TIMEOUT`, `LOCALAI_NATS_BACKEND_UPGRADE_TIMEOUT` and `LOCALAI_NATS_MODEL_LOAD_TIMEOUT` are **not** in the table above and must **not** be removed. Despite their names they were never broker settings: each one is a control-RPC budget the frontend applies to a worker, and each is still read and still enforced. They are documented with the other frontend flags in [Frontend Configuration](#frontend-configuration). The names are kept because renaming them would break every existing deployment for cosmetics. +{{% /notice %}} {{% notice note %}} `LOCALAI_AUTH` (HTTP users/sessions) is unrelated. HTTP registration still uses `LOCALAI_REGISTRATION_TOKEN`, and every worker control plane sits behind that same bearer check. @@ -655,7 +660,7 @@ absolute snapshot path and skip transfer. Otherwise, the controller stages the complete snapshot tree to each worker before loading the backend. With an object store configured the controller uploads to the bucket and commands the worker to fetch over its tunnel; without one it pushes the files to the worker's HTTP file -transfer server directly. Neither path uses NATS. +transfer server directly. {{% notice warning %}} Every controller and worker must have enough disk space for its own snapshot @@ -713,7 +718,7 @@ local-ai worker \ --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. +There is no broker flag here. A serve-backend worker dials one outbound tunnel to `--register-to` and serves every request the frontend makes of it over that. A `--nats-url` left over from an older command line is still accepted and ignored; see [Migrating off the message broker](#migrating-off-the-message-broker). | Flag | Env Var | Default | Description | |------|---------|---------|-------------| @@ -728,7 +733,6 @@ There is no `--nats-url` here. A serve-backend worker connects to no message bus | `--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` | *(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). | @@ -1208,7 +1212,7 @@ engine_args: The ds4 backend (DeepSeek V4 Flash) supports **layer-parallel** distributed inference: a single model that is too large for one machine is split by transformer layer across several machines. Each machine must have the GGUF present locally, but loads **only its own slice** of the layers. This lets you run a model whose weights exceed any single host's memory. -This is **not** routed through the SmartRouter: it is a model-internal split, configured manually (Phase 1). It is unrelated to the NATS/PostgreSQL distributed mode described above. +This is **not** routed through the SmartRouter: it is a model-internal split, configured manually (Phase 1). It is unrelated to the PostgreSQL-backed distributed mode described above. ### Topology @@ -1506,8 +1510,8 @@ Notes: - Check that `--registration-token` matches on both frontend and worker - Ensure auth is enabled on the frontend (`LOCALAI_AUTH=true`) -**NATS connection errors:** -- Nothing in LocalAI connects to NATS any more, on any component. If a release you are running still logs one, it predates the tunnel migration; on this release, look at the failing component's tunnel and its `--register-to` instead. +**Message-broker connection errors:** +- Nothing in LocalAI connects to a broker any more, on any component. A release that logs such an error predates the tunnel migration; on this release, look at the failing component's tunnel and its `--register-to` instead. See [Migrating off the message broker](#migrating-off-the-message-broker). **PostgreSQL connection errors:** - Verify the connection URL format: `postgresql://user:password@host:5432/dbname?sslmode=disable` @@ -1562,7 +1566,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 and to `LOCALAI_REGISTER_TO`. It needs no inbound access at all, and no access to NATS +- The worker needs OUTBOUND network access to the gallery and to `LOCALAI_REGISTER_TO`. It needs no inbound access at all ## Roadmap: Routing and Caching Enhancements diff --git a/docs/content/features/distributed_inferencing.md b/docs/content/features/distributed_inferencing.md index f30b97a60..c14c01072 100644 --- a/docs/content/features/distributed_inferencing.md +++ b/docs/content/features/distributed_inferencing.md @@ -9,7 +9,7 @@ aliases = ["/features/distribution/"] ![Federated vs worker mode: federated routes a whole request to one node; worker shards one model across nodes](/images/diagrams/federated-vs-worker.png) {{% notice tip %}} -Looking for production-grade horizontal scaling with PostgreSQL and NATS? See [Distributed Mode]({{% relref "features/distributed-mode" %}}). +Looking for production-grade horizontal scaling backed by PostgreSQL alone? See [Distributed Mode]({{% relref "features/distributed-mode" %}}). {{% /notice %}} ## Choosing a distributed mode @@ -19,7 +19,7 @@ LocalAI can spread inference across multiple machines in three ways. Pick the on | Mode | Best for | Guide | |------|----------|-------| | **P2P / Federated inference** | Ad-hoc clusters, community sharing, quick experimentation. Nodes discover each other via a shared libp2p token, with no central server. | This page | -| **Distributed Mode (PostgreSQL + NATS)** | Production deployments, Kubernetes, and managed infrastructure. Stateless frontends behind a load balancer, workers self-register, and state lives in PostgreSQL. | [Distributed Mode]({{% relref "features/distributed-mode" %}}) | +| **Distributed Mode (PostgreSQL)** | Production deployments, Kubernetes, and managed infrastructure. Stateless frontends behind a load balancer, workers self-register over an outbound tunnel, and state lives in PostgreSQL. No message broker is needed. | [Distributed Mode]({{% relref "features/distributed-mode" %}}) | | **MLX distributed** | Apple Silicon clusters running MLX models over the MLX distributed runtime. | [MLX Distributed]({{% relref "features/mlx-distributed" %}}) | For the low-level protocol and endpoints used by P2P workers, see the [P2P API reference]({{% relref "reference/p2p-api" %}}). diff --git a/docs/content/features/text-generation.md b/docs/content/features/text-generation.md index bcb5f0194..aa003acb2 100644 --- a/docs/content/features/text-generation.md +++ b/docs/content/features/text-generation.md @@ -338,8 +338,9 @@ replicas, so retrieval, `previous_response_id` chaining and cancellation work regardless of which replica the load balancer picks: - `GET /v1/responses/{id}` returns the response from any replica. -- `POST /v1/responses/{id}/cancel` is delegated over NATS to the replica that is - actually generating, so generation really stops. If that replica is gone, the +- `POST /v1/responses/{id}/cancel` is delegated over the PostgreSQL broadcast + carrier to the replica that is actually generating, so generation really + stops. If that replica is gone, the response is reported as `cancelled` without blocking. - **Streaming resume (`?stream=true`) is served only by the replica that created the response.** The event buffer lives in that process's memory and is not diff --git a/docs/static/images/diagrams/distributed-mode-arch.html b/docs/static/images/diagrams/distributed-mode-arch.html index f1e49b708..39758350d 100644 --- a/docs/static/images/diagrams/distributed-mode-arch.html +++ b/docs/static/images/diagrams/distributed-mode-arch.html @@ -54,7 +54,7 @@
@@ -99,15 +99,17 @@ fY.forEach((y,i)=>{ // ---------- STATE PLANE ---------- txt(560,30,"SHARED STATE PLANE",{w:700,sz:13,ls:".2em",fill:SOFT}); -const SPX=560, SPW=300, SPY=46, SPH=470; +const SPX=560, SPW=300, SPY=46, SPH=326; shadowRect(SPX,SPY,SPW,SPH,PAPER,INK,4); svg.appendChild(el("rect",{x:SPX,y:SPY,width:SPW,height:58,fill:RUST})); svg.appendChild(el("line",{x1:SPX,y1:SPY+58,x2:SPX+SPW,y2:SPY+58,stroke:INK,"stroke-width":4})); txt(SPX+22,SPY+38,"Control plane",{f:"Bricolage Grotesque",w:800,sz:26,fill:PAPER}); // chips +// No broker chip. Every cross-replica broadcast and every queued job is a row +// or a NOTIFY on the database below, and no worker connects to this plane at +// all: a worker is reached over the tunnel it dials out to a frontend. const chips=[ - {n:"PostgreSQL", s:"shared config & state"}, - {n:"NATS", s:"jobs · messaging bus"}, + {n:"PostgreSQL", s:"state · registry · broadcast · jobs"}, {n:"S3 (optional)", s:"model & artifact store"}, ]; const CHX=SPX+24, CHW=SPW-48, CHH=104; let cy=SPY+82; @@ -151,14 +153,15 @@ fY.forEach((y)=>{ arrow(FX+FW, y+FH/2, SPX, ty, INK); }); -// NATS messaging bus -> workers (dashed). Workers coordinate via NATS; -// PostgreSQL is the frontends' shared state, not something workers connect to. -const natsY = SPY+82+CHH+18 + CHH/2; // NATS chip center y -wY.forEach((y)=> arrow(SPX+SPW, natsY, WX, y+WH/2, RUSTD, "2 8")); -// label the NATS bus arrows -const labW=140, labH=26, labX=(SPX+SPW+WX)/2-labW/2, labY=natsY-46; +// frontends -> workers over the tunnel (dashed), routed in the clear band BELOW +// the state plane. The route is the point: control verbs go frontend to worker +// and never touch the plane, and the worker opened the connection outward, so +// nothing dials into a worker and no port is published on one. +const tunY = 430; +arrow(FX+FW, tunY, WX, tunY, RUSTD, "2 8"); +const labW=250, labH=26, labX=(FX+FW+WX)/2-labW/2, labY=tunY-40; svg.appendChild(el("rect",{x:labX,y:labY,width:labW,height:labH,fill:PAPER,stroke:RUSTD,"stroke-width":2})); -txt(labX+labW/2,labY+18,"backend.install",{f:"Bricolage Grotesque",w:700,sz:14,a:"middle",fill:RUSTD}); +txt(labX+labW/2,labY+18,"backend.install \u00b7 worker-dialled tunnel",{f:"Bricolage Grotesque",w:700,sz:14,a:"middle",fill:RUSTD}); // ---- annotated arrow: frontend -> worker : LoadModel (gRPC) ---- arrow(FX+FW, fY[2]+FH-24, WX, wY[2]+WH-30, COLD, "4 7"); diff --git a/docs/static/images/diagrams/distributed-mode-arch.png b/docs/static/images/diagrams/distributed-mode-arch.png index 52117732e..e4b1d530d 100644 Binary files a/docs/static/images/diagrams/distributed-mode-arch.png and b/docs/static/images/diagrams/distributed-mode-arch.png differ diff --git a/pkg/natsauth/permissions_coverage_test.go b/pkg/natsauth/permissions_coverage_test.go index b06c45c42..e5b9db21a 100644 --- a/pkg/natsauth/permissions_coverage_test.go +++ b/pkg/natsauth/permissions_coverage_test.go @@ -1,8 +1,6 @@ package natsauth_test import ( - "os" - "regexp" "strings" "github.com/mudler/LocalAI/core/services/messaging" @@ -157,32 +155,10 @@ var _ = Describe("WorkerPermissions subject coverage", func() { }) }) -var allowPubRe = regexp.MustCompile(`--allow-pub "([^"]*)"`) - -var _ = Describe("Documented NATS service-user permissions", func() { - // scripts/nats-auth-setup.sh ships the recommended service (frontend) JWT - // permissions. They must cover every subject the frontend actually publishes, - // or prefix-cache sync (and friends) break once LOCALAI_NATS_REQUIRE_AUTH is on. - const scriptPath = "../../scripts/nats-auth-setup.sh" - - // Representative subjects the frontend publishes on the control plane. - // prefixcache.* is emitted by prefixcache.Sync in core/application/distributed.go. - frontendPublishes := []string{ - messaging.SubjectPrefixCacheObserve, - messaging.SubjectPrefixCacheInvalidate, - messaging.SubjectGalleryProgress("op-1"), - } - - It("cover every subject the frontend publishes", func() { - raw, err := os.ReadFile(scriptPath) - Expect(err).ToNot(HaveOccurred(), "cannot read %s", scriptPath) - m := allowPubRe.FindStringSubmatch(string(raw)) - Expect(m).To(HaveLen(2), "no --allow-pub list found in %s", scriptPath) - allow := strings.Split(m[1], ",") - - for _, subject := range frontendPublishes { - Expect(anyAllows(allow, subject)).To(BeTrue(), - "service-user --allow-pub %v does not cover %s (frontend publishes it)", allow, subject) - } - }) -}) +// The "Documented NATS service-user permissions" suite that stood here read +// scripts/nats-auth-setup.sh and required its --allow-pub list to cover every +// subject the frontend publishes. Both the script and the frontend's bus +// connection are gone: cross-replica fan-out is a PostgreSQL NOTIFY, which has +// no allow-list to fall out of sync with. The property was retired with the +// artifact it guarded, not moved, and the surviving suites above still pin the +// MINTED credentials, which are Task 17's to remove. diff --git a/scripts/nats-auth-setup.sh b/scripts/nats-auth-setup.sh deleted file mode 100755 index fb29fda49..000000000 --- a/scripts/nats-auth-setup.sh +++ /dev/null @@ -1,220 +0,0 @@ -#!/usr/bin/env bash -# Generate NATS JWT authentication material and server configuration -# for LocalAI distributed mode. -# -# Requires: nsc (https://docs.nats.io/running-a-nats-service/configuration/securing_nats/auth_intro/nsc) -# -# Outputs: -# ./nats-keys/localai-nats.env -# ./nats-keys/localai-frontend.creds -# ./nats-keys/nats-auth.conf -# ./nats-keys/nats-server.conf -# -# Environment overrides: -# NATS_OPERATOR_NAME -# NATS_ACCOUNT_NAME -# NATS_SERVICE_USER -# NATS_KEYS_DIR - -# -# LocalAI workers receive their own JWT and user seed when registering -# with the frontend. - -set -euo pipefail - -# Ensure newly created secret files are private by default. -umask 077 - -if ! command -v nsc >/dev/null 2>&1; then - echo "nsc is required. Install from https://github.com/nats-io/nsc/releases" >&2 - exit 1 -fi - -OPERATOR="${NATS_OPERATOR_NAME:-localai-operator}" -ACCOUNT="${NATS_ACCOUNT_NAME:-localai}" -SYSTEM_ACCOUNT="${NATS_SYSTEM_ACCOUNT_NAME:-SYS}" -SERVICE_USER="${NATS_SERVICE_USER:-localai-frontend}" -OUTPUT_DIR="${NATS_KEYS_DIR:-./nats-keys}" - -CREDS_FILE="$OUTPUT_DIR/${SERVICE_USER}.creds" -ENV_FILE="$OUTPUT_DIR/localai-nats.env" -AUTH_CONFIG_FILE="$OUTPUT_DIR/nats-auth.conf" -SERVER_CONFIG_FILE="$OUTPUT_DIR/nats-server.conf" - -mkdir -p "$OUTPUT_DIR" - -echo "Configuring NATS operator: $OPERATOR" - -# Create the operator if it does not exist, otherwise select it. -if nsc select operator "$OPERATOR" >/dev/null 2>&1; then - echo "[ OK ] using existing operator '$OPERATOR'" -else - nsc add operator \ - -n "$OPERATOR" \ - --generate-signing-key - - nsc select operator "$OPERATOR" >/dev/null -fi - -# Create and assign the NATS system account. -if nsc describe account \ - -n "$SYSTEM_ACCOUNT" >/dev/null 2>&1; then - echo "[ OK ] using existing system account '$SYSTEM_ACCOUNT'" -else - nsc add account -n "$SYSTEM_ACCOUNT" -fi - -nsc edit operator \ - --system-account "$SYSTEM_ACCOUNT" - -# Create the LocalAI account if it does not exist. -if nsc describe account -n "$ACCOUNT" >/dev/null 2>&1; then - echo "[ OK ] using existing account '$ACCOUNT'" -else - nsc add account -n "$ACCOUNT" -fi - -nsc select account "$ACCOUNT" >/dev/null - -# Create the frontend service user if it does not exist. -if nsc describe user \ - -n "$SERVICE_USER" \ - --account "$ACCOUNT" >/dev/null 2>&1; then - echo "[ OK ] using existing user '$SERVICE_USER'" -else - nsc add user \ - -n "$SERVICE_USER" \ - --account "$ACCOUNT" -fi - -# Frontend control-plane permissions. -nsc edit user \ - -n "$SERVICE_USER" \ - --account "$ACCOUNT" \ - --allow-pub "nodes.>,gallery.>,agent.>,staging.>,state.>,jobs.>,mcp.>,cache.>,prefixcache.>,finetune.>" \ - --allow-sub "nodes.>,gallery.>,agent.>,staging.>,state.>,jobs.>,mcp.>,cache.>,prefixcache.>,_INBOX.>" - -# Generate a credentials file containing the frontend user JWT and seed. -rm -f "$CREDS_FILE" - -nsc generate creds \ - -a "$ACCOUNT" \ - -n "$SERVICE_USER" \ - -o "$CREDS_FILE" - -# Extract the frontend JWT from the credentials file. -SERVICE_JWT="$( - awk ' - /BEGIN NATS USER JWT/ { - capture = 1 - next - } - /END NATS USER JWT/ { - capture = 0 - } - capture - ' "$CREDS_FILE" | tr -d '\r\n' -)" - -# Extract the frontend user seed from the credentials file. -SERVICE_SEED="$( - awk ' - /BEGIN USER NKEY SEED/ { - capture = 1 - next - } - /END USER NKEY SEED/ { - capture = 0 - } - capture - ' "$CREDS_FILE" | tr -d '\r\n' -)" - -# Retrieve the seed belonging to this exact account rather than taking -# the first account key found in the keystore. -ACCOUNT_SEED="$( - nsc list keys \ - --account "$ACCOUNT" \ - --accounts \ - --show-seeds | - awk -F '|' -v expected="$ACCOUNT" ' - function trim(value) { - gsub(/^[[:space:]]+|[[:space:]]+$/, "", value) - return value - } - - NF >= 3 { - entity = trim($2) - seed = trim($3) - - if (entity == expected && seed ~ /^SA[A-Z0-9]+$/) { - print seed - exit - } - } - ' -)" - -# Validate all extracted values before writing output files. -if [[ ! "$ACCOUNT_SEED" =~ ^SA[A-Z0-9]+$ ]]; then - echo "Unable to extract the account seed for '$ACCOUNT'." >&2 - exit 1 -fi - -if [[ ! "$SERVICE_JWT" =~ ^eyJ ]]; then - echo "Unable to extract the service JWT from '$CREDS_FILE'." >&2 - exit 1 -fi - -if [[ ! "$SERVICE_SEED" =~ ^SU[A-Z0-9]+$ ]]; then - echo "Unable to extract the service seed from '$CREDS_FILE'." >&2 - exit 1 -fi - -# Generate the trusted operator and memory resolver configuration. -# This contains public operator/account JWT claims, not the private seeds. -nsc generate config \ - --mem-resolver \ - --config-file "$AUTH_CONFIG_FILE" \ - --force - -# Generate the primary NATS server configuration. -# The include path matches the Docker Compose mounts shown below. -cat >"$SERVER_CONFIG_FILE" <<'NATS_CONFIG' -server_name: localai-nats -port: 4222 -http: 8222 - -jetstream { - store_dir: /data/jetstream -} - -include nats-auth.conf -NATS_CONFIG - -# Generate the environment file consumed by the LocalAI frontend. -cat >"$ENV_FILE" <=", 2)) - - eventMu.Lock() - defer eventMu.Unlock() - - var hasAgentMessage, hasCompleted bool - for _, evt := range receivedEvents { - if evt.EventType == "json_message" && evt.Sender == "agent" { - hasAgentMessage = true - Expect(evt.Content).To(ContainSubstring("systems operational")) - } - if evt.EventType == "json_message_status" && evt.Metadata != "" { - var meta map[string]string - json.Unmarshal([]byte(evt.Metadata), &meta) - if meta["status"] == "completed" { - hasCompleted = true + for _, evt := range receivedEvents { + if evt.EventType == "json_message" && evt.Sender == "agent" { + agentMessage = true + Expect(evt.Content).To(ContainSubstring("systems operational")) + } + if evt.EventType == "json_message_status" && evt.Metadata != "" { + var meta map[string]string + // A metadata blob this spec cannot read is not a + // completed status, and saying so beats failing the + // whole run on one malformed event. + if err := json.Unmarshal([]byte(evt.Metadata), &meta); err == nil && meta["status"] == "completed" { + completed = true + } } } + return agentMessage, completed } - Expect(hasAgentMessage).To(BeTrue(), "background agent should produce a response via EventBridge") - Expect(hasCompleted).To(BeTrue(), "background agent should complete") + + Eventually(func() bool { _, completed := seen(); return completed }, "15s"). + Should(BeTrue(), "background agent should complete") + agentMessage, _ := seen() + Expect(agentMessage).To(BeTrue(), "background agent should produce a response via EventBridge") }) }) diff --git a/tests/e2e/distributed/cluster_baseline_test.go b/tests/e2e/distributed/cluster_baseline_test.go index 6f69898c2..e6cb851c6 100644 --- a/tests/e2e/distributed/cluster_baseline_test.go +++ b/tests/e2e/distributed/cluster_baseline_test.go @@ -288,10 +288,16 @@ func startClusterOnFreshDB(frontends, workers int, customise ...func(*cluster.Op Binary: binary, MockBackend: mockBackend, PGDSN: infra.PGURL, - NatsURL: infra.NatsURL, - LogDir: logDir, - Frontends: frontends, - Workers: workers, + // Deliberately a dead address. Frontends and agent workers are still + // handed LOCALAI_NATS_URL so this suite keeps covering the promise that + // an operator's existing command line starts unchanged after the broker + // is shut down. Pointing it at a running server would make a regression + // that dialled it invisible; pointing it at nothing makes such a + // regression a startup failure in every cluster spec. + NatsURL: staleBusURL, + LogDir: logDir, + Frontends: frontends, + Workers: workers, } for _, apply := range customise { apply(&options) diff --git a/tests/e2e/distributed/cluster_control_test.go b/tests/e2e/distributed/cluster_control_test.go index 9daad1b5c..c6f213f83 100644 --- a/tests/e2e/distributed/cluster_control_test.go +++ b/tests/e2e/distributed/cluster_control_test.go @@ -413,8 +413,11 @@ var _ = Describe("Control plane over the worker tunnel", Label("Distributed"), L Expect(entry).ToNot(HavePrefix("LOCALAI_NATS_URL="), "the worker was handed a bus URL, so this spec is not about a worker that has none") } - // And the deployment it joined DOES have a bus, so "no NATS anywhere" - // is not what makes this pass. + // And the harness DOES still hand a LOCALAI_NATS_URL to the other + // processes in this cluster, so the worker's lack of one is a property + // of the worker and not of a harness that stopped setting the variable + // at all. There is no broker behind that URL any more, which is the + // point: nothing dials it, so nothing notices. Expect(c.NatsURL()).ToNot(BeEmpty()) probe := newRosterProbe(c, client, 0) diff --git a/tests/e2e/distributed/foundation_test.go b/tests/e2e/distributed/foundation_test.go index c96bd71e5..cca2e271f 100644 --- a/tests/e2e/distributed/foundation_test.go +++ b/tests/e2e/distributed/foundation_test.go @@ -30,7 +30,6 @@ var _ = Describe("Phase 0: Foundation", Label("Distributed"), func() { It("should reject --distributed without PostgreSQL configured", func() { appCfg := config.NewApplicationConfig( config.EnableDistributed, - config.WithNatsURL(infra.NatsURL), // No auth/PostgreSQL configured ) Expect(appCfg.Distributed.Enabled).To(BeTrue()) @@ -38,26 +37,30 @@ var _ = Describe("Phase 0: Foundation", Label("Distributed"), func() { Expect(appCfg.Auth.Enabled).To(BeFalse()) }) - It("should reject --distributed without NATS configured", func() { + It("leaves the inert bus URL empty when nothing sets it", func() { appCfg := config.NewApplicationConfig( config.EnableDistributed, config.WithAuthEnabled(true), config.WithAuthDatabaseURL(infra.PGURL), - // No NATS URL ) Expect(appCfg.Distributed.NatsURL).To(BeEmpty()) }) It("should accept valid distributed configuration", func() { + // staleBusURL points at nothing on purpose. It is the shape of an + // operator's existing command line after the broker was shut down, + // and the promise this spec holds is that such a command line still + // STARTS: the value is parsed, stored and never dialled. Pointing + // it at a live server would let a regression that dialled it pass. appCfg := config.NewApplicationConfig( config.EnableDistributed, config.WithAuthEnabled(true), config.WithAuthDatabaseURL(infra.PGURL), - config.WithNatsURL(infra.NatsURL), + config.WithNatsURL(staleBusURL), ) Expect(appCfg.Distributed.Enabled).To(BeTrue()) Expect(appCfg.Auth.Enabled).To(BeTrue()) - Expect(appCfg.Distributed.NatsURL).To(Equal(infra.NatsURL)) + Expect(appCfg.Distributed.NatsURL).To(Equal(staleBusURL)) }) It("should generate unique frontend ID on startup", func() { diff --git a/tests/e2e/distributed/gallery_distributed_test.go b/tests/e2e/distributed/gallery_distributed_test.go index b6bf9a6e2..b5b4294e4 100644 --- a/tests/e2e/distributed/gallery_distributed_test.go +++ b/tests/e2e/distributed/gallery_distributed_test.go @@ -70,8 +70,9 @@ var _ = Describe("Gallery Distributed", Label("Distributed"), func() { // The gallery families ride the broadcast carrier, not NATS. // - // These used to publish and subscribe on infra.NC, which asserted that NATS - // delivers to itself and nothing about this deployment: they would have + // These used to publish and subscribe on a message-bus client, which + // asserted that the bus delivers to itself and nothing about this + // deployment: they would have // stayed green through the whole migration while the gallery service had // already moved. Two carriers on the deployment's own database is the shape // a fleet has, and it is the shape that fails when one end moves and the diff --git a/tests/e2e/distributed/mcp_ci_job_helper_test.go b/tests/e2e/distributed/mcp_ci_job_helper_test.go index 27370b476..9210c1f62 100644 --- a/tests/e2e/distributed/mcp_ci_job_helper_test.go +++ b/tests/e2e/distributed/mcp_ci_job_helper_test.go @@ -16,10 +16,20 @@ import ( "github.com/mudler/xlog" ) +// publishTestEvent is the one place this helper publishes from, so a carrier +// that refuses a publish is reported rather than swallowed. A dropped progress +// line has no other symptom: the spec's subscriber simply never fires and the +// failure surfaces as a timeout that names the wrong thing. +func publishTestEvent(pub messaging.Publisher, subject string, payload any) { + if err := pub.Publish(subject, payload); err != nil { + xlog.Error("test worker failed to publish", "subject", subject, "error", err) + } +} + // processMCPCIJobForTest replicates the logic of handleMCPCIJob from agent_worker.go // for testing purposes. This allows e2e testing of the full MCP CI job execution path // without needing to start an actual agent worker binary. -func processMCPCIJobForTest(data []byte, apiURL, apiToken string, natsClient *messaging.Client) { +func processMCPCIJobForTest(data []byte, apiURL, apiToken string, pub messaging.Publisher) { var evt jobs.JobEvent if err := json.Unmarshal(data, &evt); err != nil { xlog.Error("Failed to unmarshal job event", "error", err) @@ -30,30 +40,30 @@ func processMCPCIJobForTest(data []byte, apiURL, apiToken string, natsClient *me task := evt.Task if job == nil || task == nil { xlog.Error("MCP CI job missing enriched data", "jobID", evt.JobID) - publishTestJobResult(natsClient, evt.JobID, "failed", "", "job or task data missing from NATS event") + publishTestJobResult(pub, evt.JobID, "failed", "", "job or task data missing from the job event") return } modelCfg := evt.ModelConfig if modelCfg == nil { - publishTestJobResult(natsClient, evt.JobID, "failed", "", "model config missing from job event") + publishTestJobResult(pub, evt.JobID, "failed", "", "model config missing from job event") return } // Publish running status - natsClient.Publish(messaging.SubjectJobProgress(evt.JobID), jobs.ProgressEvent{ + publishTestEvent(pub, messaging.SubjectJobProgress(evt.JobID), jobs.ProgressEvent{ JobID: evt.JobID, Status: "running", Message: "Job started on test worker", }) // Parse MCP config if modelCfg.MCP.Servers == "" && modelCfg.MCP.Stdio == "" { - publishTestJobResult(natsClient, evt.JobID, "failed", "", "no MCP servers configured for model") + publishTestJobResult(pub, evt.JobID, "failed", "", "no MCP servers configured for model") return } remote, stdio, err := modelCfg.MCP.MCPConfigFromYAML() if err != nil { - publishTestJobResult(natsClient, evt.JobID, "failed", "", fmt.Sprintf("failed to parse MCP config: %v", err)) + publishTestJobResult(pub, evt.JobID, "failed", "", fmt.Sprintf("failed to parse MCP config: %v", err)) return } @@ -64,7 +74,7 @@ func processMCPCIJobForTest(data []byte, apiURL, apiToken string, natsClient *me if err != nil { errMsg = fmt.Sprintf("failed to create MCP sessions: %v", err) } - publishTestJobResult(natsClient, evt.JobID, "failed", "", errMsg) + publishTestJobResult(pub, evt.JobID, "failed", "", errMsg) return } @@ -92,7 +102,7 @@ func processMCPCIJobForTest(data []byte, apiURL, apiToken string, natsClient *me defer cancel() // Publish running status - publishTestJobStatus(natsClient, evt.JobID, "running", "") + publishTestJobStatus(pub, evt.JobID, "running", "") // Buffer stream tokens var reasoningBuf, contentBuf strings.Builder @@ -100,13 +110,13 @@ func processMCPCIJobForTest(data []byte, apiURL, apiToken string, natsClient *me flushStreamBuf := func() { if reasoningBuf.Len() > 0 { - natsClient.Publish(messaging.SubjectJobProgress(evt.JobID), jobs.ProgressEvent{ + publishTestEvent(pub, messaging.SubjectJobProgress(evt.JobID), jobs.ProgressEvent{ JobID: evt.JobID, TraceType: "reasoning", TraceContent: reasoningBuf.String(), }) reasoningBuf.Reset() } if contentBuf.Len() > 0 { - natsClient.Publish(messaging.SubjectJobProgress(evt.JobID), jobs.ProgressEvent{ + publishTestEvent(pub, messaging.SubjectJobProgress(evt.JobID), jobs.ProgressEvent{ JobID: evt.JobID, TraceType: "content", TraceContent: contentBuf.String(), }) contentBuf.Reset() @@ -119,13 +129,13 @@ func processMCPCIJobForTest(data []byte, apiURL, apiToken string, natsClient *me cogito.WithMCPs(sessions...), cogito.WithStatusCallback(func(status string) { flushStreamBuf() - natsClient.Publish(messaging.SubjectJobProgress(evt.JobID), jobs.ProgressEvent{ + publishTestEvent(pub, messaging.SubjectJobProgress(evt.JobID), jobs.ProgressEvent{ JobID: evt.JobID, TraceType: "status", TraceContent: status, }) }), cogito.WithToolCallResultCallback(func(t cogito.ToolStatus) { flushStreamBuf() - natsClient.Publish(messaging.SubjectJobProgress(evt.JobID), jobs.ProgressEvent{ + publishTestEvent(pub, messaging.SubjectJobProgress(evt.JobID), jobs.ProgressEvent{ JobID: evt.JobID, TraceType: "tool_result", TraceContent: fmt.Sprintf("%s: %s", t.Name, t.Result), }) }), @@ -140,7 +150,7 @@ func processMCPCIJobForTest(data []byte, apiURL, apiToken string, natsClient *me case cogito.StreamEventContent: contentBuf.WriteString(ev.Content) case cogito.StreamEventToolCall: - natsClient.Publish(messaging.SubjectJobProgress(evt.JobID), jobs.ProgressEvent{ + publishTestEvent(pub, messaging.SubjectJobProgress(evt.JobID), jobs.ProgressEvent{ JobID: evt.JobID, TraceType: "tool_call", TraceContent: fmt.Sprintf("%s(%s)", ev.ToolName, ev.ToolArgs), }) } @@ -155,7 +165,7 @@ func processMCPCIJobForTest(data []byte, apiURL, apiToken string, natsClient *me flushStreamBuf() if err != nil { - publishTestJobResult(natsClient, evt.JobID, "failed", "", fmt.Sprintf("cogito execution failed: %v", err)) + publishTestJobResult(pub, evt.JobID, "failed", "", fmt.Sprintf("cogito execution failed: %v", err)) return } @@ -163,27 +173,27 @@ func processMCPCIJobForTest(data []byte, apiURL, apiToken string, natsClient *me if msg := f.LastMessage(); msg != nil { result = msg.Content } - publishTestJobResult(natsClient, evt.JobID, "completed", result, "") + publishTestJobResult(pub, evt.JobID, "completed", result, "") } -func publishTestJobStatus(nc *messaging.Client, jobID, status, message string) { - nc.Publish(messaging.SubjectJobResult(jobID), jobs.JobResultEvent{ +func publishTestJobStatus(pub messaging.Publisher, jobID, status, message string) { + publishTestEvent(pub, messaging.SubjectJobResult(jobID), jobs.JobResultEvent{ JobID: jobID, Status: status, }) - nc.Publish(messaging.SubjectJobProgress(jobID), jobs.ProgressEvent{ + publishTestEvent(pub, messaging.SubjectJobProgress(jobID), jobs.ProgressEvent{ JobID: jobID, Status: status, Message: message, }) } -func publishTestJobResult(nc *messaging.Client, jobID, status, result, errMsg string) { - nc.Publish(messaging.SubjectJobResult(jobID), jobs.JobResultEvent{ +func publishTestJobResult(pub messaging.Publisher, jobID, status, result, errMsg string) { + publishTestEvent(pub, messaging.SubjectJobResult(jobID), jobs.JobResultEvent{ JobID: jobID, Status: status, Result: result, Error: errMsg, }) - nc.Publish(messaging.SubjectJobProgress(jobID), jobs.ProgressEvent{ + publishTestEvent(pub, messaging.SubjectJobProgress(jobID), jobs.ProgressEvent{ JobID: jobID, Status: status, Message: errMsg, }) } diff --git a/tests/e2e/distributed/mcp_ci_job_test.go b/tests/e2e/distributed/mcp_ci_job_test.go index 62c696850..638d4397a 100644 --- a/tests/e2e/distributed/mcp_ci_job_test.go +++ b/tests/e2e/distributed/mcp_ci_job_test.go @@ -280,10 +280,18 @@ func startMockLLMServer() (string, func()) { var _ = Describe("MCP CI Job Execution", Label("Distributed", "MCPCIJob"), func() { var ( infra *TestInfra + bus messaging.Broadcaster ) + // One carrier per spec, on THIS spec's database. It is the PostgreSQL + // LISTEN/NOTIFY carrier the deployment runs, and it is what the SSE routes + // read: a spec that published its progress and result lines onto a message + // bus instead would keep passing after that carrier had gone silent, + // because a publish onto the wrong carrier succeeds and simply arrives + // nowhere. BeforeEach(func() { - infra = SetupNATSOnly() + infra = SetupInfra("mcp_ci_job") + bus = infra.Bus() }) Context("Full MCP CI Job Flow", func() { @@ -303,7 +311,7 @@ var _ = Describe("MCP CI Job Execution", Label("Distributed", "MCPCIJob"), func( var resultEvent *jobs.JobResultEvent var eventMu sync.Mutex - progressSub, err := infra.NC.Subscribe(messaging.SubjectJobProgress(jobID), func(data []byte) { + progressSub, err := bus.Subscribe(messaging.SubjectJobProgress(jobID), func(data []byte) { var evt jobs.ProgressEvent if json.Unmarshal(data, &evt) == nil { eventMu.Lock() @@ -314,7 +322,7 @@ var _ = Describe("MCP CI Job Execution", Label("Distributed", "MCPCIJob"), func( Expect(err).ToNot(HaveOccurred()) defer progressSub.Unsubscribe() - resultSub, err := infra.NC.Subscribe(messaging.SubjectJobResult(jobID), func(data []byte) { + resultSub, err := bus.Subscribe(messaging.SubjectJobResult(jobID), func(data []byte) { var evt jobs.JobResultEvent if json.Unmarshal(data, &evt) == nil { eventMu.Lock() @@ -325,8 +333,6 @@ var _ = Describe("MCP CI Job Execution", Label("Distributed", "MCPCIJob"), func( Expect(err).ToNot(HaveOccurred()) defer resultSub.Unsubscribe() - FlushNATS(infra.NC) - // Build MCP config YAML pointing to mock MCP server mcpRemoteJSON := fmt.Sprintf(`{"mcpServers":{"weather-api":{"url":"%s"}}}`, mcpURL) modelCfg := &config.ModelConfig{ @@ -364,9 +370,7 @@ var _ = Describe("MCP CI Job Execution", Label("Distributed", "MCPCIJob"), func( raw, err := json.Marshal(evt) Expect(err).ToNot(HaveOccurred()) - FlushNATS(infra.NC) - - go processMCPCIJobForTest(raw, llmURL, "test-token", infra.NC) + go processMCPCIJobForTest(raw, llmURL, "test-token", bus) // Wait for result Eventually(func() bool { @@ -404,7 +408,7 @@ var _ = Describe("MCP CI Job Execution", Label("Distributed", "MCPCIJob"), func( var resultEvent *jobs.JobResultEvent var eventMu sync.Mutex - resultSub, err := infra.NC.Subscribe(messaging.SubjectJobResult(jobID), func(data []byte) { + resultSub, err := bus.Subscribe(messaging.SubjectJobResult(jobID), func(data []byte) { var evt jobs.JobResultEvent if json.Unmarshal(data, &evt) == nil { eventMu.Lock() @@ -415,8 +419,6 @@ var _ = Describe("MCP CI Job Execution", Label("Distributed", "MCPCIJob"), func( Expect(err).ToNot(HaveOccurred()) defer resultSub.Unsubscribe() - FlushNATS(infra.NC) - // MCP config pointing to unreachable server mcpRemoteJSON := `{"mcpServers":{"bad-server":{"url":"http://127.0.0.1:1/mcp"}}}` modelCfg := &config.ModelConfig{ @@ -448,7 +450,7 @@ var _ = Describe("MCP CI Job Execution", Label("Distributed", "MCPCIJob"), func( // Process directly (no worker subscription needed) evtData, _ := json.Marshal(evt) - go processMCPCIJobForTest(evtData, "http://localhost:9999", "token", infra.NC) + go processMCPCIJobForTest(evtData, "http://localhost:9999", "token", bus) // Wait for failure result Eventually(func() bool { @@ -477,7 +479,7 @@ var _ = Describe("MCP CI Job Execution", Label("Distributed", "MCPCIJob"), func( var resultEvent *jobs.JobResultEvent var eventMu sync.Mutex - resultSub, err := infra.NC.Subscribe(messaging.SubjectJobResult(jobID), func(data []byte) { + resultSub, err := bus.Subscribe(messaging.SubjectJobResult(jobID), func(data []byte) { var evt jobs.JobResultEvent if json.Unmarshal(data, &evt) == nil { eventMu.Lock() @@ -488,8 +490,6 @@ var _ = Describe("MCP CI Job Execution", Label("Distributed", "MCPCIJob"), func( Expect(err).ToNot(HaveOccurred()) defer resultSub.Unsubscribe() - FlushNATS(infra.NC) - mcpRemoteJSON := fmt.Sprintf(`{"mcpServers":{"weather-api":{"url":"%s"}}}`, mcpURL) modelCfg := &config.ModelConfig{ MCP: config.MCPConfig{ @@ -523,7 +523,7 @@ var _ = Describe("MCP CI Job Execution", Label("Distributed", "MCPCIJob"), func( } evtData, _ := json.Marshal(evt) - go processMCPCIJobForTest(evtData, llmURL, "test-token", infra.NC) + go processMCPCIJobForTest(evtData, llmURL, "test-token", bus) Eventually(func() bool { eventMu.Lock() @@ -545,7 +545,7 @@ var _ = Describe("MCP CI Job Execution", Label("Distributed", "MCPCIJob"), func( var resultEvent *jobs.JobResultEvent var eventMu sync.Mutex - resultSub, err := infra.NC.Subscribe(messaging.SubjectJobResult(jobID), func(data []byte) { + resultSub, err := bus.Subscribe(messaging.SubjectJobResult(jobID), func(data []byte) { var evt jobs.JobResultEvent if json.Unmarshal(data, &evt) == nil { eventMu.Lock() @@ -556,8 +556,6 @@ var _ = Describe("MCP CI Job Execution", Label("Distributed", "MCPCIJob"), func( Expect(err).ToNot(HaveOccurred()) defer resultSub.Unsubscribe() - FlushNATS(infra.NC) - // Event with no Job or Task evt := jobs.JobEvent{ JobID: jobID, @@ -566,7 +564,7 @@ var _ = Describe("MCP CI Job Execution", Label("Distributed", "MCPCIJob"), func( } evtData, _ := json.Marshal(evt) - go processMCPCIJobForTest(evtData, "http://localhost:9999", "token", infra.NC) + go processMCPCIJobForTest(evtData, "http://localhost:9999", "token", bus) Eventually(func() bool { eventMu.Lock() @@ -586,7 +584,7 @@ var _ = Describe("MCP CI Job Execution", Label("Distributed", "MCPCIJob"), func( var resultEvent *jobs.JobResultEvent var eventMu sync.Mutex - resultSub, err := infra.NC.Subscribe(messaging.SubjectJobResult(jobID), func(data []byte) { + resultSub, err := bus.Subscribe(messaging.SubjectJobResult(jobID), func(data []byte) { var evt jobs.JobResultEvent if json.Unmarshal(data, &evt) == nil { eventMu.Lock() @@ -597,8 +595,6 @@ var _ = Describe("MCP CI Job Execution", Label("Distributed", "MCPCIJob"), func( Expect(err).ToNot(HaveOccurred()) defer resultSub.Unsubscribe() - FlushNATS(infra.NC) - // ModelConfig with empty MCP modelCfg := &config.ModelConfig{} modelCfg.Name = "no-mcp-model" @@ -624,7 +620,7 @@ var _ = Describe("MCP CI Job Execution", Label("Distributed", "MCPCIJob"), func( } evtData, _ := json.Marshal(evt) - go processMCPCIJobForTest(evtData, "http://localhost:9999", "token", infra.NC) + go processMCPCIJobForTest(evtData, "http://localhost:9999", "token", bus) Eventually(func() bool { eventMu.Lock() diff --git a/tests/e2e/distributed/nats_jwt_helpers_test.go b/tests/e2e/distributed/nats_jwt_helpers_test.go deleted file mode 100644 index 74f74355d..000000000 --- a/tests/e2e/distributed/nats_jwt_helpers_test.go +++ /dev/null @@ -1,156 +0,0 @@ -package distributed_test - -import ( - "bytes" - "context" - "fmt" - "strings" - "time" - - "github.com/mudler/LocalAI/core/services/messaging" - "github.com/mudler/LocalAI/pkg/natsauth" - "github.com/nats-io/jwt/v2" - "github.com/nats-io/nkeys" - - . "github.com/onsi/ginkgo/v2" - . "github.com/onsi/gomega" - - "github.com/testcontainers/testcontainers-go" - tcnats "github.com/testcontainers/testcontainers-go/modules/nats" -) - -// JWTTestInfra holds a NATS server configured with JWT auth and minted worker credentials. -type JWTTestInfra struct { - *TestInfra - AccountSeed string - NodeID string - WorkerJWT string - WorkerSeed string -} - -// SetupJWTInfra starts NATS with an in-memory JWT resolver and returns worker credentials -// minted the same way as node registration (pkg/natsauth). -func SetupJWTInfra() *JWTTestInfra { - GinkgoHelper() - - infra := &JWTTestInfra{TestInfra: &TestInfra{Ctx: context.Background()}} - - operatorJWT, accountJWT, accountSeed, err := jwtResolverMaterial() - Expect(err).ToNot(HaveOccurred()) - infra.AccountSeed = accountSeed - - conf := fmt.Sprintf(`listen: 0.0.0.0:4222 - -operator: %s - -resolver: MEMORY -resolver_preload: { - %s: %s -} -`, operatorJWT, accountPublicKeyFromSeed(accountSeed), accountJWT) - - var natsContainer *tcnats.NATSContainer - // Override default testcontainers -js: JetStream fails without a system account in JWT mode. - natsContainer, err = tcnats.Run(infra.Ctx, "nats:2-alpine", - tcnats.WithConfigFile(bytes.NewBufferString(conf)), - testcontainers.WithCmd("-c", "/etc/nats.conf"), - ) - Expect(err).ToNot(HaveOccurred()) - infra.NATSContainer = natsContainer - - infra.NatsURL, err = infra.NATSContainer.ConnectionString(infra.Ctx) - Expect(err).ToNot(HaveOccurred()) - - infra.NodeID = "550e8400-e29b-41d4-a716-446655440000" - cfg := natsauth.Config{AccountSeed: infra.AccountSeed, WorkerJWTTTL: time.Hour} - infra.WorkerJWT, infra.WorkerSeed, err = cfg.MintWorkerJWT(infra.NodeID, "backend") - Expect(err).ToNot(HaveOccurred()) - - infra.NC, err = messaging.New(infra.NatsURL, messaging.WithUserJWT(infra.WorkerJWT, infra.WorkerSeed)) - Expect(err).ToNot(HaveOccurred()) - - DeferCleanup(func() { - if infra.NC != nil { - infra.NC.Close() - } - if infra.NATSContainer != nil { - _ = infra.NATSContainer.Terminate(context.Background()) - } - }) - - return infra -} - -// jwtResolverMaterial builds operator + account JWTs for a MEMORY resolver. -// Follows the NATS JWT tutorial: self-signed account, then operator re-sign, with the -// account identity key listed as a signing key so MintWorkerJWT can use the account seed. -func jwtResolverMaterial() (operatorJWT, accountJWT, accountSeed string, err error) { - okp, err := nkeys.CreateOperator() - if err != nil { - return "", "", "", err - } - opk, err := okp.PublicKey() - if err != nil { - return "", "", "", err - } - oc := jwt.NewOperatorClaims(opk) - oc.Name = "localai-test-operator" - oskp, err := nkeys.CreateOperator() - if err != nil { - return "", "", "", err - } - ospk, err := oskp.PublicKey() - if err != nil { - return "", "", "", err - } - oc.SigningKeys.Add(ospk) - operatorJWT, err = oc.Encode(okp) - if err != nil { - return "", "", "", err - } - - akp, err := nkeys.CreateAccount() - if err != nil { - return "", "", "", err - } - seed, err := akp.Seed() - if err != nil { - return "", "", "", err - } - accountSeed = string(seed) - - apk, err := akp.PublicKey() - if err != nil { - return "", "", "", err - } - ac := jwt.NewAccountClaims(apk) - ac.Name = "localai-test-account" - ac.SigningKeys.Add(apk) - accountJWT, err = ac.Encode(akp) - if err != nil { - return "", "", "", err - } - ac, err = jwt.DecodeAccountClaims(accountJWT) - if err != nil { - return "", "", "", err - } - accountJWT, err = ac.Encode(oskp) - if err != nil { - return "", "", "", err - } - return operatorJWT, accountJWT, accountSeed, nil -} - -func accountPublicKeyFromSeed(accountSeed string) string { - akp, err := nkeys.FromSeed([]byte(accountSeed)) - Expect(err).ToNot(HaveOccurred()) - pk, err := akp.PublicKey() - Expect(err).ToNot(HaveOccurred()) - return pk -} - -// nodeSubjectPrefix returns the sanitized nodes.* prefix for a node ID. -func nodeSubjectPrefix(nodeID string) string { - tok := strings.NewReplacer(".", "-", "*", "-", ">", "-", " ", "-", "\t", "-", "\n", "-").Replace(nodeID) - return "nodes." + tok -} diff --git a/tests/e2e/distributed/nats_jwt_test.go b/tests/e2e/distributed/nats_jwt_test.go deleted file mode 100644 index 46444c197..000000000 --- a/tests/e2e/distributed/nats_jwt_test.go +++ /dev/null @@ -1,176 +0,0 @@ -package distributed_test - -import ( - "time" - - "github.com/mudler/LocalAI/core/services/messaging" - "github.com/mudler/LocalAI/pkg/natsauth" - - . "github.com/onsi/ginkgo/v2" - . "github.com/onsi/gomega" -) - -var _ = Describe("NATS JWT Auth", Label("Distributed", "NatsJWT"), func() { - var infra *JWTTestInfra - - BeforeEach(func() { - infra = SetupJWTInfra() - }) - - It("connects with a minted backend worker JWT and publishes on its one remaining allowed subject", func() { - // 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, 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()) - // ConfirmRoundTrip is the flush AND the server's verdict in one call. - // Read separately they were two assertions that could drift apart; the - // verdict is the one that matters, because a permission violation does - // not close the connection. - Expect(infra.NC.ConfirmRoundTrip(2 * time.Second)).To(Succeed()) - Expect(infra.NC.IsConnected()).To(BeTrue()) - }) - - It("denies a backend worker the file-staging subjects it no longer serves", func() { - // This spec used to assert the OPPOSITE, and kept passing after the - // grant was deleted. A NATS permission violation does not close the - // connection, so a spec that checks only FlushTimeout and IsConnected - // cannot tell an allowed publish from a denied one; LastError is what - // actually reads the server's verdict, which is why the sibling below - // has always used it. - subject := nodeSubjectPrefix(infra.NodeID) + ".files.stage" - Expect(infra.NC.Publish(subject, map[string]string{"path": "/tmp/model"})).To(Succeed()) - Eventually(func() error { - return infra.NC.ConfirmRoundTrip(500 * time.Millisecond) - }, "3s", "50ms").Should(HaveOccurred()) - }) - - 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) {}) - if err == nil { - defer func() { _ = sub.Unsubscribe() }() - Eventually(func() error { - return infra.NC.ConfirmRoundTrip(500 * time.Millisecond) - }, "3s", "50ms").Should(HaveOccurred()) - } - }) - - It("rejects anonymous publish on the JWT-enabled server", func() { - anon, err := messaging.New(infra.NatsURL) - Expect(err).ToNot(HaveOccurred()) - defer anon.Close() - - err = anon.Publish("nodes.any.files.x", map[string]string{"x": "1"}) - Expect(err).ToNot(HaveOccurred()) - Expect(anon.ConfirmRoundTrip(2 * time.Second)).To(HaveOccurred()) - }) - - It("denies backend publish to another node's subjects", func() { - other := nodeSubjectPrefix("other-node-id") + ".files.stage" - Expect(infra.NC.Publish(other, map[string]string{"stage": "nope"})).To(Succeed()) - Eventually(func() error { - return infra.NC.ConfirmRoundTrip(500 * time.Millisecond) - }, "3s", "50ms").Should(HaveOccurred()) - }) - - It("mints agent JWT without backend.install in claims", func() { - cfg := natsauth.Config{AccountSeed: infra.AccountSeed} - token, _, err := cfg.MintWorkerJWT("agent-node-1", "agent") - Expect(err).ToNot(HaveOccurred()) - - claims, err := natsauth.DecodeUserClaims(token) - Expect(err).ToNot(HaveOccurred()) - // agent.execute has left this list: agent execution is a streaming - // control verb on the tunnel now, driven by a claim a frontend replica - // took off the job store. What must still be here is the cancel - // broadcast, which cannot become an RPC. - Expect(claims.Permissions.Sub.Allow).To(ContainElement("agent.*.cancel")) - Expect(claims.Permissions.Sub.Allow).ToNot(ContainElement("agent.execute")) - for _, subj := range claims.Permissions.Sub.Allow { - Expect(subj).NotTo(ContainSubstring("backend.install")) - } - }) - - // Regression guard for the silent permission gaps: decoding the JWT claims - // (above) only proves the agent JWT is *restrictive*, not that it is - // *sufficient*. Stand a real agent connection up against the enforcing - // server and exercise every subscription core/cli/agent_worker.go actually - // makes — a denied SUB now surfaces synchronously via confirmSubscription, - // so a missing allow rule fails this test instead of silently dropping - // backend.stop / MCP-CI deliveries at runtime. - It("lets an agent-minted JWT establish all the subscriptions the agent worker uses", func() { - const nodeID = "agent-node-subs" - cfg := natsauth.Config{AccountSeed: infra.AccountSeed, WorkerJWTTTL: time.Hour} - token, seed, err := cfg.MintWorkerJWT(nodeID, "agent") - Expect(err).ToNot(HaveOccurred()) - - nc, err := messaging.New(infra.NatsURL, messaging.WithUserJWT(token, seed)) - Expect(err).ToNot(HaveOccurred()) - DeferCleanup(nc.Close) - - // Mirror core/cli/agent_worker.go exactly. MCP tool execution and - // discovery are absent, so is the per-node backend.stop, and so now are - // agent execution and MCP CI runs: none of them is a bus subject any - // more. The frontend selects an agent worker itself and reaches it with - // a control RPC over the tunnel that worker holds, and the work it - // hands over is a row it claimed on the job store. - // - // What is left is the cancel broadcast, which cannot become an RPC: the - // replica holding a run is not the one an API cancel lands on. - _, err = nc.Subscribe(messaging.SubjectAgentCancelWildcard, func([]byte) {}) - Expect(err).ToNot(HaveOccurred(), "agent JWT must allow %s (cancellation)", messaging.SubjectAgentCancelWildcard) - - _, err = nc.Subscribe(messaging.SubjectJobProgressWildcard, func([]byte) {}) - Expect(err).ToNot(HaveOccurred(), "agent JWT must allow %s (progress bridging)", messaging.SubjectJobProgressWildcard) - }) - - // The narrowing, proved against the enforcing server rather than against - // the allow list that feeds it. The subject is written out by hand because - // its builder is deleted; that literal is what a worker from an older - // release would still send, and this is what the server now answers it. - // - // It is a narrowing and not a lockout: the two subscriptions above are made - // on a JWT minted the same way and both succeed, so the list this trims is - // demonstrably not the empty one NATS would read as unrestricted. - It("refuses an agent-minted JWT the retired per-node backend.stop subject", func() { - const nodeID = "agent-node-stop" - cfg := natsauth.Config{AccountSeed: infra.AccountSeed, WorkerJWTTTL: time.Hour} - token, seed, err := cfg.MintWorkerJWT(nodeID, "agent") - Expect(err).ToNot(HaveOccurred()) - - nc, err := messaging.New(infra.NatsURL, messaging.WithUserJWT(token, seed)) - Expect(err).ToNot(HaveOccurred()) - DeferCleanup(nc.Close) - - _, err = nc.Subscribe("nodes."+nodeID+".backend.stop", func([]byte) {}) - Expect(err).To(HaveOccurred(), - "backend.stop is a control RPC on the worker's tunnel; the bus must not carry it") - }) - - // The same narrowing for the two queue subjects that became claim rows. - // Written out by hand for the same reason: those literals are what a worker - // from an older release would still subscribe to, and this is what the - // server now answers it. - DescribeTable("refuses an agent-minted JWT a retired queue subject", - func(subject string) { - cfg := natsauth.Config{AccountSeed: infra.AccountSeed, WorkerJWTTTL: time.Hour} - token, seed, err := cfg.MintWorkerJWT("agent-node-queues", "agent") - Expect(err).ToNot(HaveOccurred()) - - nc, err := messaging.New(infra.NatsURL, messaging.WithUserJWT(token, seed)) - Expect(err).ToNot(HaveOccurred()) - DeferCleanup(nc.Close) - - _, err = nc.Subscribe(subject, func([]byte) {}) - Expect(err).To(HaveOccurred(), - "%s became a claim on the job store; the bus must not carry it", subject) - }, - Entry("agent execution", "agent.execute"), - Entry("mcp ci jobs", "jobs.mcp-ci.new"), - ) -}) diff --git a/tests/e2e/distributed/testhelpers_test.go b/tests/e2e/distributed/testhelpers_test.go index 84b2f0112..0508f0675 100644 --- a/tests/e2e/distributed/testhelpers_test.go +++ b/tests/e2e/distributed/testhelpers_test.go @@ -8,14 +8,12 @@ import ( "sync/atomic" "time" - "github.com/mudler/LocalAI/core/services/messaging" "github.com/mudler/LocalAI/core/services/pgbus" . "github.com/onsi/ginkgo/v2" . "github.com/onsi/gomega" "github.com/testcontainers/testcontainers-go" - tcnats "github.com/testcontainers/testcontainers-go/modules/nats" tcpostgres "github.com/testcontainers/testcontainers-go/modules/postgres" "github.com/testcontainers/testcontainers-go/wait" "gorm.io/driver/postgres" @@ -23,37 +21,44 @@ import ( gormlogger "gorm.io/gorm/logger" ) -// TestInfra holds shared test containers and connection strings. +// TestInfra holds the shared test container and the connection strings derived +// from it. // -// PGContainer and NATSContainer are the SUITE-WIDE containers, shared by every -// spec. Never call Terminate or Stop on them from a spec: it ends the run for -// everything after it. They are exposed only because nats_jwt_helpers_test.go -// builds its own TestInfra around a dedicated NATS container. +// PGContainer is the SUITE-WIDE container, shared by every spec. Never call +// Terminate or Stop on it from a spec: it ends the run for everything after it. +// +// There is one container and not two. Every carrier these specs exercise is a +// PostgreSQL LISTEN/NOTIFY channel and every worker verb is an HTTP route on a +// tunnel, so a message broker in this suite would be infrastructure no spec and +// no production path can reach. type TestInfra struct { - Ctx context.Context - PGContainer *tcpostgres.PostgresContainer - NATSContainer *tcnats.NATSContainer - PGURL string - NatsURL string - NC *messaging.Client + Ctx context.Context + PGContainer *tcpostgres.PostgresContainer + PGURL string } -// Containers are suite-scoped, not spec-scoped. Starting a Postgres (~10s) and a -// NATS (~3.5s) per spec cost roughly 48 minutes of pure startup across the 213 -// specs behind SetupInfra, which is why this suite was never wired into CI. -// Isolation now comes from a database per spec (~67ms), which is what the dbName -// argument was always describing. +// staleBusURL is the address of a broker that is not running, and is not meant +// to be. LOCALAI_NATS_URL and --nats-url are still accepted and ignored so an +// operator's existing command line, unit file or Helm values file starts +// unchanged after the broker is shut down; specs that exercise that promise +// hand over THIS value, because a value pointing at a live server would let a +// regression that dialled it pass unnoticed. +const staleBusURL = "nats://127.0.0.1:1" + +// The container is suite-scoped, not spec-scoped. Starting a Postgres (~10s) per +// spec cost roughly 36 minutes of pure startup across the 213 specs behind +// SetupInfra, which is why this suite was never wired into CI. Isolation now +// comes from a database per spec (~67ms), which is what the dbName argument was +// always describing. // // Plain BeforeSuite rather than SynchronizedBeforeSuite is deliberate: under -// `ginkgo -p` each process gets its own container pair, which keeps NATS subjects -// isolated per process. A single shared NATS across parallel processes would let -// specs on different processes see each other's messages on the same subject. +// `ginkgo -p` each process gets its own container, and a database per spec on +// top of that keeps two processes from reading each other's notifications on a +// channel of the same name. var ( - suitePG *tcpostgres.PostgresContainer - suiteNATS *tcnats.NATSContainer - suitePGDSN string - suiteNatsURL string - dbCounter atomic.Int64 + suitePG *tcpostgres.PostgresContainer + suitePGDSN string + dbCounter atomic.Int64 ) var _ = BeforeSuite(func() { @@ -74,12 +79,6 @@ var _ = BeforeSuite(func() { suitePGDSN, err = suitePG.ConnectionString(ctx, "sslmode=disable") Expect(err).ToNot(HaveOccurred()) - - suiteNATS, err = tcnats.Run(ctx, "nats:2-alpine") - Expect(err).ToNot(HaveOccurred()) - - suiteNatsURL, err = suiteNATS.ConnectionString(ctx) - Expect(err).ToNot(HaveOccurred()) }) var _ = AfterSuite(func() { @@ -87,9 +86,6 @@ var _ = AfterSuite(func() { if suitePG != nil { _ = suitePG.Terminate(ctx) } - if suiteNATS != nil { - _ = suiteNATS.Terminate(ctx) - } }) // sanitizeDBName maps a spec-supplied label onto a legal unquoted Postgres @@ -156,18 +152,16 @@ func closeDB(db *gorm.DB) { } } -// SetupInfra provisions a dedicated database on the suite-scoped Postgres and -// returns a client connected to the suite-scoped NATS. Call in BeforeEach; -// cleanup is registered with DeferCleanup. +// SetupInfra provisions a dedicated database on the suite-scoped Postgres. Call +// in BeforeEach; cleanup is registered with DeferCleanup. A spec that needs a +// broadcast carrier opens one with Bus(). func SetupInfra(dbName string) *TestInfra { GinkgoHelper() - Expect(suitePG).ToNot(BeNil(), "SetupInfra called before BeforeSuite started the shared containers") + Expect(suitePG).ToNot(BeNil(), "SetupInfra called before BeforeSuite started the shared container") infra := &TestInfra{ - Ctx: context.Background(), - PGContainer: suitePG, - NATSContainer: suiteNATS, - NatsURL: suiteNatsURL, + Ctx: context.Background(), + PGContainer: suitePG, } db := fmt.Sprintf("%s_%d", sanitizeDBName(dbName), dbCounter.Add(1)) @@ -180,12 +174,9 @@ func SetupInfra(dbName string) *TestInfra { Expect(admin.Exec(fmt.Sprintf("CREATE DATABASE %q", db)).Error).To(Succeed()) }() - // Registered before anything else can fail: a NATS connect error below would - // otherwise leave the database behind for the rest of the suite. + // Registered immediately after the CREATE, so no later failure in this + // helper can leave the database behind for the rest of the suite. DeferCleanup(func() { - if infra.NC != nil { - infra.NC.Close() - } drop, err := tryAdminDB() if err != nil { AddReportEntry("drop database skipped", fmt.Sprintf("%s: %v", db, err)) @@ -200,63 +191,23 @@ func SetupInfra(dbName string) *TestInfra { infra.PGURL = replaceDBName(suitePGDSN, db) - var err error - infra.NC, err = messaging.New(infra.NatsURL) - Expect(err).ToNot(HaveOccurred()) - return infra } -// SetupNATSOnly returns a client on the suite-scoped NATS for specs that need no -// database. -func SetupNATSOnly() *TestInfra { - GinkgoHelper() - Expect(suiteNATS).ToNot(BeNil(), "SetupNATSOnly called before BeforeSuite started the shared containers") - - infra := &TestInfra{ - Ctx: context.Background(), - NATSContainer: suiteNATS, - NatsURL: suiteNatsURL, - } - - var err error - infra.NC, err = messaging.New(infra.NatsURL) - Expect(err).ToNot(HaveOccurred()) - - DeferCleanup(func() { - if infra.NC != nil { - infra.NC.Close() - } - }) - - return infra -} - -// FlushNATS ensures all subscriptions are registered server-side before publishing. -// -// It asserts the server's verdict too, not only that the round trip completed: -// on a permission-enforcing server a denied SUB leaves the connection open and -// the flush succeeding, so a helper that checked the flush alone would let a -// spec proceed to publish into a subscription the server had already refused. -func FlushNATS(nc *messaging.Client) { - GinkgoHelper() - Expect(nc.ConfirmRoundTrip(5 * time.Second)).To(Succeed()) -} - // Bus opens a broadcast carrier on THIS spec's database. // // It is the carrier the job, agent and response families travel on, and it is // what these specs must build their dispatchers and bridges with. Publishing on // one carrier while the subscriber reads another is a defect with no error // anywhere: the publish succeeds and the SSE stream is simply empty, so a spec -// that used the NATS client here would keep passing after production had gone -// silent. +// that reached for a message-bus client here would keep passing after +// production had gone silent. // // Every call returns a SEPARATE carrier on the same database, so a spec can // build two and assert across them, which is the shape a deployment has. func (i *TestInfra) Bus() *pgbus.Bus { GinkgoHelper() - Expect(i.PGURL).ToNot(BeEmpty(), "Bus needs a database; use SetupInfra rather than SetupNATSOnly") + Expect(i.PGURL).ToNot(BeEmpty(), "Bus needs a database; call SetupInfra first") db, err := gorm.Open(postgres.Open(i.PGURL), &gorm.Config{Logger: gormlogger.Default.LogMode(gormlogger.Silent)}) Expect(err).ToNot(HaveOccurred())