From ec6f655e8bbc797915b33225d55bfa1c81c341d0 Mon Sep 17 00:00:00 2001 From: Sylvester Kaczmarek <16242628+sylvesterkaczmarek@users.noreply.github.com> Date: Thu, 10 Sep 2026 12:30:51 +0100 Subject: [PATCH] Persist manual nodes in the broker Signed-off-by: Sylvester Kaczmarek <16242628+sylvesterkaczmarek@users.noreply.github.com> --- desktop/docs/services-parity.md | 3 +- .../electron/service-bridge/empty-handlers.ts | 19 +- .../service-bridge/manual-nodes-store.ts | 96 --------- .../electron/service-bridge/modular-state.ts | 25 --- .../service-bridge/modular-supervisor.ts | 25 --- .../modular-supervisor-readiness.test.ts | 4 - services/nvpair-ui-broker/README.md | 4 +- services/nvpair-ui-broker/broker.go | 140 ++++++++----- services/nvpair-ui-broker/manualnodes.go | 156 +++++++++++++++ .../manualnodes_persistence_test.go | 189 ++++++++++++++++++ services/versions.json | 2 +- 11 files changed, 446 insertions(+), 217 deletions(-) delete mode 100644 desktop/src/electron/service-bridge/manual-nodes-store.ts create mode 100644 services/nvpair-ui-broker/manualnodes_persistence_test.go diff --git a/desktop/docs/services-parity.md b/desktop/docs/services-parity.md index 93d13290..33efb583 100644 --- a/desktop/docs/services-parity.md +++ b/desktop/docs/services-parity.md @@ -21,7 +21,7 @@ history. | Process supervision | Complete | Electron starts only `nvpair-ui-broker`; the broker supervises all workers | | Discovery | Complete | Broker discovery snapshots drive available nodes and node state | | Node telemetry | Integrated with direct poll | Electron polls advertised `/v1/node-info` (plain HTTP); remote OS and some remote telemetry are backend-limited | -| Manual nodes | Complete with local persistence | Broker owns probing and proxy registration; Electron persists entries for replay | +| Manual nodes | Complete with local persistence | Broker owns probing, durable entries, restart replay, and proxy registration | | Ollama routing | Complete | Broker relay and backend scheduler drive proxy routing | | LM Studio routing | Complete | Parallel broker relay and scheduler path | | Local engine lifecycle | Complete | Install, start, stop, uninstall, update, and port configuration | @@ -419,7 +419,6 @@ provide an equivalent client-facing contract: | Responsibility | Location | | ----------------------------------------------------------- | ------------------------------------------------ | | Poll node telemetry over `/v1/node-info` | `node-info-poller.ts` | -| Persist and replay manual node entries | `manual-nodes-store.ts`, `modular-supervisor.ts` | | Bridge the local node into engine proxies | `modular-supervisor.ts` | | Present optimistic engine transition state | `pending-actions.store.ts`, bridge state | | Serve the model hub (Ollama committed list, LM Studio live) | `src/electron/model-hub/` | diff --git a/desktop/src/electron/service-bridge/empty-handlers.ts b/desktop/src/electron/service-bridge/empty-handlers.ts index dc5c2dd8..cdd36b6b 100644 --- a/desktop/src/electron/service-bridge/empty-handlers.ts +++ b/desktop/src/electron/service-bridge/empty-handlers.ts @@ -23,7 +23,6 @@ import { import type { ProxyEngine } from './modular-state' import type { JsonObject, JsonValue } from './json-rpc-subprocess' import { emptyInvite, parseClusterNodes, parseInvite, parseNodeIdentity } from './cluster-json' -import { removeManualNodeEntry, resolveManualNodeKey } from './manual-nodes-store' type BridgeHandler = ( payload?: WsInvokeRequest @@ -908,20 +907,10 @@ async function handleNodeRemoveMember( const selfId = state.getSelfId() const isSelfLeave = selfId !== null && payload.nodeId === selfId - // `payload.nodeId` is the node's UUID, but the manual-nodes store and the - // broker's `node/remove` relay key a manual entry by the name it was added - // with — never the UUID. Map the UUID back through the node's reachable - // addresses to the persisted entry so it is actually pruned (and does not - // reappear on the next replay). If it is not a manual node (no address match) - // fall back to the display hostname, then the raw id; both are harmless - // no-ops for a purely-discovered peer. The broker supervises - // nvpair-manual-nodes and relays `node/*` verbatim. - const manualKey = - resolveManualNodeKey(state.getNodeAddresses(payload.nodeId)) || - state.getNodeHostname(payload.nodeId) || - payload.nodeId - removeManualNodeEntry(manualKey) - supervisor.callProcess('broker', 'node/remove', { id: manualKey }).catch(() => {}) + // The broker owns manual-node persistence and resolves a stable node UUID + // back to the manual alias expected by nvpair-manual-nodes. For a purely + // discovered peer this remains a harmless no-op. + supervisor.callProcess('broker', 'node/remove', { id: payload.nodeId }).catch(() => {}) if (isSelfLeave) { // Self-departure is `cluster:leave`, not `nodes:remove` (the cluster-manager diff --git a/desktop/src/electron/service-bridge/manual-nodes-store.ts b/desktop/src/electron/service-bridge/manual-nodes-store.ts deleted file mode 100644 index d79ccea9..00000000 --- a/desktop/src/electron/service-bridge/manual-nodes-store.ts +++ /dev/null @@ -1,96 +0,0 @@ -// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. -// SPDX-License-Identifier: Apache-2.0 - -import fs from 'fs' -import path from 'path' -import { getPaths } from '@/electron/globals' -import type { JsonObject, JsonValue } from './json-rpc-subprocess' - -interface ManualNodeEntry { - id: string - address: string - name: string -} - -function configFilePath(): string { - return path.join(getPaths().getUserData(), 'configs', 'manual-nodes.json') -} - -function objectValue(value: JsonValue | undefined): JsonObject | null { - if (!value || typeof value !== 'object' || Array.isArray(value)) return null - return value -} - -function stringValue(value: JsonValue | undefined): string { - return typeof value === 'string' ? value : '' -} - -function entryValue(value: JsonValue | undefined): ManualNodeEntry | null { - const obj = objectValue(value) - if (!obj) return null - - const address = stringValue(obj.address) - if (!address) return null - - const name = stringValue(obj.name) || address - const id = stringValue(obj.id) || name - return { id, address, name } -} - -export function listManualNodeEntries(): ManualNodeEntry[] { - try { - const filePath = configFilePath() - if (!fs.existsSync(filePath)) return [] - - const raw = fs.readFileSync(filePath, 'utf8') - const parsed: JsonValue = JSON.parse(raw) - if (!Array.isArray(parsed)) return [] - - const entries: ManualNodeEntry[] = [] - for (const item of parsed) { - const entry = entryValue(item) - if (entry) entries.push(entry) - } - return entries - } catch { - return [] - } -} - -export function removeManualNodeEntry(nodeId: string): void { - const entries = listManualNodeEntries().filter( - entry => entry.id !== nodeId && entry.address !== nodeId && entry.name !== nodeId - ) - saveManualNodeEntries(entries) -} - -/** - * Resolve a persisted manual entry from a set of a node's reachable addresses to - * the key `nvpair-manual-nodes` uses for it. `replayManualNodes` re-adds entries - * via `node/add` with `{ address, name }`, and the backend keys the resulting - * manual node by that `name` (`nodeID(entry)`), so returning `entry.name` gives a - * key that both {@link removeManualNodeEntry} and the broker `node/remove` relay - * match. A node's stable UUID never matches (it is not what the entry is keyed - * by), so callers must map it back through the node's addresses. Returns null - * when no persisted entry owns any of `addresses` (i.e. not a manual node). - */ -export function resolveManualNodeKey(addresses: readonly string[]): string | null { - if (addresses.length === 0) return null - const known = new Set(addresses) - for (const entry of listManualNodeEntries()) { - if (known.has(entry.address)) return entry.name - } - return null -} - -function saveManualNodeEntries(entries: ManualNodeEntry[]): void { - try { - const filePath = configFilePath() - fs.mkdirSync(path.dirname(filePath), { recursive: true }) - const tmp = `${filePath}.tmp` - fs.writeFileSync(tmp, JSON.stringify(entries, null, 2), 'utf8') - fs.renameSync(tmp, filePath) - } catch { - /* best-effort */ - } -} diff --git a/desktop/src/electron/service-bridge/modular-state.ts b/desktop/src/electron/service-bridge/modular-state.ts index ab6c5fb5..0cbf2e95 100644 --- a/desktop/src/electron/service-bridge/modular-state.ts +++ b/desktop/src/electron/service-bridge/modular-state.ts @@ -1016,31 +1016,6 @@ class ModularBridgeState { return this.selfId } - /** - * The display hostname for a node key (UUID), or '' when unknown. Used as a - * last-resort manual-removal key when address matching finds no persisted - * entry (see {@link resolveManualNodeKey}). - */ - getNodeHostname(nodeId: string): string { - return this.nodes.get(nodeId)?.name ?? '' - } - - /** - * Every address a node key (UUID) is known to be reachable at — its - * canonical `reachableAddress`, its discovered addresses, and its `host`. - * Used to map a UUID back to the manual-nodes store entry (keyed by the - * user-entered address) when removing a manual node. Empty for an unknown id. - */ - getNodeAddresses(nodeId: string): string[] { - const node = this.nodes.get(nodeId) - if (!node) return [] - const out = new Set() - if (node.reachableAddress) out.add(node.reachableAddress) - for (const address of nodeAddresses(node)) out.add(address) - if (node.host) out.add(node.host) - return Array.from(out) - } - /** The authoritative live inbound invites awaiting a PIN, oldest first. */ getPendingInvites(): Invite[] { return Array.from(this.pendingInvites.values()).sort((a, b) => a.createdAt - b.createdAt) diff --git a/desktop/src/electron/service-bridge/modular-supervisor.ts b/desktop/src/electron/service-bridge/modular-supervisor.ts index 943d0f84..50912f84 100644 --- a/desktop/src/electron/service-bridge/modular-supervisor.ts +++ b/desktop/src/electron/service-bridge/modular-supervisor.ts @@ -38,7 +38,6 @@ import { isModularLogLevel, type ModularLogLevel } from '@/shared/constants/modular-runtime' -import { listManualNodeEntries } from './manual-nodes-store' import { MODULAR_RUNTIME_BINARIES, modularBinaryFileName @@ -841,29 +840,6 @@ class ModularSupervisor { return [...args, ...this.logLevelArgs()] } - /** - * Re-add persisted manual nodes through the broker's `node/add` relay. The - * broker's `nvpair-manual-nodes` loses its in-memory entries on restart (N86), - * so PAIR UI owns the durable list (`manual-nodes.json`) and replays it once - * the broker is ready. - */ - private async replayManualNodes(): Promise { - const entries = listManualNodeEntries() - for (const entry of entries) { - try { - await this.callProcess('broker', 'node/add', { - address: entry.address, - name: entry.name - }) - } catch (err) { - log.warn({ - sublevel: 'manual-nodes', - message: `Failed to replay manual node ${entry.address}: ${getErrorString(err)}` - }) - } - } - } - private async onBrokerReady(): Promise { const subscribe = async (method: string, label: string): Promise => { try { @@ -885,7 +861,6 @@ class ModularSupervisor { await this.syncClusterIdentityToManager() await this.resolveSelfId() - await this.replayManualNodes() await this.hydrateEngineManager() this.startInstalledEnginesOnFirstOpen() await this.seedClusterPeerIds() diff --git a/desktop/tests/modular/modular-supervisor-readiness.test.ts b/desktop/tests/modular/modular-supervisor-readiness.test.ts index a476be9c..62d921a0 100644 --- a/desktop/tests/modular/modular-supervisor-readiness.test.ts +++ b/desktop/tests/modular/modular-supervisor-readiness.test.ts @@ -40,10 +40,6 @@ vi.mock('@/electron/service-bridge/broadcaster', () => ({ emitBridgePush: mocks.emitBridgePush })) -vi.mock('@/electron/service-bridge/manual-nodes-store', () => ({ - listManualNodeEntries: () => [] -})) - vi.mock('@/electron/service-bridge/node-info-poller', () => ({ startNodeInfoPoller: vi.fn(), stopNodeInfoPoller: vi.fn() diff --git a/services/nvpair-ui-broker/README.md b/services/nvpair-ui-broker/README.md index d2b0e174..d035a496 100644 --- a/services/nvpair-ui-broker/README.md +++ b/services/nvpair-ui-broker/README.md @@ -475,7 +475,7 @@ Any `settings/*` request is forwarded to `nvpair-node-settings` and its response #### `node/add` / `node/remove` / `nodes/list` (manual nodes) -Relayed to `nvpair-manual-nodes`. `node/add` (`{ address, name?, tls_port?, mtls? }`) registers a user-added node and probes it; `node/remove` (`{ id }`) drops it; `nodes/list` returns the tracked manual nodes. Manually added nodes also surface in the shared `discovery:get-nodes` / `discovery:nodes-changed` snapshot — the broker merges `nvpair-manual-nodes`' `node/discovered|updated|removed` into the same store the scanner feeds. A `nvpair-manual-nodes` restart loses the in-memory entries because neither that worker nor the broker persists an authoritative copy, so clients must re-add manual nodes after a restart. Error `-32000 "manual-nodes not available"` when no manual-nodes worker is supervised. +Relayed to `nvpair-manual-nodes`. `node/add` (`{ address, name?, tls_port?, mtls? }`) registers a user-added node and probes it; `node/remove` (`{ id }`) drops it; `nodes/list` returns the tracked manual nodes. Manually added nodes also surface in the shared `discovery:get-nodes` / `discovery:nodes-changed` snapshot — the broker merges `nvpair-manual-nodes`' `node/discovered|updated|removed` into the same store the scanner feeds. The broker persists successful `node/add` operations to `configs/manual-nodes.json` in the per-user data directory and replays them into a replacement `nvpair-manual-nodes` worker after startup. Successful `node/remove` operations update the same durable list. Error `-32000 "manual-nodes not available"` when no manual-nodes worker is supervised. **Manual → proxy bridge.** When the broker supervises both `nvpair-manual-nodes` and a proxy, it also bridges a manual node whose engine is reachable into that proxy via `node/add-manual` (host/port from the node's per-engine status), so inference can route to it through `proxy:node/select` / `lmstudio-proxy:node/select` just like a relay-discovered node. This is per-engine: a node whose `ollama_*` status is up is bridged into `ollama-proxy` (host/port from `ollama_port`), and one whose `lmstudio_*` status is up into `lmstudio-proxy` (from `lmstudio_port`) — a node running both is bridged into both. Manual nodes are by definition the ones that never appear via the daemon's `_nvpair-node` discovery, so this explicit add is what makes them routable. The bridge tracks reachability: an engine that goes down (or a node that is removed, or whose prober crashes) is pulled back out with `node/remove-manual`. A proxy that isn't supervised → that leg is a no-op; manual nodes still appear in the discovery snapshot as before. @@ -555,7 +555,7 @@ Attach to a pre-existing endpoint: - **Engine-advertise control surface.** Engine registration is auto-driven only: the broker tracks local ollama / LM Studio on their fixed coordinates and registers `ol` / `lm` with the daemon while up. There's no manual-advertise RPC (custom service, port, name, or TXT), and no way to advertise anything other than the detected engines. - **node-info control surface.** node-info is spawned and torn down with the broker, and the broker pushes it only two things over stdin: the log level, and this node's cluster principal (`nodeinfo:set-cluster-identity`, sent on spawn and on every membership or pin-set change, because node-info holds no cluster dir and so cannot read membership itself). Otherwise it's hands-off: the broker registers its port with the daemon (which enriches over plain HTTP) but doesn't pass through TLS material (`--cert` / `--key` / `--client-ca`) or a custom `--port`, and exposes no RPC to query or reconfigure it. It runs with its own defaults plus those two pushes. -- **Manual-node persistence across restarts.** `nvpair-manual-nodes` keeps its entries only in memory and the broker holds no authoritative copy, so a manual-nodes crash-and-restart loses the user's manual nodes (the broker evicts the orphaned entries from the snapshot; clients must re-add them). +- **Manual-node persistence across restarts.** `nvpair-manual-nodes` remains an in-memory worker; the broker owns `configs/manual-nodes.json`, clears stale live projections when the worker exits, and replays the durable entries into its replacement. - **Per-event push semantics.** `discovery:nodes-changed` always carries the full current snapshot, not a delta. For small N this is fine and lets the client treat the payload as authoritative without state reconciliation. `errors:update` is likewise a full snapshot. - **BYO-TLS node-info discovery.** The daemon enriches peers over plain HTTP per the consolidated transport policy (node-info is plain on the broker path). Reading a node-info served over operator-configured BYO-TLS / mTLS is not supported through the broker; discovery works for the plain-HTTP case (the common one). - **Per-worker log-level granularity.** `log/set-level` fans the broker's single level out to every running child, but there's no way to set a different level per worker — it's one level for the whole process tree. diff --git a/services/nvpair-ui-broker/broker.go b/services/nvpair-ui-broker/broker.go index 0d189578..c0ebf532 100644 --- a/services/nvpair-ui-broker/broker.go +++ b/services/nvpair-ui-broker/broker.go @@ -145,22 +145,23 @@ type ProxyStatusResult struct { // client connection in listen mode so future per-session caches (auth // tokens, watched-resource cursors, etc.) don't bleed across clients. type Broker struct { - codec *Codec - cancel context.CancelFunc - startedAt time.Time - nodeID string - scannerPath string - nodeInfoPath string - proxyPath string - lmstudioProxyPath string - workloadMgrPath string - errorsPath string - engineMgrPath string - manualNodesPath string - settingsPath string - clusterMgrPath string - schedulerPath string - clusterDir string + codec *Codec + cancel context.CancelFunc + startedAt time.Time + nodeID string + scannerPath string + nodeInfoPath string + proxyPath string + lmstudioProxyPath string + workloadMgrPath string + errorsPath string + engineMgrPath string + manualNodesPath string + manualNodesConfigPath string + settingsPath string + clusterMgrPath string + schedulerPath string + clusterDir string // Managed-port state is prepared before proxy startup and read by the proxy // supervisor/reader goroutines. Ollama commits its pending backend move after // its proxy reserves :11434; LM Studio moves through engine-manager first, @@ -305,6 +306,8 @@ type Broker struct { manualMu sync.Mutex manualNodeKeys map[string]string manualNodeStatuses map[string]manualNodeStatusEntry + manualPersistMu sync.Mutex + manualMutationMu sync.Mutex // schedMu guards each engine's cached priority and generation. Per-engine // delivery locks serialize asynchronous node/set-priority calls; a stale @@ -362,30 +365,31 @@ func NewBroker(codec *Codec, paths workerPaths) *Broker { // its localNodeID stays in lockstep with what the broker stamps. nodeID := resolveLocalNodeID(paths.clusterDir) return &Broker{ - codec: codec, - startedAt: time.Now(), - nodeID: nodeID, - scannerPath: paths.scanner, - nodeInfoPath: paths.nodeInfo, - proxyPath: paths.proxy, - lmstudioProxyPath: paths.lmstudioProxy, - workloadMgrPath: paths.workloadMgr, - errorsPath: paths.errors, - engineMgrPath: paths.engineMgr, - manualNodesPath: paths.manualNodes, - settingsPath: paths.settings, - clusterMgrPath: paths.clusterMgr, - schedulerPath: paths.scheduler, - clusterDir: paths.clusterDir, - store: newDiscoveryStore(), - telemetry: newTelemetryCache(), - relayDir: relay.NewDirectory(), - regCache: relay.NewRegistrationCache(), - manualNodeKeys: make(map[string]string), - manualNodeStatuses: make(map[string]manualNodeStatusEntry), - workloads: workloadstore.New(), - ollamaPortReady: make(chan struct{}), - lmstudioPortReady: make(chan struct{}), + codec: codec, + startedAt: time.Now(), + nodeID: nodeID, + scannerPath: paths.scanner, + nodeInfoPath: paths.nodeInfo, + proxyPath: paths.proxy, + lmstudioProxyPath: paths.lmstudioProxy, + workloadMgrPath: paths.workloadMgr, + errorsPath: paths.errors, + engineMgrPath: paths.engineMgr, + manualNodesPath: paths.manualNodes, + manualNodesConfigPath: defaultManualNodesConfigPath(), + settingsPath: paths.settings, + clusterMgrPath: paths.clusterMgr, + schedulerPath: paths.scheduler, + clusterDir: paths.clusterDir, + store: newDiscoveryStore(), + telemetry: newTelemetryCache(), + relayDir: relay.NewDirectory(), + regCache: relay.NewRegistrationCache(), + manualNodeKeys: make(map[string]string), + manualNodeStatuses: make(map[string]manualNodeStatusEntry), + workloads: workloadstore.New(), + ollamaPortReady: make(chan struct{}), + lmstudioPortReady: make(chan struct{}), } } @@ -828,6 +832,7 @@ func (b *Broker) spawnManualNodes() (supervisedHandle, error) { return nil, err } b.setManualNodes(w) + go b.replayPersistedManualNodes(w) slog.Info("manual-nodes started", "path", b.manualNodesPath, "pid", w.cmd.Process.Pid) return w, nil } @@ -1346,10 +1351,8 @@ func (b *Broker) survivingAliasLocked(key string) (manualNodeStatusEntry, bool) // clearManualNodesState is the manual-nodes supervisor's clearHandle: on a // crash it drops the worker handle and evicts every manual-origin node from -// the discovery store. The restarted process comes up with no entries (it -// keeps no persistent state and the broker doesn't re-feed them), so -// leaving the old manual nodes in the snapshot would strand stale entries -// that never age out. Clients re-add manual nodes after a restart. +// the discovery store. The durable broker-owned list is left intact and is +// replayed into the replacement worker after it starts. func (b *Broker) clearManualNodesState() { b.setManualNodes(nil) b.manualMu.Lock() @@ -1368,8 +1371,8 @@ func (b *Broker) clearManualNodesState() { b.store.Remove(key, sourceManual) b.removeTelemetry(sourceManual, key) // Pull the now-orphaned node out of every proxy too, so inference - // doesn't keep a stale manual target the crashed prober can no - // longer vouch for. Clients re-add manual nodes after the restart. + // doesn't keep a stale manual target while the replacement worker + // restores and re-probes the durable entry. b.removeManualNodeFromProxies(key) } } @@ -3159,7 +3162,29 @@ func (b *Broker) relayToManualNodes(msg *Message) { } id := msg.ID method := msg.Method - relayErr := mn.RelayRequest(method, msg.Params, func(result json.RawMessage, rpcErr *RPCError, err error) { + params := append(json.RawMessage(nil), msg.Params...) + mutates := method == "node/add" || method == "node/remove" + if mutates { + b.manualMutationMu.Lock() + } + var added persistedManualNode + removeID := "" + if method == "node/add" { + _ = json.Unmarshal(params, &added) + } + if method == "node/remove" { + var remove struct { + ID string `json:"id"` + } + if json.Unmarshal(params, &remove) == nil && remove.ID != "" { + removeID = b.manualAliasForKey(remove.ID) + params, _ = json.Marshal(map[string]string{"id": removeID}) + } + } + relayErr := mn.RelayRequest(method, params, func(result json.RawMessage, rpcErr *RPCError, err error) { + if mutates { + defer b.manualMutationMu.Unlock() + } switch { case err != nil: if e := b.codec.RespondError(id, -32000, fmt.Sprintf("manual-nodes call failed: %v", err)); e != nil { @@ -3170,12 +3195,33 @@ func (b *Broker) relayToManualNodes(msg *Message) { log.Printf("failed to relay manual-nodes error for %s: %v", method, e) } default: + if method == "node/add" && added.Address != "" { + var status struct { + ID string `json:"id"` + } + if json.Unmarshal(result, &status) == nil { + added.ID = status.ID + } + if err := b.persistManualNode(added); err != nil { + _ = b.codec.RespondError(id, -32000, fmt.Sprintf("manual-nodes persistence failed: %v", err)) + return + } + } + if method == "node/remove" && removeID != "" { + if err := b.removePersistedManualNode(removeID); err != nil { + _ = b.codec.RespondError(id, -32000, fmt.Sprintf("manual-nodes persistence failed: %v", err)) + return + } + } if e := b.codec.Respond(id, result); e != nil { log.Printf("failed to relay manual-nodes result for %s: %v", method, e) } } }) if relayErr != nil { + if mutates { + b.manualMutationMu.Unlock() + } if err := b.codec.RespondError(msg.ID, -32000, fmt.Sprintf("manual-nodes call failed: %v", relayErr)); err != nil { log.Printf("failed to respond to %s: %v", msg.Method, err) } diff --git a/services/nvpair-ui-broker/manualnodes.go b/services/nvpair-ui-broker/manualnodes.go index 47fe0ff1..ba1e28f1 100644 --- a/services/nvpair-ui-broker/manualnodes.go +++ b/services/nvpair-ui-broker/manualnodes.go @@ -6,9 +6,14 @@ package main import ( "context" "encoding/json" + "errors" "log/slog" + "os" + "path/filepath" + "sort" "time" + "nvpair-shared/appdir" "nvpair-shared/noderec" ) @@ -46,6 +51,157 @@ type manualNodeStatusEntry struct { receivedAt time.Time } +type persistedManualNode struct { + ID string `json:"id,omitempty"` + Address string `json:"address"` + Name string `json:"name,omitempty"` + TLSPort int `json:"tls_port,omitempty"` + MTLS bool `json:"mtls,omitempty"` +} + +func (e persistedManualNode) key() string { + if e.ID != "" { + return e.ID + } + if e.Name != "" { + return e.Name + } + return "manual:" + e.Address +} + +func defaultManualNodesConfigPath() string { + if path, err := appdir.Path("configs", "manual-nodes.json"); err == nil { + return path + } + if exe, err := os.Executable(); err == nil { + return filepath.Join(filepath.Dir(exe), "configs", "manual-nodes.json") + } + return filepath.Join("configs", "manual-nodes.json") +} + +func loadPersistedManualNodes(path string) ([]persistedManualNode, error) { + data, err := os.ReadFile(path) + if errors.Is(err, os.ErrNotExist) { + return nil, nil + } + if err != nil { + return nil, err + } + var entries []persistedManualNode + if err := json.Unmarshal(data, &entries); err != nil { + return nil, err + } + valid := entries[:0] + for _, entry := range entries { + if entry.Address != "" { + valid = append(valid, entry) + } + } + return valid, nil +} + +func writePersistedManualNodes(path string, entries []persistedManualNode) error { + sort.Slice(entries, func(i, j int) bool { return entries[i].key() < entries[j].key() }) + data, err := json.MarshalIndent(entries, "", " ") + if err != nil { + return err + } + data = append(data, '\n') + if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil { + return err + } + tmp := path + ".tmp" + if err := os.WriteFile(tmp, data, 0o600); err != nil { + return err + } + if err := os.Rename(tmp, path); err != nil { + _ = os.Remove(tmp) + return err + } + return nil +} + +func (b *Broker) persistManualNode(entry persistedManualNode) error { + if entry.ID == "" { + entry.ID = entry.key() + } + b.manualPersistMu.Lock() + defer b.manualPersistMu.Unlock() + entries, err := loadPersistedManualNodes(b.manualNodesConfigPath) + if err != nil { + return err + } + replaced := false + for i := range entries { + if entries[i].key() == entry.key() { + entries[i] = entry + replaced = true + break + } + } + if !replaced { + entries = append(entries, entry) + } + return writePersistedManualNodes(b.manualNodesConfigPath, entries) +} + +func (b *Broker) removePersistedManualNode(id string) error { + b.manualPersistMu.Lock() + defer b.manualPersistMu.Unlock() + entries, err := loadPersistedManualNodes(b.manualNodesConfigPath) + if err != nil { + return err + } + out := entries[:0] + for _, entry := range entries { + if entry.key() != id && entry.ID != id && entry.Name != id { + out = append(out, entry) + } + } + return writePersistedManualNodes(b.manualNodesConfigPath, out) +} + +func (b *Broker) replayPersistedManualNodes(worker *rpcWorker) { + b.manualMutationMu.Lock() + defer b.manualMutationMu.Unlock() + b.manualPersistMu.Lock() + entries, err := loadPersistedManualNodes(b.manualNodesConfigPath) + b.manualPersistMu.Unlock() + if err != nil { + slog.Warn("failed to load persisted manual nodes", "err", err) + return + } + for _, entry := range entries { + params, err := json.Marshal(entry) + if err != nil { + continue + } + if _, rpcErr, err := worker.Call(context.Background(), "node/add", params); err != nil { + slog.Warn("failed to replay manual node", "id", entry.key(), "err", err) + } else if rpcErr != nil { + slog.Warn("manual node replay rejected", "id", entry.key(), "code", rpcErr.Code, "msg", rpcErr.Message) + } + } +} + +func (b *Broker) manualAliasForKey(id string) string { + b.manualMu.Lock() + defer b.manualMu.Unlock() + if _, ok := b.manualNodeKeys[id]; ok { + return id + } + best := "" + for alias, key := range b.manualNodeKeys { + if key == id && (best == "" || alias < best) { + best = alias + } + } + if best != "" { + return best + } + return id +} + func manualNodeTelemetry(status manualNodeStatus, hostUUID string) noderec.NodeTelemetry { var utilization uint32 for i := range status.GPUs { diff --git a/services/nvpair-ui-broker/manualnodes_persistence_test.go b/services/nvpair-ui-broker/manualnodes_persistence_test.go new file mode 100644 index 00000000..6903a87f --- /dev/null +++ b/services/nvpair-ui-broker/manualnodes_persistence_test.go @@ -0,0 +1,189 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +package main + +import ( + "encoding/json" + "net" + "path/filepath" + "testing" + "time" +) + +func TestPersistManualNodeRoundTripsAndRemoves(t *testing.T) { + path := filepath.Join(t.TempDir(), "configs", "manual-nodes.json") + b := &Broker{manualNodesConfigPath: path} + + if err := b.persistManualNode(persistedManualNode{ID: "lab", Address: "node.local", Name: "lab", TLSPort: 14319, MTLS: true}); err != nil { + t.Fatal(err) + } + entries, err := loadPersistedManualNodes(path) + if err != nil { + t.Fatal(err) + } + if len(entries) != 1 || entries[0].ID != "lab" || entries[0].TLSPort != 14319 || !entries[0].MTLS { + t.Fatalf("persisted entries = %+v", entries) + } + + if err := b.removePersistedManualNode("lab"); err != nil { + t.Fatal(err) + } + entries, err = loadPersistedManualNodes(path) + if err != nil { + t.Fatal(err) + } + if len(entries) != 0 { + t.Fatalf("entries after remove = %+v, want empty", entries) + } +} + +func TestReplayPersistedManualNodesIntoWorker(t *testing.T) { + path := filepath.Join(t.TempDir(), "configs", "manual-nodes.json") + if err := writePersistedManualNodes(path, []persistedManualNode{{ID: "lab", Address: "node.local", Name: "lab"}}); err != nil { + t.Fatal(err) + } + + brokerSide, workerSide := net.Pipe() + defer brokerSide.Close() + defer workerSide.Close() + worker := &rpcWorker{peer: NewPeer(NewCodec(brokerSide))} + go worker.peer.Serve(nil, nil) + + got := make(chan persistedManualNode, 1) + go func() { + codec := NewCodec(workerSide) + msg, err := codec.Read() + if err != nil { + return + } + var entry persistedManualNode + if json.Unmarshal(msg.Params, &entry) == nil { + got <- entry + } + _ = codec.Respond(msg.ID, map[string]string{"id": "lab"}) + }() + + b := &Broker{manualNodesConfigPath: path} + b.replayPersistedManualNodes(worker) + + select { + case entry := <-got: + if entry.Address != "node.local" || entry.Name != "lab" { + t.Fatalf("replayed entry = %+v", entry) + } + case <-time.After(2 * time.Second): + t.Fatal("persisted manual node was not replayed") + } +} + +func TestManualAliasForOperationalKey(t *testing.T) { + b := &Broker{manualNodeKeys: map[string]string{"second": "node-uuid", "first": "node-uuid"}} + if got := b.manualAliasForKey("node-uuid"); got != "first" { + t.Fatalf("alias = %q, want first", got) + } + if got := b.manualAliasForKey("second"); got != "second" { + t.Fatalf("direct alias = %q, want second", got) + } +} + +func TestManualNodeRelayPersistsSuccessfulAdd(t *testing.T) { + path := filepath.Join(t.TempDir(), "configs", "manual-nodes.json") + clientBroker, clientSide := net.Pipe() + workerBroker, workerSide := net.Pipe() + defer clientBroker.Close() + defer clientSide.Close() + defer workerBroker.Close() + defer workerSide.Close() + + worker := &rpcWorker{peer: NewPeer(NewCodec(workerBroker))} + go worker.peer.Serve(nil, nil) + go func() { + codec := NewCodec(workerSide) + msg, err := codec.Read() + if err != nil { + return + } + _ = codec.Respond(msg.ID, map[string]any{"id": "manual:node.local", "address": "node.local"}) + }() + + b := &Broker{codec: NewCodec(clientBroker), manualNodesConfigPath: path} + b.setManualNodes(worker) + id := json.RawMessage("1") + b.relayToManualNodes(&Message{ + JSONRPC: "2.0", + ID: &id, + Method: "node/add", + Params: json.RawMessage(`{"address":"node.local"}`), + }) + response, err := NewCodec(clientSide).Read() + if err != nil || response.Error != nil { + t.Fatalf("add response error = %v, frame = %+v", err, response) + } + + entries, err := loadPersistedManualNodes(path) + if err != nil { + t.Fatal(err) + } + if len(entries) != 1 || entries[0].ID != "manual:node.local" || entries[0].Address != "node.local" { + t.Fatalf("persisted add = %+v", entries) + } +} + +func TestManualNodeRelayRemovesByOperationalKey(t *testing.T) { + path := filepath.Join(t.TempDir(), "configs", "manual-nodes.json") + if err := writePersistedManualNodes(path, []persistedManualNode{{ID: "lab", Address: "node.local", Name: "lab"}}); err != nil { + t.Fatal(err) + } + clientBroker, clientSide := net.Pipe() + workerBroker, workerSide := net.Pipe() + defer clientBroker.Close() + defer clientSide.Close() + defer workerBroker.Close() + defer workerSide.Close() + + worker := &rpcWorker{peer: NewPeer(NewCodec(workerBroker))} + go worker.peer.Serve(nil, nil) + seenID := make(chan string, 1) + go func() { + codec := NewCodec(workerSide) + msg, err := codec.Read() + if err != nil { + return + } + var params struct { + ID string `json:"id"` + } + _ = json.Unmarshal(msg.Params, ¶ms) + seenID <- params.ID + _ = codec.Respond(msg.ID, map[string]bool{"removed": true}) + }() + + b := &Broker{ + codec: NewCodec(clientBroker), + manualNodesConfigPath: path, + manualNodeKeys: map[string]string{"lab": "node-uuid"}, + } + b.setManualNodes(worker) + id := json.RawMessage("2") + b.relayToManualNodes(&Message{ + JSONRPC: "2.0", + ID: &id, + Method: "node/remove", + Params: json.RawMessage(`{"id":"node-uuid"}`), + }) + response, err := NewCodec(clientSide).Read() + if err != nil || response.Error != nil { + t.Fatalf("remove response error = %v, frame = %+v", err, response) + } + if got := <-seenID; got != "lab" { + t.Fatalf("worker remove id = %q, want lab", got) + } + entries, err := loadPersistedManualNodes(path) + if err != nil { + t.Fatal(err) + } + if len(entries) != 0 { + t.Fatalf("persisted entries after remove = %+v", entries) + } +} diff --git a/services/versions.json b/services/versions.json index 29d8c230..e5039ca3 100644 --- a/services/versions.json +++ b/services/versions.json @@ -11,7 +11,7 @@ "nvpair-workload-manager": "0.13.3", "nvpair-errors": "0.7.4", "nvpair-node-settings": "1.0.4", - "nvpair-ui-broker": "0.40.2", + "nvpair-ui-broker": "0.40.3", "nvpair-engine-manager": "0.17.4", "nvpair-cluster-manager": "1.1.4", "nvpair-job-scheduler": "0.4.1",