diff --git a/desktop/docs/services-parity.md b/desktop/docs/services-parity.md index 04aedfe87..938ac3dae 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 | @@ -446,7 +446,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 0f5bcd830..0bae0f5a5 100644 --- a/desktop/src/electron/service-bridge/empty-handlers.ts +++ b/desktop/src/electron/service-bridge/empty-handlers.ts @@ -34,7 +34,6 @@ import type { ProxyEngine } from './modular-state' import type { WorkerStatus } from '@/shared/types/workers' 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 @@ -990,20 +989,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 d79ccea9a..000000000 --- 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 0786ad174..742802576 100644 --- a/desktop/src/electron/service-bridge/modular-state.ts +++ b/desktop/src/electron/service-bridge/modular-state.ts @@ -1654,31 +1654,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 e7501fd87..2ad4bc057 100644 --- a/desktop/src/electron/service-bridge/modular-supervisor.ts +++ b/desktop/src/electron/service-bridge/modular-supervisor.ts @@ -39,7 +39,6 @@ import { isModularLogLevel, type ModularLogLevel } from '@/shared/constants/modular-runtime' -import { listManualNodeEntries } from './manual-nodes-store' import { MODULAR_RUNTIME_BINARIES, modularBinaryFileName @@ -1108,29 +1107,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 { @@ -1154,7 +1130,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/epoch-resync.test.ts b/desktop/tests/modular/epoch-resync.test.ts index 16ba05636..8a15c0bf5 100644 --- a/desktop/tests/modular/epoch-resync.test.ts +++ b/desktop/tests/modular/epoch-resync.test.ts @@ -29,10 +29,6 @@ vi.mock('@/electron/config/ui-config', () => ({ isFirstRun: () => false })) vi.mock('@/electron/service-bridge/broadcaster', () => ({ emitBridgePush: vi.fn() })) -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/desktop/tests/modular/modular-supervisor-readiness.test.ts b/desktop/tests/modular/modular-supervisor-readiness.test.ts index a476be9c5..62d921a03 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/desktop/tests/modular/partial-startup-cleanup.test.ts b/desktop/tests/modular/partial-startup-cleanup.test.ts index 316733184..1988727f2 100644 --- a/desktop/tests/modular/partial-startup-cleanup.test.ts +++ b/desktop/tests/modular/partial-startup-cleanup.test.ts @@ -49,7 +49,6 @@ vi.mock('@/electron/config/ui-config', () => ({ getModularLogLevel: () => 'info' })) vi.mock('@/electron/service-bridge/broadcaster', () => ({ emitBridgePush: vi.fn() })) -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/desktop/tests/modular/pending-invite-sweep.test.ts b/desktop/tests/modular/pending-invite-sweep.test.ts index a66838540..00bb8ec80 100644 --- a/desktop/tests/modular/pending-invite-sweep.test.ts +++ b/desktop/tests/modular/pending-invite-sweep.test.ts @@ -38,10 +38,6 @@ vi.mock('@/electron/config/ui-config', () => ({ isFirstRun: () => false })) vi.mock('@/electron/service-bridge/broadcaster', () => ({ emitBridgePush: vi.fn() })) -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/desktop/tests/modular/silence-reconnect-connector.test.ts b/desktop/tests/modular/silence-reconnect-connector.test.ts index 590592b5a..b14cf94d6 100644 --- a/desktop/tests/modular/silence-reconnect-connector.test.ts +++ b/desktop/tests/modular/silence-reconnect-connector.test.ts @@ -27,7 +27,6 @@ vi.mock('@/electron/config/ui-config', () => ({ getModularLogLevel: () => 'info' })) vi.mock('@/electron/service-bridge/broadcaster', () => ({ emitBridgePush: vi.fn() })) -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/desktop/tests/modular/silence-reconnect.test.ts b/desktop/tests/modular/silence-reconnect.test.ts index 5462c67e3..182d88737 100644 --- a/desktop/tests/modular/silence-reconnect.test.ts +++ b/desktop/tests/modular/silence-reconnect.test.ts @@ -19,7 +19,6 @@ vi.mock('@/shared/utils/log', () => ({ })) vi.mock('@/electron/config/ui-config', () => ({ isFirstRun: () => false })) vi.mock('@/electron/service-bridge/broadcaster', () => ({ emitBridgePush: vi.fn() })) -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/desktop/tests/modular/workers.test.ts b/desktop/tests/modular/workers.test.ts index d12063164..1be2e2563 100644 --- a/desktop/tests/modular/workers.test.ts +++ b/desktop/tests/modular/workers.test.ts @@ -26,7 +26,6 @@ vi.mock('@/shared/utils/log', () => ({ })) vi.mock('@/electron/config/ui-config', () => ({ isFirstRun: () => false })) vi.mock('@/electron/service-bridge/broadcaster', () => ({ emitBridgePush: vi.fn() })) -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 f8f6de770..661d3f38f 100644 --- a/services/nvpair-ui-broker/README.md +++ b/services/nvpair-ui-broker/README.md @@ -804,7 +804,7 @@ report an error even when deletion succeeded. #### `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. Separately from 10-second topology probes, `nvpair-manual-nodes` runs a 2 s ± 250 ms telemetry loop emitting internal `node/telemetry` frames (`id`, `hostUuid`, `gpuUtilizationPct`, `telemetryValid`, `msSince`, `sampleSeq`) for reachable nodes; the broker ingests these observations into its scheduling telemetry cache without modifying discovery state or proxy bindings. 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. Separately from 10-second topology probes, `nvpair-manual-nodes` runs a 2 s ± 250 ms telemetry loop emitting internal `node/telemetry` frames (`id`, `hostUuid`, `gpuUtilizationPct`, `telemetryValid`, `msSince`, `sampleSeq`) for reachable nodes; the broker ingests these observations into its scheduling telemetry cache without modifying discovery state or proxy bindings. 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` / `omlx-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`), one whose `lmstudio_*` status is up into `lmstudio-proxy` (from `lmstudio_port`), and one whose `omlx_*` status is up into `omlx-proxy` (from `omlx_port` or default `:1236`) — a node running multiple engines is bridged into each. 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. @@ -886,7 +886,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 2ee1c2748..5ed756dc0 100644 --- a/services/nvpair-ui-broker/broker.go +++ b/services/nvpair-ui-broker/broker.go @@ -180,17 +180,19 @@ type Broker struct { // heartbeatInterval is the period runHeartbeat ticks at. Injected so tests // can drive the stream without waiting out jsonrpc.HeartbeatInterval; // NewBroker defaults it, and heartbeatEvery() falls back again if unset. - heartbeatInterval time.Duration - nodeID string - scannerPath string - nodeInfoPath string - proxyPath string - lmstudioProxyPath string - omlxProxyPath string - workloadMgrPath string - errorsPath string - engineMgrPath string - manualNodesPath string + heartbeatInterval time.Duration + nodeID string + scannerPath string + nodeInfoPath string + proxyPath string + lmstudioProxyPath string + omlxProxyPath string + workloadMgrPath string + errorsPath string + engineMgrPath string + manualNodesPath string + // manualNodesConfigPath is the durable manual-node list the broker owns. + manualNodesConfigPath string settingsPath string clusterMgrPath string schedulerPath string @@ -401,6 +403,14 @@ type Broker struct { manualMu sync.Mutex manualNodeKeys map[string]string manualNodeStatuses map[string]manualNodeStatusEntry + manualPersistMu sync.Mutex + manualMutationMu sync.Mutex + // manualMutationTail is closed when the most recently queued manual-node + // mutation finishes. Each node/add or node/remove waits for the one before + // it, so mutations apply in the order clients sent them without blocking + // the request loop. Guarded by manualMutationQueueMu. + manualMutationQueueMu sync.Mutex + manualMutationTail chan struct{} // schedMu guards each engine's cached priority and generation. Per-engine // delivery locks serialize asynchronous node/set-priority calls; a stale @@ -479,6 +489,7 @@ func NewBroker(codec *Codec, paths workerPaths) *Broker { errorsPath: paths.errors, engineMgrPath: paths.engineMgr, manualNodesPath: paths.manualNodes, + manualNodesConfigPath: defaultManualNodesConfigPath(), settingsPath: paths.settings, clusterMgrPath: paths.clusterMgr, schedulerPath: paths.scheduler, @@ -977,6 +988,7 @@ func (b *Broker) spawnManualNodes() (supervisedHandle, error) { // default is background, so re-assert demand now rather than at the next // transition. b.pushSchedulingDemandOnStart() + go b.replayPersistedManualNodes(w) slog.Info("manual-nodes started", "path", b.manualNodesPath, "pid", w.cmd.Process.Pid) return w, nil } @@ -1620,10 +1632,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() @@ -1644,8 +1654,8 @@ func (b *Broker) clearManualNodesState() { } 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) } } @@ -3574,16 +3584,67 @@ func (b *Broker) relayToSettings(msg *Message) { // notification is merged into the discovery store by // forwardManualNodesNotification, so the snapshot stays in sync. func (b *Broker) relayToManualNodes(msg *Message) { + if msg.Method != "node/add" && msg.Method != "node/remove" { + b.relayManualNodesRequest(msg, nil) + return + } + // A mutation waits for the one before it and for any replay into a new + // worker, and that wait can last as long as a hung worker takes to be + // restarted. It must not hold up the request loop meanwhile, so it runs in + // its own goroutine, chained behind the previous mutation. + b.manualMutationQueueMu.Lock() + prev := b.manualMutationTail + done := make(chan struct{}) + b.manualMutationTail = done + b.manualMutationQueueMu.Unlock() + go func() { + defer close(done) + if prev != nil { + <-prev + } + b.manualMutationMu.Lock() + defer b.manualMutationMu.Unlock() + finished := make(chan struct{}) + b.relayManualNodesRequest(msg, finished) + <-finished + }() +} + +// relayManualNodesRequest relays one request to the current manual-nodes +// worker. finished, when set, is closed once the request is settled. +func (b *Broker) relayManualNodesRequest(msg *Message, finished chan struct{}) { + settle := func() { + if finished != nil { + close(finished) + } + } mn := b.getManualNodes() if mn == nil { if err := b.codec.RespondError(msg.ID, -32000, "manual-nodes not available"); err != nil { log.Printf("failed to respond to %s: %v", msg.Method, err) } + settle() return } 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...) + 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) { + defer settle() switch { case err != nil: if e := b.codec.RespondError(id, -32000, fmt.Sprintf("manual-nodes call failed: %v", err)); e != nil { @@ -3594,12 +3655,31 @@ 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 { + defer settle() 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 d475164bd..f5ba498b8 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/enginenative" "nvpair-shared/noderec" @@ -73,6 +78,157 @@ type manualNodeTelemetryParams struct { SampleSeq uint64 `json:"sampleSeq,omitempty"` } +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 000000000..6903a87fd --- /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/nvpair-ui-broker/manualnodes_queue_test.go b/services/nvpair-ui-broker/manualnodes_queue_test.go new file mode 100644 index 000000000..1817f10ff --- /dev/null +++ b/services/nvpair-ui-broker/manualnodes_queue_test.go @@ -0,0 +1,120 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +package main + +import ( + "encoding/json" + "net" + "os" + "path/filepath" + "testing" + "time" +) + +// A manual-nodes worker that never answers must not stall the broker's request +// loop: a second mutation queues behind the first instead of blocking the +// caller, and mutations still apply in order once the worker answers. +func TestManualNodeMutationsDoNotBlockTheRequestLoop(t *testing.T) { + path := filepath.Join(t.TempDir(), "configs", "manual-nodes.json") + clientBroker, clientSide := net.Pipe() + workerBroker, workerSide := net.Pipe() + t.Cleanup(func() { + clientBroker.Close() + clientSide.Close() + workerBroker.Close() + workerSide.Close() + }) + worker := &rpcWorker{peer: NewPeer(NewCodec(workerBroker))} + go worker.peer.Serve(nil, nil) + + // The worker reads requests but answers only when released. + release := make(chan struct{}) + received := make(chan string, 4) + go func() { + codec := NewCodec(workerSide) + for { + msg, err := codec.Read() + if err != nil { + return + } + received <- msg.Method + <-release + _ = codec.Respond(msg.ID, map[string]any{"id": "manual:node.local"}) + } + }() + responses := make(chan *Message, 4) + go func() { + codec := NewCodec(clientSide) + for { + msg, err := codec.Read() + if err != nil { + return + } + responses <- msg + } + }() + + b := &Broker{codec: NewCodec(clientBroker), manualNodesConfigPath: path} + b.setManualNodes(worker) + send := func(id, method, params string) { + raw := json.RawMessage(id) + b.relayToManualNodes(&Message{JSONRPC: "2.0", ID: &raw, Method: method, Params: json.RawMessage(params)}) + } + + returned := make(chan struct{}) + go func() { + send("1", "node/add", `{"address":"node.local"}`) + send("2", "node/remove", `{"id":"manual:node.local"}`) + close(returned) + }() + select { + case <-returned: + case <-time.After(2 * time.Second): + t.Fatal("a second mutation blocked the caller while the first was unanswered") + } + if got := <-received; got != "node/add" { + t.Fatalf("first request = %s, want node/add", got) + } + select { + case got := <-received: + t.Fatalf("%s reached the worker before node/add was answered", got) + case <-time.After(100 * time.Millisecond): + } + + close(release) + if got := <-received; got != "node/remove" { + t.Fatalf("second request = %s, want node/remove", got) + } + for _, want := range []string{"1", "2"} { + select { + case r := <-responses: + if string(*r.ID) != want || r.Error != nil { + t.Fatalf("response = %s %+v, want id %s success", *r.ID, r.Error, want) + } + case <-time.After(2 * time.Second): + t.Fatalf("no response for request %s", want) + } + } + entries, err := loadPersistedManualNodes(path) + if err != nil || len(entries) != 0 { + t.Fatalf("persisted = %+v, %v; want the add then the remove applied", entries, err) + } +} + +// The file the desktop app used to own carries {id, address, name}; the broker +// reads it as is, so existing manual nodes survive the move. +func TestManualNodesFileFromTheDesktopStoreLoads(t *testing.T) { + path := filepath.Join(t.TempDir(), "manual-nodes.json") + legacy := `[{"id":"gpu-box","address":"100.101.102.103","name":"gpu-box"}]` + if err := os.WriteFile(path, []byte(legacy), 0o600); err != nil { + t.Fatal(err) + } + entries, err := loadPersistedManualNodes(path) + if err != nil || len(entries) != 1 { + t.Fatalf("entries = %+v, %v", entries, err) + } + if e := entries[0]; e.ID != "gpu-box" || e.Address != "100.101.102.103" || e.Name != "gpu-box" || e.key() != "gpu-box" { + t.Fatalf("entry = %+v", e) + } +} diff --git a/services/versions.json b/services/versions.json index c87972651..ef9094637 100644 --- a/services/versions.json +++ b/services/versions.json @@ -11,7 +11,7 @@ "nvpair-workload-manager": "1.3.1", "nvpair-errors": "1.1.1", "nvpair-node-settings": "3.1.0", - "nvpair-ui-broker": "2.4.1", + "nvpair-ui-broker": "2.5.0", "nvpair-engine-manager": "1.4.0", "nvpair-cluster-manager": "2.1.1", "nvpair-job-scheduler": "1.1.0",